Merge remote-tracking branch 'origin/feat/waitpoint-envelope-resolver-tri-13441' into feat/waitpoint-mint-flag-wiring-tri-13442

Author: d-cs

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

@@ -40,7 +40,7 @@ const _pointerKeys: Exact<keyof CompletedWaitpointsPointer, "cycleSeq" | "count"
 
 const _argsKeys: Exact<
   keyof ResolveCompletedWaitpointsArgs,
-  "runId" | "batchId" | "pointer" | "order" | "records"
+  "runId" | "batchId" | "pointer" | "order" | "distinctIds" | "records"
 > = true;
 import { enhanceExecutionSnapshotWithWaitpoints } from "./executionSnapshotSystem.js";
 
@@ -193,6 +193,7 @@ async function assertParity(
     batchId: batchId ?? undefined,
     pointer: { cycleSeq: 1, count: order.length },
     order,
+    distinctIds: [...new Set(waitpoints.map((w) => w.id))],
     records: waitpoints.map(toRecord),
   };
   // count-carried-forward behaviour (order.length, not the record count) is covered by
@@ -406,6 +407,7 @@ describe("the completed-waitpoints freeze", () => {
       batchId: undefined,
       pointer: { cycleSeq: 1, count: 1 },
       order: ["wp_hook"],
+      distinctIds: ["wp_hook"],
       records: [toRecord(w)],
     });
     expect(resolved).toHaveLength(1);
@@ -611,6 +613,7 @@ describe("the exhaustive parity grid", () => {
                           batchId: readingBatchId ?? undefined,
                           pointer: { cycleSeq: 1, count: order.length },
                           order,
+                          distinctIds: [w.id],
                           records: [toRecord(w)],
                         };
                         const resolved = await referenceResolver(

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

@@ -173,7 +173,7 @@ async function getSnapshotWaitpointIdsWithPresence(
  * This is necessary because waitpoints can have large outputs (100KB+),
  * and fetching many at once can exceed Node.js string limits.
  */
-async function fetchWaitpointsInChunks(
+export async function fetchWaitpointsInChunks(
   prisma: PrismaClientOrTransaction,
   waitpointIds: string[],
   runStore?: RunStore,

internal-packages/run-engine/src/engine/systems/waitpointSystem.ts

@@ -491,14 +491,6 @@ export class WaitpointSystem {
         };
       }
 
-      // The record set rides the wait cycle's key once per resume, so build it here rather
-      // than at each append site. Nothing mints a store-format waitpoint yet, so
-      // #completedWaitpointRecordsFor returns undefined on every live path today.
-      const completedWaitpointRecords = await this.#completedWaitpointRecordsFor(
-        runId,
-        blockingWaitpoints
-      );
-
       // 3. Get the run (run-ops scalars) + resolve its environment via the control-plane resolver,
       // so the run-ops DB can split without a cross-provider join.
       const run = await this.$.runStore.findRun(
@@ -616,6 +608,13 @@ export class WaitpointSystem {
           };
         }
         case "EXECUTING_WITH_WAITPOINTS": {
+          // Built inside the branch, not before the switch: the statuses above return without
+          // appending, and they must not pay an envelope read to do it.
+          const completedWaitpointRecords = await this.#completedWaitpointRecordsFor(
+            runId,
+            blockingWaitpoints
+          );
+
           const newSnapshot = await this.executionSnapshotSystem.createExecutionSnapshot(
             this.$.prisma,
             {
@@ -684,6 +683,11 @@ export class WaitpointSystem {
             );
           }
 
+          const completedWaitpointRecords = await this.#completedWaitpointRecordsFor(
+            runId,
+            blockingWaitpoints
+          );
+
           //put it back in the queue, with the original timestamp (w/ priority)
           //this prioritizes dequeuing waiting runs over new runs
           const newSnapshot = await this.enqueueSystem.enqueueRun({
@@ -761,31 +765,36 @@ export class WaitpointSystem {
   }
 
   /**
-   * The record set for one resume, or undefined when this wait has no store-resident half.
+   * The record set for one resume, or undefined when no blocking waitpoint carries a store-format
+   * id.
+   *
+   * Gated on id FORMAT, not residency. The two are not the same during a migration: a
+   * store-format id can still be served by the Postgres arm, exactly as run-ops ids were for
+   * runs. Whichever arm owns it answers, so the gate only decides whether to ask at all.
    *
-   * The classification gate is what keeps this inert. `parseWaitpointId` reports legacy for
-   * every id minted today, so no live resume reads an envelope or writes a record until a
-   * waitpoint mints in store format.
+   * That gate is what keeps this inert. `parseWaitpointId` reports legacy for every id minted
+   * today, so no live resume reads an envelope or writes a record until a waitpoint mints in
+   * store format.
    */
   async #completedWaitpointRecordsFor(
     runId: string,
     blockingWaitpoints: RunBlockEdge[]
   ): Promise<CompletedWaitpointRecord[] | undefined> {
-    const storeResidentIds = [
+    const storeFormatIds = [
       ...new Set(
         blockingWaitpoints
           .map((b) => b.waitpoint.id)
           .filter((id) => parseWaitpointId(id).format === "b32hexW")
       ),
     ];
 
-    if (storeResidentIds.length === 0) {
+    if (storeFormatIds.length === 0) {
       return undefined;
     }
 
     const sources = await this.coordinator.readCompletionEnvelopes({
       runId,
-      waitpointIds: storeResidentIds,
+      waitpointIds: storeFormatIds,
     });
 
     return buildCompletedWaitpointRecords(sources);