test(run-engine): update snapshot-count expectations for the collapsed trigger write

The store-routing counters now also count createRun, since the trigger
path's QUEUED snapshot is written nested in the run create. The
replica-lag test starts the run attempt so it still has a middle
snapshot to exercise rather than degenerating to an empty comparison.

Author: ericallam

.server-changes/collapse-trigger-queued-snapshot.md

@@ -3,4 +3,4 @@ area: webapp
 type: improvement
 ---
 
-Triggering a task now does one less database write, so runs reach the queue marginally faster.
+Triggering a task now reaches the queue marginally faster.

internal-packages/run-engine/src/engine/systems/enqueueSystem.test.ts

@@ -44,10 +44,10 @@ function createEngineOptions(redisOptions: any, prisma: any, store?: PostgresRun
 }
 
 /**
- * A real PostgresRunStore subclass that counts the snapshot create method that enqueueRun's
- * snapshot write routes through (via executionSnapshotSystem.createExecutionSnapshot). super.*
- * runs the genuine store implementation, so the routing is observed over real containers without
- * ever mocking prisma or the store.
+ * A real PostgresRunStore subclass that counts the store methods a run's QUEUED snapshot can be
+ * written through: nested in `createRun` on the trigger path, or standalone via
+ * `createExecutionSnapshot` on every re-enqueue. super.* runs the genuine store implementation, so
+ * the routing is observed over real containers without ever mocking prisma or the store.
  */
 class CountingPostgresRunStore extends PostgresRunStore {
   public snapshotCreates = 0;
@@ -59,12 +59,19 @@ class CountingPostgresRunStore extends PostgresRunStore {
     this.snapshotCreates++;
     return super.createExecutionSnapshot(input, tx);
   }
+
+  override async createRun(
+    params: Parameters<PostgresRunStore["createRun"]>[0],
+    tx?: any
+  ): ReturnType<PostgresRunStore["createRun"]> {
+    this.snapshotCreates++;
+    return super.createRun(params, tx);
+  }
 }
 
 describe("RunEngine enqueueRun store routing", () => {
-  // The QUEUED snapshot written while enqueuing a run routes through the injected store.
   containerTest(
-    "enqueueRun snapshot routes through the store",
+    "the QUEUED snapshot routes through the store",
     async ({ prisma, redisOptions }) => {
       const countingStore = new CountingPostgresRunStore({ prisma, readOnlyPrisma: prisma });
       const engine = new RunEngine(createEngineOptions(redisOptions, prisma, countingStore));

internal-packages/run-engine/src/engine/systems/executionSnapshotSystem.test.ts

@@ -65,6 +65,14 @@ class CountingPostgresRunStore extends PostgresRunStore {
     return super.createExecutionSnapshot(input, tx);
   }
 
+  override async createRun(
+    params: Parameters<PostgresRunStore["createRun"]>[0],
+    tx?: any
+  ): ReturnType<PostgresRunStore["createRun"]> {
+    this.creates++;
+    return super.createRun(params, tx);
+  }
+
   override async findLatestExecutionSnapshot(
     runId: string,
     client?: any

internal-packages/run-engine/src/engine/tests/getSnapshotsSince.test.ts

@@ -1149,11 +1149,16 @@ describe("RunEngine getSnapshotsSince", () => {
         );
 
         await setTimeout(500);
-        await engine.dequeueFromWorkerQueue({
+        const dequeued = await engine.dequeueFromWorkerQueue({
           consumerId: "test_replica_stale_tail",
           workerQueue: "main",
         });
 
+        await engine.startRunAttempt({
+          runId: dequeued[0].run.id,
+          snapshotId: dequeued[0].snapshot.id,
+        });
+
         const allSnapshots = await prisma.taskRunExecutionSnapshot.findMany({
           where: { runId: run.id, isValid: true },
           orderBy: { createdAt: "asc" },