feat(run-engine): durable write-ahead guard for manual waitpoint completion

Internal groundwork for recovering a waitpoint completion that commits
before its run-resume fanout is enqueued. The guard is opt-in and has no
user-facing effect on its own.

Mono-RevId: a70cf1849ee8b86a05db41f09c71fa021576de7f

Author: d-cs

apps/webapp/app/v3/runOpsMigration/unblockRouteCatalog.ts

@@ -47,6 +47,12 @@ export const UNBLOCK_ROUTES: readonly UnblockRoute[] = [
     site: WAITPOINT_SYSTEM,
     symbol: "completeWaitpoint (sink declaration)",
   },
+  {
+    id: "wp.ensureCompleted",
+    kind: "RUN",
+    site: WAITPOINT_SYSTEM,
+    symbol: "ensureWaitpointCompleted (completion-guard redelivery)",
+  },
   {
     id: "wp.blockAndComplete",
     kind: "RUN",

internal-packages/run-engine/src/engine/errors.ts

@@ -105,6 +105,36 @@ export class RunOneTimeUseTokenError extends Error {
   }
 }
 
+/**
+ * Wraps an error thrown by `completeWaitpoint` AFTER the write-ahead completion guard was successfully
+ * armed. Its presence is proof the guard is durably persisted and now owns eventual completion, so the
+ * API boundary may return success for a RETRYABLE cause. A failure BEFORE the guard is armed (including
+ * the arm's own enqueue failing) is NOT wrapped, so it propagates and the caller does not report
+ * success. Detect with {@link isWaitpointCompletionGuardArmedError} (a branded property check, robust
+ * across module boundaries) rather than the requested `armGuard` boolean.
+ */
+export class WaitpointCompletionGuardArmedError extends Error {
+  readonly isWaitpointCompletionGuardArmed = true as const;
+  readonly waitpointId: string;
+  readonly cause?: unknown;
+  constructor(waitpointId: string, options?: { cause?: unknown }) {
+    super(`Waitpoint ${waitpointId} completion failed after its durable guard was armed`);
+    this.name = "WaitpointCompletionGuardArmedError";
+    this.waitpointId = waitpointId;
+    this.cause = options?.cause;
+  }
+}
+
+export function isWaitpointCompletionGuardArmedError(
+  error: unknown
+): error is WaitpointCompletionGuardArmedError {
+  return (
+    error instanceof Error &&
+    (error as { isWaitpointCompletionGuardArmed?: unknown }).isWaitpointCompletionGuardArmed ===
+      true
+  );
+}
+
 export class ExecutionSnapshotNotFoundError extends Error {
   constructor(public readonly snapshotId: string) {
     super(`No execution snapshot found for id ${snapshotId}`);

internal-packages/run-engine/src/engine/index.ts

@@ -315,6 +315,12 @@ export class RunEngine {
             deferCount: payload.deferCount,
           });
         },
+        ensureWaitpointCompleted: async ({ payload }) => {
+          await this.waitpointSystem.ensureWaitpointCompleted({
+            waitpointId: payload.waitpointId,
+            output: payload.output,
+          });
+        },
         enqueueDelayedRun: async ({ payload }) => {
           await this.delayedRunSystem.enqueueDelayedRun({ runId: payload.runId });
         },
@@ -423,6 +429,7 @@ export class RunEngine {
       resources,
       executionSnapshotSystem: this.executionSnapshotSystem,
       enqueueSystem: this.enqueueSystem,
+      completionGuardDelayMs: options.completionGuardDelayMs,
     });
 
     this.ttlSystem = new TtlSystem({
@@ -2120,13 +2127,21 @@ export class RunEngine {
   async completeWaitpoint({
     id,
     output,
+    armGuard,
   }: {
     id: string;
     output?: {
       value: string;
       type?: string;
       isError: boolean;
     };
+    /**
+     * Arm the durable write-ahead completion guard for this call. The engine does NOT arm implicitly:
+     * left undefined it defaults to false, so a caller must opt in explicitly. The runtime-flag gate
+     * lives at the guarded boundary (completeWaitpointWithGuard in the webapp), which passes armGuard
+     * only when runStoreInfraRetryEnabled is on. Internal/system callers leave it unset (unarmed).
+     */
+    armGuard?: boolean;
   }): Promise<Waitpoint> {
     // Consult the cross-seam guard FIRST so an unclassifiable id fails loudly
     // here (never a silent local apply). Do NOT branch on decision.store: store routing is
@@ -2136,7 +2151,7 @@ export class RunEngine {
     if (guard) {
       await guard({ waitpointId: id, routeKind: "RESUME_TOKEN" });
     }
-    return this.waitpointSystem.completeWaitpoint({ id, output });
+    return this.waitpointSystem.completeWaitpoint({ id, output, armGuard: armGuard ?? false });
   }
 
   /**