fix(server): bound automatic restart recovery

apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts

@@ -1712,6 +1712,126 @@ it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))(
   },
 );
 
+it.effect.each(["manual", "mcp", "stop"] as const)(
+  "holds a recovered child's parent wake until %s without losing the result",
+  (action) => {
+    return Effect.gen(function* () {
+      const orchestrator = yield* Orchestrator.OrchestratorV2;
+      const sink = yield* EventSink.EventSinkV2;
+      const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests;
+      using offered = vi.spyOn(requests, "offer");
+      const now = yield* DateTime.now;
+      const threadId = ThreadId.make(`thread:recovery-wake:${action}`);
+      const projectId = ProjectId.make(`project:recovery-wake:${action}`);
+      const runId = RunId.make(`run:recovery-wake:${action}`);
+      const rootNodeId = NodeId.make(`node:recovery-wake:${action}`);
+      yield* seedParentWithTerminalTask({
+        threadId,
+        projectId,
+        runId,
+        rootNodeId,
+        taskId: NodeId.make(`node:recovery-wake:seed:${action}`),
+        deliveryState: "acknowledged",
+        now,
+      });
+      const parent = (yield* orchestrator.getThreadProjection(threadId)).runs[0]!;
+      yield* sink.write({
+        commandId: reconcileCommandId(`recovery-wake:${action}`),
+        events: [
+          {
+            id: EventId.make(`event:recovery-wake:cut:${action}`),
+            type: "run.updated",
+            threadId,
+            occurredAt: now,
+            payload: { ...parent, status: "cancelled", completedAt: now },
+          },
+        ],
+      });
+      const child = yield* seedRestartCancelledChild({
+        parentThreadId: threadId,
+        projectId,
+        parentRunId: runId,
+        rootNodeId,
+        name: `recovery-wake-child:${action}`,
+        completionWake: "always",
+        continuationPending: true,
+        now,
+      });
+      yield* continueRestartedRun({
+        threadId: child.childThreadId,
+        sourceRunId: child.childRunId,
+      }).pipe(
+        Effect.provide(ServerSettings.layerTest({ continueThreadsAfterServerUpdate: false })),
+      );
+      yield* orchestrator.recoverDelegatedTasks;
+      const recovered = yield* orchestrator.getThreadProjection(threadId);
+      assert.equal(
+        recovered.subagents.find((task) => task.id === child.taskId)?.status,
+        "cancelled",
+      );
+      assert.isNotNull(recovered.runs[0]?.delegatedCompletion?.delivery);
+      const delivery = recovered.runs[0]!.delegatedCompletion!.delivery!;
+      const wakeCommand = {
+        type: "message.dispatch" as const,
+        threadId,
+        commandId: CommandId.make(`command:recovery-wake:delivery:${action}`),
+        messageId: delivery.messageId,
+        text: "Child cancelled",
+        attachments: [],
+        dispatchMode: { type: "queue_after_active" as const },
+        createdBy: "agent" as const,
+        creationSource: "server" as const,
+        delegatedCompletion: {
+          parentRunId: runId,
+          generation: delivery.generation,
+          taskIds: delivery.taskIds,
+        },
+      };
+      yield* orchestrator.dispatch(wakeCommand);
+      yield* orchestrator.dispatch({
+        ...wakeCommand,
+        commandId: CommandId.make(`command:recovery-wake:repeated:${action}`),
+      });
+      assert.lengthOf((yield* orchestrator.getThreadProjection(threadId)).runs, 1);
+      assert.equal(offered.mock.calls.length, 0);
+      if (action !== "stop") {
+        yield* orchestrator.dispatch({
+          type: "message.dispatch",
+          threadId,
+          commandId: CommandId.make("command:recovery-wake:manual"),
+          messageId: MessageId.make("message:recovery-wake:manual"),
+          text: "Continue",
+          attachments: [],
+          dispatchMode: { type: "start_immediately" },
+          createdBy: action === "mcp" ? "agent" : "user",
+          creationSource: action === "mcp" ? "mcp" : "web",
+        });
+        assert.equal(offered.mock.calls.length, 1);
+        const request = yield* requests.take;
+        assert.equal(request.delegatedCompletion?.messageId, delivery.messageId);
+        yield* orchestrator.dispatch({
+          ...wakeCommand,
+          commandId: CommandId.make("command:recovery-wake:retry"),
+        });
+        const resumed = yield* orchestrator.getThreadProjection(threadId);
+        assert.lengthOf(resumed.runs, 3);
+        assert.equal(
+          resumed.runs.find((run) => run.userMessageId === delivery.messageId)?.status,
+          "queued",
+        );
+      } else {
+        yield* orchestrator.dispatch({
+          type: "thread.stop",
+          threadId,
+          commandId: CommandId.make("command:recovery-wake:stop"),
+        });
+        yield* orchestrator.recoverDelegatedTasks;
+        assert.lengthOf((yield* orchestrator.getThreadProjection(threadId)).runs, 1);
+      }
+    }).pipe(Effect.provide(Layer.merge(layerTest, ProviderContinuationRequests.layer)));
+  },
+);
+
 it.layer(layerTest)("delegated tasks across a server restart", (it) => {
   it.effect("queues a child result without creating a provider history handoff", () =>
     Effect.gen(function* () {

apps/server/src/orchestration-v2/EffectOutbox.ts

@@ -33,6 +33,7 @@ export const OrchestrationEffectRequestV2 = Schema.Union([
   Schema.Struct({
     type: Schema.Literal("provider-runtime.continue"),
     sourceRunId: RunId,
+    processLossCount: Schema.optional(Schema.Int.check(Schema.isGreaterThanOrEqualTo(0))),
   }),
   Schema.Struct({
     type: Schema.Literal("provider-session.detach"),
@@ -321,6 +322,8 @@ export const layer: Layer.Layer<EffectOutboxV2, never, SqlClient.SqlClient> = La
       ).pipe(Effect.asVoid);
     // An earlier effect waiting out a retry backoff still blocks later ones, so
     // a turn cannot start while a failed rollback is about to restore files.
+    // Dispatch finishes before the turn does. Keep admission through that run,
+    // and recover children first so a waiting parent cannot keep them out.
     const claimableCandidatePredicate = (
       availableBefore?: string,
       excludeRestartContinuations = false,
@@ -332,6 +335,31 @@ export const layer: Layer.Layer<EffectOutboxV2, never, SqlClient.SqlClient> = La
             : sql`candidate.available_at <= ${availableBefore}`
         }
         AND candidate.status = 'pending'
+        AND (
+          candidate.effect_type != 'provider-runtime.continue'
+          OR (
+            NOT EXISTS (
+              SELECT 1 FROM orchestration_v2_effect_outbox AS recovery
+              WHERE recovery.effect_type = 'provider-runtime.continue'
+                AND recovery.status = 'running'
+            )
+            AND NOT EXISTS (
+              SELECT 1 FROM orchestration_v2_projection_runs AS recovery_run
+              WHERE recovery_run.status IN ('starting', 'running', 'waiting')
+                AND json_extract(recovery_run.payload_json, '$.restartContinuationOfRunId') IS NOT NULL
+            )
+            AND NOT EXISTS (
+              SELECT 1 FROM orchestration_v2_projection_subagents AS child
+              JOIN orchestration_v2_effect_outbox AS child_recovery
+                ON child_recovery.thread_id = child.child_thread_id
+              WHERE child.thread_id = candidate.thread_id
+                AND child.origin = 'app_owned'
+                AND child.status IN ('pending', 'running', 'waiting')
+                AND child_recovery.effect_type = 'provider-runtime.continue'
+                AND child_recovery.status IN ('pending', 'running')
+            )
+          )
+        )
         AND NOT EXISTS (
           SELECT 1
           FROM orchestration_v2_effect_outbox AS active
@@ -344,7 +372,7 @@ export const layer: Layer.Layer<EffectOutboxV2, never, SqlClient.SqlClient> = La
                 AND ${
                   excludeRestartContinuations
                     ? sql`active.effect_type != 'provider-runtime.continue'`
-                    : sql`1 = 1`
+                    : sql`(active.effect_type != 'provider-runtime.continue' OR candidate.effect_type = 'provider-runtime.continue')`
                 }
               )
             )
@@ -693,6 +721,8 @@ export const layer: Layer.Layer<EffectOutboxV2, never, SqlClient.SqlClient> = La
         const cutoff = DateTime.formatIso(
           DateTime.subtractDuration(yield* DateTime.now, SETTLED_EFFECT_RETENTION),
         );
+        // A live continuation still needs its intent's crash budget on its next
+        // restart, even if that turn runs longer than the normal retention.
         // Each batch is its own statement, so a large backlog never holds the
         // write lock for long and writers can commit between batches.
         const deleteBatch = sql<{ readonly effect_id: string }>`
@@ -702,6 +732,16 @@ export const layer: Layer.Layer<EffectOutboxV2, never, SqlClient.SqlClient> = La
             FROM orchestration_v2_effect_outbox
             WHERE status IN ('succeeded', 'cancelled')
               AND completed_at < ${cutoff}
+              AND NOT (
+                effect_type = 'provider-runtime.continue'
+                AND EXISTS (
+                  SELECT 1 FROM orchestration_v2_projection_runs AS recovery_run
+                  WHERE recovery_run.status IN ('starting', 'running', 'waiting')
+                    AND json_extract(recovery_run.payload_json, '$.restartContinuationOfRunId') IS NOT NULL
+                    AND json_extract(recovery_run.payload_json, '$.restartContinuationOfRunId') =
+                      json_extract(orchestration_v2_effect_outbox.payload_json, '$.sourceRunId')
+                )
+              )
             LIMIT ${PRUNE_BATCH_SIZE}
           )
           RETURNING effect_id

apps/server/src/orchestration-v2/EffectWorker.test.ts

@@ -4,13 +4,16 @@ import {
   CheckpointScopeId,
   CommandId,
   NodeId,
+  ProjectId,
+  ProviderInstanceId,
   ProviderSessionId,
   ProviderThreadId,
   ProviderTurnId,
   RunAttemptId,
   RunId,
   ThreadId,
   VcsProcessExitError,
+  type OrchestrationV2ThreadProjection,
 } from "@t3tools/contracts";
 import * as Cause from "effect/Cause";
 import * as DateTime from "effect/DateTime";
@@ -475,6 +478,62 @@ it.effect("terminalizes a process-bound claim when success settlement fails", ()
   }),
 );
 
+it.effect.each([0, 3, 4])(
+  "enforces process-loss count %s after 22 minutes through the executor",
+  (processLossCount) =>
+    Effect.gen(function* () {
+      const now = yield* DateTime.now;
+      const instanceId = ProviderInstanceId.make("codex");
+      const events = yield* Ref.make<ReadonlyArray<string>>([]);
+      const layer = layerExecutorFor({
+        events,
+        continueAfterRestart: true,
+        threads: {
+          getThreadRecords: () =>
+            Effect.succeed({
+              thread: {
+                id: threadId,
+                projectId: ProjectId.make("project:executor-recovery"),
+                providerInstanceId: instanceId,
+                archivedAt: null,
+                deletedAt: null,
+              },
+              runs: [
+                {
+                  id: runId,
+                  providerInstanceId: instanceId,
+                  status: "cancelled",
+                  requestedAt: DateTime.subtract(now, { minutes: 22 }),
+                  completedAt: now,
+                  ordinal: 1,
+                },
+              ],
+              messages: [],
+              providerTurns: [],
+              runtimeRequests: [],
+              turnItems: [],
+              attempts: [],
+            } as unknown as OrchestrationV2ThreadProjection),
+          dispatch: () =>
+            Ref.update(events, (entries) => [...entries, "dispatch"]).pipe(Effect.as({} as never)),
+          recoverDelegatedTask: () =>
+            Ref.update(events, (entries) => [...entries, "recover-child"]),
+        },
+      });
+      yield* Effect.gen(function* () {
+        const executor = yield* EffectWorker.OrchestrationEffectExecutorV2;
+        yield* executor.execute({
+          ...restartEffect(now, { type: "detach" }),
+          request: { type: "provider-runtime.continue", sourceRunId: runId, processLossCount },
+        });
+      }).pipe(Effect.provide(layer));
+      assert.deepEqual(
+        yield* Ref.get(events),
+        processLossCount > 3 ? ["recover-child"] : ["dispatch", "recover-child"],
+      );
+    }),
+);
+
 it.effect("settles a delegated child once its restart continuation fails for good", () =>
   Effect.gen(function* () {
     const timestamp = DateTime.formatIso(yield* DateTime.now);

apps/server/src/orchestration-v2/EffectWorker.ts

@@ -100,7 +100,13 @@ export const layerExecutor: Layer.Layer<
         switch (effect.request.type) {
           case "provider-runtime.continue": {
             const sourceRunId = effect.request.sourceRunId;
-            return continueRestartedRun({ threadId: effect.threadId, sourceRunId }).pipe(
+            return continueRestartedRun({
+              threadId: effect.threadId,
+              sourceRunId,
+              ...(effect.request.processLossCount === undefined
+                ? {}
+                : { processLossCount: effect.request.processLossCount }),
+            }).pipe(
               Effect.provideService(ThreadManagementService.ThreadManagementService, threads),
               Effect.provideService(ServerSettings.ServerSettingsService, settings),
               Effect.tapError(() =>