fix(server): restart continuations retry failed starts and respect a parent's stop

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

@@ -2448,6 +2448,65 @@ it.layer(layerTest)("delegated tasks across a server restart", (it) => {
       const task = settled.subagents.find((row) => row.id === child.taskId);
       assert.equal(task?.status, "cancelled");
       assert.equal(task?.completionDelivery?.state, "claimed");
+
+      // The parent's Stop committed, but the restart cut the effect that stops its children.
+      const stoppedChild = yield* seedRestartCancelledChild({
+        parentThreadId: threadId,
+        projectId,
+        parentRunId: runId,
+        rootNodeId,
+        name: "restart-parent-stopped-child",
+        completionWake: "always",
+        continuationPending: true,
+        now,
+      });
+      yield* (yield* EventSink.EventSinkV2).write({
+        commandId: CommandId.make("command:restart-declined-parent:stop"),
+        events: [
+          {
+            id: EventId.make("event:restart-declined-parent:interrupt-request"),
+            type: "turn-item.updated",
+            threadId,
+            runId,
+            nodeId: rootNodeId,
+            providerInstanceId: modelSelection.instanceId,
+            occurredAt: now,
+            payload: {
+              id: TurnItemId.make("turn-item:restart-declined-parent:interrupt-request"),
+              type: "run_interrupt_request",
+              threadId,
+              runId,
+              nodeId: rootNodeId,
+              providerThreadId: null,
+              providerTurnId: null,
+              nativeItemRef: null,
+              parentItemId: null,
+              ordinal: 90,
+              status: "completed",
+              title: "Interrupt requested",
+              startedAt: now,
+              completedAt: now,
+              updatedAt: now,
+              message: "Stopped by user",
+            },
+          },
+        ],
+      });
+      yield* continueRestartedRun({
+        threadId: stoppedChild.childThreadId,
+        sourceRunId: stoppedChild.childRunId,
+      }).pipe(Effect.provide(ServerSettings.layerTest({ continueThreadsAfterServerUpdate: true })));
+      const afterStop = yield* orchestrator.getThreadProjection(threadId);
+      assert.equal(
+        afterStop.subagents.find((row) => row.id === stoppedChild.taskId)?.status,
+        "cancelled",
+      );
+      assert.deepEqual(
+        (yield* orchestrator.getThreadProjection(stoppedChild.childThreadId)).runs.map(
+          (run) => run.id,
+        ),
+        [stoppedChild.childRunId],
+      );
     }),
   );
 

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

@@ -107,6 +107,8 @@ import { ContextHandoffServiceV2 } from "./ContextHandoffService.ts";
 import { notificationTurnItem } from "@t3tools/provider-core/server/notification";
 import {
   cancelledBeforeStart,
+  continuationFailedBeforePrompt,
+  continuationStartRetryDelayMs,
   isRestartContinuationSource,
   runOutcome,
 } from "./RestartBackgroundNote.ts";
@@ -2209,6 +2211,35 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
       ),
   );
 
+  // Under load a resumed provider can time out before taking the prompt; the restart still owes the turn.
+  const retryContinuationStart = Effect.fn("Orchestrator.retryContinuationStart")(function* (
+    threadId: ThreadId,
+    runId: RunId,
+  ) {
+    const projection = yield* projectionStore.getThreadRecords(
+      threadId,
+      ["runs", "providerTurns", "turnItems"],
+      { turnItemTypes: ["error"], turnItemRunIds: [runId] },
+    );
+    const run = projection.runs.find((candidate) => candidate.id === runId);
+    if (run === undefined) return;
+    const delayMs = continuationStartRetryDelayMs(run, projection);
+    if (delayMs === undefined) return;
+    const commandId = CommandId.make(`command:restart-continuation-retry:${run.id}`);
+    yield* writeSystemEvents(
+      [],
+      [
+        {
+          id: `effect:restart-continuation:${run.id}`,
+          commandId,
+          threadId,
+          request: { type: "provider-runtime.continue", sourceRunId: run.id },
+          availableAt: DateTime.add(yield* DateTime.now, { milliseconds: delayMs }),
+        },
+      ],
+    );
+  });
+
   // A hold waits for a person to review what they queued. Nobody reviews messages from agents,
   // the server or scheduled tasks, so a later completed run releases those unless it was stopped.
   const releaseUnattendedHolds = Effect.fn("Orchestrator.releaseUnattendedHolds")(function* (
@@ -5188,7 +5219,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
       const source = projection.runs.find((run) => run.id === command.restartContinuationOfRunId);
       if (
         !source ||
-        !isRestartContinuationSource(source, projection.providerTurns) ||
+        !(
+          isRestartContinuationSource(source, projection.providerTurns) ||
+          continuationFailedBeforePrompt(source, projection.providerTurns)
+        ) ||
         projection.thread.archivedAt !== null ||
         projection.thread.deletedAt !== null ||
         projection.thread.providerInstanceId !== source.providerInstanceId ||
@@ -10419,7 +10453,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
       // A resumed queue can have a lower ordinal than a pending continuation source.
       // Stop must reach those sources even when interrupting the resumed run succeeds.
       if (
-        run.status === "cancelled" &&
+        (run.status === "cancelled" ||
+          (run.status === "failed" && run.restartContinuationOfRunId !== undefined)) &&
         (yield* awaitsRestartContinuation(run).pipe(mapDispatchError(input.command)))
       ) {
         yield* markStoppedRun({ ...input, thread, run });
@@ -13129,6 +13164,18 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
       // while writing child events, so nesting parent inside child would invert the order.
       // The keyed semaphores are neither reentrant nor deadlock-aware.
       if (delegatedTaskTerminalStatus(run.status) !== null) {
+        // Before the parent hears of the failure: a pending continuation keeps the task open.
+        if (run.status === "failed" && run.restartContinuationOfRunId !== undefined) {
+          yield* threadDispatch.withLock(threadId, retryContinuationStart(threadId, run.id)).pipe(
+            Effect.catchCause((cause) =>
+              Effect.logWarning("Failed to schedule a restart continuation retry", {
+                threadId,
+                runId: run.id,
+                cause,
+              }),
+            ),
+          );
+        }
         yield* threadDispatch.withLock(
           threadId,
           finalizeDelegatedCompletionDelivery(threadId, run.id),

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

@@ -6,7 +6,7 @@ import type {
   OrchestrationV2RunAttempt,
   OrchestrationV2TurnItem,
 } from "@t3tools/contracts";
-import { runRanAfter } from "@t3tools/shared/orchestrationV2ThreadError";
+import { latestRootProviderFailure, runRanAfter } from "@t3tools/shared/orchestrationV2ThreadError";
 
 type Work = OrchestrationV2RestartCancelledBackgroundWork;
 type Attempt = Pick<OrchestrationV2RunAttempt, "id" | "runId">;
@@ -181,6 +181,50 @@ export function cancelledBeforeStart(
   return run.status === "cancelled" && neverStarted(run);
 }
 
+/** The provider never accepted this continuation's prompt, so starting it again repeats nothing. */
+export function continuationFailedBeforePrompt(
+  run: Pick<OrchestrationV2Run, "status" | "activeAttemptId" | "restartContinuationOfRunId">,
+  providerTurns: ReadonlyArray<Pick<ProviderTurnState, "runAttemptId">>,
+): boolean {
+  return (
+    run.status === "failed" &&
+    run.restartContinuationOfRunId !== undefined &&
+    !providerTurns.some((turn) => turn.runAttemptId === run.activeAttemptId)
+  );
+}
+
+const CONTINUATION_START_RETRY_DELAYS_MS = [60_000, 180_000];
+
+/** Backoff before restarting a continuation that failed to start, or undefined to let it fail. */
+export function continuationStartRetryDelayMs(
+  run: OrchestrationV2Run,
+  projection: {
+    readonly runs: ReadonlyArray<OrchestrationV2Run>;
+    readonly providerTurns: ReadonlyArray<Pick<ProviderTurnState, "runAttemptId">>;
+    readonly turnItems: Parameters<typeof latestRootProviderFailure>[1];
+  },
+): number | undefined {
+  if (!continuationFailedBeforePrompt(run, projection.providerTurns)) return;
+  // These fail the same way however often they start.
+  const failureClass = latestRootProviderFailure(run, projection.turnItems)?.class;
+  if (
+    failureClass === "usage_limit" ||
+    failureClass === "permission_error" ||
+    failureClass === "validation_error"
+  )
+    return;
+  let failedStarts = 0;
+  for (
+    let current: OrchestrationV2Run | undefined = run;
+    current !== undefined && continuationFailedBeforePrompt(current, projection.providerTurns);
+    current = projection.runs.find(
+      (candidate) => candidate.id === current?.restartContinuationOfRunId,
+    )
+  )
+    failedStarts += 1;
+  return CONTINUATION_START_RETRY_DELAYS_MS[failedStarts - 1];
+}
+
 export function isRestartContinuationSource(
   source: Pick<OrchestrationV2Run, "status" | "activeAttemptId" | "restartCancelledBackgroundWork">,
   providerTurns: ReadonlyArray<Pick<ProviderTurnState, "runAttemptId" | "status">>,

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

@@ -20,6 +20,7 @@ import * as Effect from "effect/Effect";
 import * as Layer from "effect/Layer";
 import * as ServerSettings from "../serverSettings.ts";
 import { restartContinuationRun, continueRestartedRun } from "./RestartContinuation.ts";
+import { continuationStartRetryDelayMs } from "./RestartBackgroundNote.ts";
 import * as ThreadManagementService from "./ThreadManagementService.ts";
 import * as ProviderRuntimeRecovery from "./ProviderRuntimeRecoveryService.ts";
 import * as ProjectionStore from "./ProjectionStore.ts";
@@ -46,6 +47,8 @@ function makeProjection() {
       providerInstanceId: instanceId,
       archivedAt: null,
       deletedAt: null,
+      lineage: { parentThreadId: null, relationshipToParent: null },
+      forkedFrom: null,
     },
     runs: [
       {
@@ -839,10 +842,13 @@ it.effect("does not cancel or resume a run that completes while shutdown intent
   }),
 );
 
-const continuationDispatches = (projection: OrchestrationV2ThreadProjection) =>
+const continuationDispatches = (
+  projection: OrchestrationV2ThreadProjection,
+  sourceRunId: RunId = runId,
+) =>
   Effect.gen(function* () {
     const dispatches: Array<Extract<OrchestrationV2Command, { type: "message.dispatch" }>> = [];
-    yield* continueRestartedRun({ threadId, sourceRunId: runId }).pipe(
+    yield* continueRestartedRun({ threadId, sourceRunId }).pipe(
       Effect.provide(
         Layer.merge(
           Layer.mock(ThreadManagementService.ThreadManagementService)({
@@ -943,6 +949,118 @@ it.effect("continues a resumed queued run that ran after an earlier continuation
   }),
 );
 
+it("backs off twice from a continuation the provider never took, then lets it fail", () => {
+  const base = makeProjection();
+  const failedStart = (id: string, of: RunId) => ({
+    ...base.runs[0]!,
+    id: RunId.make(id),
+    activeAttemptId: RunAttemptId.make(`attempt:${id}`),
+    status: "failed" as const,
+    restartContinuationOfRunId: of,
+  });
+  const first = failedStart("run:retry-1", runId);
+  const second = failedStart("run:retry-2", first.id);
+  const third = failedStart("run:retry-3", second.id);
+  const projection = (failureClass = "provider_error", providerTurns = base.providerTurns) =>
+    ({
+      runs: [{ ...base.runs[0]!, status: "cancelled" }, first, second, third],
+      providerTurns,
+      turnItems: [first, second, third].map((run) => ({
+        id: `error:${run.id}`,
+        type: "error",
+        status: "failed",
+        runId: run.id,
+        nodeId: run.rootNodeId,
+        ordinal: 1,
+        updatedAt: now,
+        failure: { class: failureClass, message: "Failed.", code: null, retryable: null },
+      })),
+    }) as unknown as Parameters<typeof continuationStartRetryDelayMs>[1];
+  assert.deepEqual(
+    [first, second, third].map((run) => continuationStartRetryDelayMs(run, projection())),
+    [60_000, 180_000, undefined],
+  );
+  for (const failureClass of ["usage_limit", "permission_error", "validation_error"])
+    assert.isUndefined(continuationStartRetryDelayMs(first, projection(failureClass)));
+  // Once the provider took the prompt, starting again would repeat the turn.
+  assert.isUndefined(
+    continuationStartRetryDelayMs(
+      first,
+      projection("provider_error", [
+        { ...base.providerTurns[0]!, runAttemptId: first.activeAttemptId, status: "failed" },
+      ] as never),
+    ),
+  );
+});
+
+it.effect("continues again after a continuation fails before its prompt", () =>
+  Effect.gen(function* () {
+    const cut = cutMidTurn();
+    const failedStart = {
+      ...cut.runs[0]!,
+      id: RunId.make("run:failed-start"),
+      ordinal: 2,
+      activeAttemptId: RunAttemptId.make("attempt:failed-start"),
+      status: "failed",
+      restartContinuationOfRunId: runId,
+    };
+    const dispatches = yield* continuationDispatches(
+      { ...cut, runs: [...cut.runs, failedStart] } as unknown as OrchestrationV2ThreadProjection,
+      failedStart.id,
+    );
+    assert.deepEqual(
+      dispatches.map((command) => command.restartContinuationOfRunId),
+      [failedStart.id],
+    );
+  }),
+);
+
+it.effect.each([
+  ["running", null, 0],
+  ["completed", "Done.", 1],
+] as const)(
+  "a parent's Stop on the task's run blocks the child only while the task is %s",
+  ([status, result, expected]) =>
+    Effect.gen(function* () {
+      const parentThreadId = ThreadId.make("thread:parent");
+      const parentRunId = RunId.make("run:parent");
+      const taskId = NodeId.make("node:task");
+      const cut = cutMidTurn();
+      const child = {
+        ...cut,
+        thread: {
+          ...cut.thread,
+          lineage: { parentThreadId, relationshipToParent: "subagent" },
+          forkedFrom: { type: "node", nodeId: taskId },
+        },
+      } as unknown as OrchestrationV2ThreadProjection;
+      const parent = {
+        thread: { id: parentThreadId },
+        subagents: [{ id: taskId, runId: parentRunId, origin: "app_owned", status, result }],
+        turnItems: [{ id: "turn-item:stop", runId: parentRunId, type: "run_interrupt_request" }],
+      };
+      let dispatched = 0;
+      yield* continueRestartedRun({ threadId, sourceRunId: runId }).pipe(
+        Effect.provide(
+          Layer.merge(
+            Layer.mock(ThreadManagementService.ThreadManagementService)({
+              getThreadRecords: (id) =>
+                Effect.succeed((id === parentThreadId ? parent : child) as never),
+              recoverDelegatedTask: () => Effect.void,
+              dispatch: () =>
+                Effect.sync(() => {
+                  dispatched += 1;
+                  return {} as never;
+                }),
+            }),
+            ServerSettings.layerTest({ continueThreadsAfterServerUpdate: true }),
+          ),
+        ),
+      );
+      assert.equal(dispatched, expected);
+    }),
+);
+
 it.effect("does not continue a run the user asked to stop before the restart", () =>
   Effect.gen(function* () {
     const texts = yield* continuationTexts({