From 6f8fcec2a5122715c9e98056678ab3ce99f654ec Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 29 Oct 2026 15:34:02 -0700 Subject: [PATCH] fix(server): restart continuations retry failed starts and respect a parent's stop After an update the server continues the turns it cut. On a loaded host a resumed Claude session can time out registering its MCP tools before it takes the prompt; that continuation failed once and was never retried, so the restart's promise silently lapsed. A continuation that fails before the provider accepted its prompt now schedules another continuation after 1 and then 3 minutes, unless the failure is a usage, permission or validation error. The pending retry keeps a delegated child's task open, so its parent hears the real outcome, and Stop during the backoff cancels it. A parent's Stop reaches its delegated children through an effect the restart can cut short. Recovery then continued those children as if nothing happened, and they ran until the replayed stop caught up minutes later. A child's continuation now stands down when its parent's run already has a stop request, and the task settles as cancelled. Co-Authored-By: Claude Opus 5.5 --- .../DelegatedCompletionDelivery.test.ts | 59 +++++++ .../src/orchestration-v2/Orchestrator.ts | 51 +++++- .../orchestration-v2/RestartBackgroundNote.ts | 46 ++++- .../RestartContinuation.test.ts | 122 +++++++++++++- .../orchestration-v2/RestartContinuation.ts | 33 +++- .../src/orchestration-v2/runtimeLayer.test.ts | 159 ++++++++++++++++++ 6 files changed, 464 insertions(+), 6 deletions(-) diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index 77e4cd11ba071..ad2bc395d083f 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/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], + ); }), ); diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 05b5601e5a80a..c65005deefb48 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/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), diff --git a/apps/server/src/orchestration-v2/RestartBackgroundNote.ts b/apps/server/src/orchestration-v2/RestartBackgroundNote.ts index ba3830f65453d..24afd61d6186a 100644 --- a/apps/server/src/orchestration-v2/RestartBackgroundNote.ts +++ b/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; @@ -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, + providerTurns: ReadonlyArray>, +): 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; + readonly providerTurns: ReadonlyArray>; + readonly turnItems: Parameters[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, providerTurns: ReadonlyArray>, diff --git a/apps/server/src/orchestration-v2/RestartContinuation.test.ts b/apps/server/src/orchestration-v2/RestartContinuation.test.ts index 061fc54a7f112..d2224720b89f1 100644 --- a/apps/server/src/orchestration-v2/RestartContinuation.test.ts +++ b/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> = []; - 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[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({ diff --git a/apps/server/src/orchestration-v2/RestartContinuation.ts b/apps/server/src/orchestration-v2/RestartContinuation.ts index c81faa6cf47cb..e4d7ac2b7f900 100644 --- a/apps/server/src/orchestration-v2/RestartContinuation.ts +++ b/apps/server/src/orchestration-v2/RestartContinuation.ts @@ -2,6 +2,7 @@ import { runRanAfter } from "@t3tools/shared/orchestrationV2ThreadError"; import { resolveProjectSettings } from "@t3tools/shared/projectSettings"; import { CommandId, + isOrchestrationV2WorkActive, latestProviderTurnForAttempt, MessageId, type OrchestrationV2Run, @@ -19,6 +20,7 @@ import { awaitingProviderThreadId, cancelledRosterTaskWork, cancelledTurnItemWork, + continuationFailedBeforePrompt, isAwaitedWork, isRestartContinuationSource, cancelledBeforeStart, @@ -200,7 +202,10 @@ export const continueRestartedRun = Effect.fn("RestartContinuation.continueResta const source = projection.runs.find((run) => run.id === input.sourceRunId); if ( !source || - !isRestartContinuationSource(source, projection.providerTurns) || + !( + isRestartContinuationSource(source, projection.providerTurns) || + continuationFailedBeforePrompt(source, projection.providerTurns) + ) || restartContinuationBlocked( projection, source, @@ -255,6 +260,32 @@ export const continueRestartedRun = Effect.fn("RestartContinuation.continueResta ) ) return; + // A parent's Stop reaches its open delegated tasks through an effect a restart can cut short. + const { lineage, forkedFrom } = projection.thread; + const taskId = + source.delegatedTaskId ?? (forkedFrom?.type === "node" ? forkedFrom.nodeId : undefined); + if ( + lineage.relationshipToParent === "subagent" && + lineage.parentThreadId !== null && + taskId !== undefined + ) { + const task = (yield* threads.getThreadRecords(lineage.parentThreadId, ["subagents"], { + subagentIds: [taskId], + })).subagents.find((candidate) => candidate.id === taskId); + // A settled task no longer owns later work in its thread. + const parentRunId = + task !== undefined && task.result === null && isOrchestrationV2WorkActive(task.status) + ? task.runId + : null; + if ( + parentRunId !== null && + (yield* threads.getThreadRecords(lineage.parentThreadId, ["turnItems"], { + turnItemTypes: ["run_interrupt_request"], + turnItemRunIds: [parentRunId], + })).turnItems.length > 0 + ) + return; + } const sourceMessage = sourceRecords.messages.find( (message) => message.id === source.userMessageId, ); diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index 11eb866b5fb9c..20158a80d3ff6 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -5465,6 +5465,165 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { }), ); + it.effect.each([false, true])( + "restarts a continuation that failed before its prompt, unless stopped (%s)", + (stopped) => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const eventSink = yield* EventSink.EventSinkV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const threadId = ThreadId.make(`runtime-layer-continuation-start-retry-${stopped}`); + const messageId = (key: string) => MessageId.make(`${threadId}:${key}`); + yield* orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make(`${threadId}:create`), + threadId, + projectId: ProjectId.make(`${threadId}:project`), + title: "Continuation start retry", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: process.cwd(), + }); + const runFor = (key: string) => + Effect.map(orchestrator.getThreadProjection(threadId), ({ runs }) => + runs.find((candidate) => candidate.userMessageId === messageId(key)), + ); + const write = ( + key: string, + payload: OrchestrationV2Run, + extra: Parameters[0]["events"] = [], + ) => + Effect.gen(function* () { + const now = yield* DateTime.now; + yield* eventSink.write({ + events: [ + ...extra, + { + id: EventId.make(`${threadId}:${key}`), + type: "run.updated", + threadId, + runId: payload.id, + providerInstanceId: payload.providerInstanceId, + occurredAt: now, + payload: { ...payload, startedAt: now, completedAt: now }, + }, + ], + }); + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make(`${threadId}:message:cut`), + threadId, + messageId: messageId("cut"), + text: "Work that the restart cut", + attachments: [], + modelSelection, + dispatchMode: { type: "start_immediately" }, + }); + const cut = (yield* runFor("cut"))!; + yield* write("cut:cancelled", { ...cut, status: "cancelled" }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "agent", + creationSource: "server", + commandId: CommandId.make(`${threadId}:message:continuation`), + threadId, + messageId: messageId("continuation"), + text: "Continue where you left off.", + attachments: [], + modelSelection, + dispatchMode: { type: "start_immediately" }, + restartContinuationOfRunId: cut.id, + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make(`${threadId}:message:queued`), + threadId, + messageId: messageId("queued"), + text: "Queued behind the continuation", + attachments: [], + modelSelection, + dispatchMode: { type: "queue_after_active" }, + }); + const continuation = (yield* runFor("continuation"))!; + assert.isNotNull(continuation.rootNodeId); + const queued = (yield* runFor("queued"))!; + // The failure holds the queue after the reactor scheduled any retry. + const held = yield* eventSink + .stream({ threadId, afterSequence: yield* orchestrator.getThreadEventSequence(threadId) }) + .pipe( + Stream.filter( + (stored) => + stored.event.type === "run.updated" && + stored.event.payload.id === queued.id && + stored.event.payload.queueHeld === true, + ), + Stream.runHead, + Effect.forkScoped, + ); + yield* Effect.yieldNow; + const now = yield* DateTime.now; + yield* write("continuation:failed", { ...continuation, status: "failed" }, [ + { + id: EventId.make(`${threadId}:continuation:error`), + type: "turn-item.updated", + threadId, + runId: continuation.id, + nodeId: continuation.rootNodeId, + providerInstanceId: continuation.providerInstanceId, + occurredAt: now, + payload: { + id: TurnItemId.make(`${threadId}:continuation:error`), + type: "error", + threadId, + runId: continuation.id, + nodeId: continuation.rootNodeId, + providerThreadId: continuation.providerThreadId, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 2, + status: "failed", + title: "Provider failure", + startedAt: now, + completedAt: now, + updatedAt: now, + failure: { + class: "provider_error", + message: "Claude timed out while registering its MCP tools.", + code: null, + retryable: null, + }, + }, + }, + ]); + yield* Fiber.join(held); + const retriesOf = Effect.map(orchestrator.getThreadProjection(threadId), ({ runs }) => + runs.filter((run) => run.restartContinuationOfRunId === continuation.id), + ); + yield* worker.drain(); + assert.deepEqual(yield* retriesOf, [], "the retry waits out its backoff"); + if (stopped) { + yield* orchestrator.dispatch({ + type: "thread.stop", + commandId: CommandId.make(`${threadId}:stop`), + threadId, + }); + } + yield* TestClock.adjust("61 seconds"); + yield* worker.drain(); + assert.lengthOf(yield* retriesOf, stopped ? 0 : 1); + }), + ); + it.effect.each(["startup", "shutdown"] as const)( "preserves and holds queued messages across %s until explicitly resumed", (trigger) =>