From 2f36b6e7a6a277942717f1849c2af9226f67fe06 Mon Sep 17 00:00:00 2001 From: hahasagined <3049408+hahasagined@users.noreply.github.com> Date: Thu, 22 Oct 2026 12:37:31 +0000 Subject: [PATCH 1/2] fix(server): Codex questions stay answerable across server restarts A pending native Codex request_user_input question was expired on startup (or cancelled on graceful shutdown) because its live callback dies with the provider process. The waiting turn could never receive the answer. Recovery now turns a pending root Codex question into a reply question (responseCapability "message") instead of expiring it. The waiting run is still cancelled and is not continued automatically; answering the question sends the answer as a message that resumes the same native conversation, through the existing async-question path, which commits one answer. Only sessions whose adapter sets the new questionsAnswerableAfterRestart capability (Codex) qualify, and only for root conversations with a strong native identity in the run being cut. Graceful shutdown leaves those questions to recovery instead of cancelling them on session release. Manual session close, approvals, secrets, subagent and delegated-task questions and other providers keep their current handling. Fixes #23597 Co-Authored-By: Claude Opus 5.5 --- .../Adapters/CodexAdapterV2.ts | 1 + .../ProviderRuntimeRecoveryService.test.ts | 174 +++++++++++++ .../ProviderRuntimeRecoveryService.ts | 76 +++++- .../ProviderSessionManager.test.ts | 82 ++++++ .../ProviderSessionManager.ts | 8 + .../src/orchestration-v2/runtimeLayer.test.ts | 238 ++++++++++++++++++ docs/user/updating.md | 6 +- packages/contracts/src/orchestrationV2.ts | 5 + 8 files changed, 586 insertions(+), 4 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 067551ff9a91..0d8d40168a9c 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -242,6 +242,7 @@ export const CodexProviderCapabilitiesV2 = { supportsProviderSwitchingViaHandoff: true, supportsRuntimeModeSwitchInSession: true, pendingRequestsSurviveRestart: false, + questionsAnswerableAfterRestart: true, }, threads: { canCreateEmptyThread: true, diff --git a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts index d0a97adb106f..80382409b1e5 100644 --- a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts @@ -12,6 +12,7 @@ import { RuntimeRequestId, ThreadId, TurnItemId, + type OrchestrationV2ProviderCapabilities, type OrchestrationV2ThreadProjection, } from "@t3tools/contracts"; import * as Effect from "effect/Effect"; @@ -24,6 +25,7 @@ import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import * as ProjectionStore from "./ProjectionStore.ts"; import * as ProviderRuntimeRecovery from "./ProviderRuntimeRecoveryService.ts"; import * as ServerSettings from "../serverSettings.ts"; +import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; it.effect("expires orphaned runtime requests before command readiness", () => { const threadId = ThreadId.make("thread_recovery_requests"); @@ -131,6 +133,178 @@ it.effect("preserves async questions across startup and shutdown", () => { }).pipe(Effect.provide(layer)); }); +const questionProjection = () => { + const threadId = ThreadId.make("thread_restart_question"); + const providerInstanceId = ProviderInstanceId.make("codex"); + const driver = ProviderDriverKind.make("codex"); + const providerSessionId = ProviderSessionId.make("session_restart_question"); + const providerThreadId = ProviderThreadId.make("provider_thread_restart_question"); + const runId = RunId.make("run_restart_question"); + const nodeId = NodeId.make("node_restart_question"); + const requestId = RuntimeRequestId.make("request_restart_question"); + return { + thread: { + id: threadId, + providerInstanceId, + lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId }, + deletedAt: null, + archivedAt: null, + }, + runs: [{ id: runId, status: "running", providerInstanceId, providerThreadId }], + attempts: [], + nodes: [{ id: nodeId, runId, providerThreadId, status: "waiting" }], + subagents: [], + providerSessions: [ + { + id: providerSessionId, + driver, + providerInstanceId, + status: "ready", + capabilities: CodexProviderCapabilitiesV2 as OrchestrationV2ProviderCapabilities, + }, + ], + providerThreads: [ + { + id: providerThreadId, + driver, + providerInstanceId, + providerSessionId, + appThreadId: threadId, + ownerNodeId: null, + status: "active", + nativeThreadRef: { driver, nativeId: "native_restart_question", strength: "strong" }, + pendingBackgroundTasks: [], + }, + ], + providerTurns: [], + runtimeRequests: [ + { + id: requestId, + nodeId, + providerTurnId: null, + kind: "user_input", + status: "pending", + responseCapability: { type: "live", providerSessionId }, + }, + ], + messages: [], + turnItems: [ + { + id: TurnItemId.make("item_restart_question"), + type: "user_input_request", + requestId, + runId, + nodeId, + status: "waiting", + questions: [{ id: "choice", header: "Choice", question: "Alpha or Beta?", options: [] }], + }, + ], + }; +}; +type QuestionProjection = ReturnType; + +const recoveredQuestion = (projection: QuestionProjection) => { + let committed: Parameters[0] | undefined; + const layer = ProviderRuntimeRecovery.layer.pipe( + Layer.provide(ServerSettings.layerTest()), + Layer.provide( + Layer.mergeAll( + Layer.mock(ProjectionStore.ProjectionStoreV2)({ + getRecoveryThreadIds: () => Effect.succeed([projection.thread.id]), + getRuntimeRecoveryProjection: () => + Effect.succeed(projection as unknown as OrchestrationV2ThreadProjection), + }), + Layer.mock(EventSink.EventSinkV2)({ + commitCommand: (input) => { + committed = input; + return Effect.succeed({ committed: true, cancelledEffectCount: 0 } as never); + }, + }), + IdAllocator.layer, + Layer.mock(EffectOutbox.EffectOutboxV2)({ + reconcileAfterProcessLoss: Effect.succeed({ requeued: 0, cancelled: 0 }), + }), + ), + ), + ); + return Effect.gen(function* () { + yield* (yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService).reconcile("startup"); + return committed?.events.flatMap((event) => + event.type === "runtime-request.updated" ? [event.payload] : [], + ); + }).pipe(Effect.provide(layer)); +}; + +it.effect("keeps a root Codex question answerable as a reply after restart", () => + Effect.gen(function* () { + const [request] = (yield* recoveredQuestion(questionProjection())) ?? []; + assert.equal(request?.status, "pending"); + assert.deepEqual(request?.responseCapability, { type: "message" }); + }), +); + +it.effect.each<[string, (projection: QuestionProjection) => QuestionProjection]>([ + [ + "a provider-native subagent", + (projection) => ({ + ...projection, + providerThreads: projection.providerThreads.map((thread) => ({ + ...thread, + ownerNodeId: NodeId.make("node_parent_tool_call") as never, + })), + }), + ], + [ + "a delegated task thread", + (projection) => ({ + ...projection, + thread: { + ...projection.thread, + lineage: { ...projection.thread.lineage, relationshipToParent: "subagent" as never }, + }, + }), + ], + [ + "a provider that cannot resume from a reply", + (projection) => ({ + ...projection, + providerSessions: projection.providerSessions.map((session) => ({ + ...session, + capabilities: { + ...session.capabilities, + sessions: { ...session.capabilities.sessions, questionsAnswerableAfterRestart: false }, + }, + })), + }), + ], + [ + "a conversation without a strong native identity", + (projection) => ({ + ...projection, + providerThreads: projection.providerThreads.map((thread) => ({ + ...thread, + nativeThreadRef: { ...thread.nativeThreadRef, strength: "weak" }, + })), + }), + ], + [ + "an approval", + (projection) => ({ + ...projection, + runtimeRequests: projection.runtimeRequests.map((request) => ({ + ...request, + kind: "command", + })), + }), + ], +])("expires the question of %s after restart", ([, change]) => + Effect.gen(function* () { + const [request] = (yield* recoveredQuestion(change(questionProjection()))) ?? []; + assert.equal(request?.status, "expired"); + assert.equal(request?.responseCapability.type, "not_resumable"); + }), +); + it.effect( "preserves a replayable waiting run while cancelling its process-bound background work", () => { diff --git a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts index a3c41fed25f5..9b0ad01baa34 100644 --- a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts +++ b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts @@ -151,6 +151,61 @@ const cancelledItem = ( ? { ...item, status: "cancelled", secretStatus: "cancelled", completedAt: now, updatedAt: now } : { ...item, status: "cancelled", completedAt: now, updatedAt: now }; +/** + * A native question's callback dies with its provider process. When the root + * conversation it blocked can resume from a follow-up message, it stays + * answerable as a reply instead, and nothing continues until it is answered. + */ +function answerableAfterRestart( + projection: ProjectionStore.ProjectionRuntimeRecoveryState, + request: OrchestrationV2ThreadProjection["runtimeRequests"][number], + cutRunIds: ReadonlySet, +): boolean { + if ( + request.kind !== "user_input" || + request.status !== "pending" || + request.responseCapability.type !== "live" || + projection.thread.lineage.relationshipToParent === "subagent" || + projection.thread.deletedAt !== null + ) + return false; + const { providerSessionId } = request.responseCapability; + const node = projection.nodes.find((candidate) => candidate.id === request.nodeId); + if (node?.runId == null || !cutRunIds.has(node.runId)) return false; + const runId = node.runId; + const run = projection.runs.find((candidate) => candidate.id === runId); + const providerThread = projection.providerThreads.find( + (candidate) => candidate.id === node.providerThreadId, + ); + const session = projection.providerSessions.find( + (candidate) => candidate.id === providerSessionId, + ); + return ( + run !== undefined && + providerThread !== undefined && + session !== undefined && + session.capabilities.sessions.questionsAnswerableAfterRestart === true && + session.driver === providerThread.driver && + providerThread.providerSessionId === providerSessionId && + run.providerThreadId === providerThread.id && + run.providerInstanceId === providerThread.providerInstanceId && + projection.thread.providerInstanceId === providerThread.providerInstanceId && + providerThread.appThreadId === projection.thread.id && + providerThread.ownerNodeId === null && + providerThread.status !== "closed" && + providerThread.status !== "archived" && + providerThread.nativeThreadRef?.nativeId != null && + providerThread.nativeThreadRef.strength === "strong" && + providerThread.nativeThreadRef.driver === providerThread.driver && + projection.turnItems.some( + (item) => + item.type === "user_input_request" && + item.requestId === request.id && + item.questions.length > 0, + ) + ); +} + function latestStartedRun( projection: ProjectionStore.ProjectionRuntimeRecoveryState, providerThreadId: ProviderThreadId, @@ -228,16 +283,24 @@ export const make = Effect.gen(function* () { } runs.push(run); } + const cutRunIds = new Set(runs.map((run) => run.id)); + const answerableRequests = projection.runtimeRequests.filter((request) => + answerableAfterRestart(projection, request, cutRunIds), + ); const messageRequestNodeIds = new Set( projection.runtimeRequests .filter( (request) => request.status === "pending" && request.responseCapability.type === "message", ) + .concat(answerableRequests) .map((request) => request.nodeId), ); const requests = projection.runtimeRequests.filter( - (request) => request.status === "pending" && request.responseCapability.type !== "message", + (request) => + request.status === "pending" && + request.responseCapability.type !== "message" && + !answerableRequests.includes(request), ); // Delegated task rows, items and nodes stay open: the child settles them. const delegatedTaskNodeIds = new Set([ @@ -334,6 +397,16 @@ export const make = Effect.gen(function* () { payload: { ...run, queueHeld: true }, }); } + for (const request of answerableRequests) { + events.push({ + id: yield* allocateEventId(), + type: "runtime-request.updated", + threadId: projection.thread.id, + nodeId: request.nodeId, + occurredAt: now, + payload: { ...request, responseCapability: { type: "message" } }, + }); + } for (const request of requests) { events.push({ id: yield* allocateEventId(), @@ -460,7 +533,6 @@ export const make = Effect.gen(function* () { // A blocking delegate_task wait dies with the turn that made it, and its // continuation never got the tool result. Let the child's result wake // the parent like an async task, so it reaches the continued turn. - const cutRunIds = new Set(runs.map((run) => run.id)); for (const task of projection.subagents ?? []) { if ( task.origin !== "app_owned" || diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index b60758e80c72..ae10c4be133c 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -13,6 +13,7 @@ import { type Project, ProjectId, ProviderDriverKind, + NodeId, ProviderInstanceId, type ProviderSessionId, ThreadId, @@ -2573,6 +2574,87 @@ it.effect("ProviderSessionManagerV2 marks pending runtime requests non-live on r yield* effect.pipe(Effect.provide(layerTest({ state, idleTimeoutMs: 1000 }))); }), ); +it.effect.each([ + ["server shutdown", "pending"], + ["manual close", "cancelled"], +] as const)( + "ProviderSessionManagerV2 leaves Codex questions to recovery only on %s", + ([release, questionStatus]) => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + const effect = Effect.gen(function* () { + const eventSink = yield* EventSink.EventSinkV2; + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + const projectionStore = yield* ProjectionStore.ProjectionStoreV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make(`thread-provider-session-manager-question-${release}`); + const providerSessionId = yield* idAllocator.allocate.providerSession({ + providerInstanceId: modelSelection.instanceId, + threadId, + }); + const providerThread = makeProviderThread({ + idAllocator, + threadId, + providerSessionId, + now, + }); + yield* eventSink.write({ + events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })], + }); + const approval = yield* makePendingRuntimeRequestEvents({ + idAllocator, + threadId, + providerSessionId, + providerThread, + now, + }); + const questionId = yield* idAllocator.allocate.runtimeRequest({ + driver: CODEX_DRIVER, + nativeRequestId: "pending-question", + }); + yield* eventSink.write({ + events: [ + ...approval.events, + { + id: yield* idAllocator.allocate.event({ threadId, providerSessionId }), + type: "runtime-request.updated", + threadId, + nodeId: NodeId.make(`${threadId}:question`), + driver: CODEX_DRIVER, + occurredAt: now, + payload: { + id: questionId, + nodeId: NodeId.make(`${threadId}:question`), + providerTurnId: null, + nativeRequestRef: { + driver: CODEX_DRIVER, + nativeId: "pending-question", + strength: "strong", + }, + kind: "user_input", + status: "pending", + responseCapability: { type: "live", providerSessionId }, + createdAt: now, + resolvedAt: null, + }, + }, + ], + }); + yield* manager.open({ threadId, providerSessionId, modelSelection, runtimePolicy }); + yield* release === "server shutdown" ? manager.shutdown : manager.close(providerSessionId); + + const { runtimeRequests } = yield* projectionStore.getThreadProjection(threadId); + const status = (id: typeof questionId) => + runtimeRequests.find((request) => request.id === id)?.status; + assert.equal(status(questionId), questionStatus); + assert.equal(status(approval.requestId), "cancelled"); + }); + + yield* effect.pipe(Effect.provide(layerTest({ state, idleTimeoutMs: 60_000 }))); + }), +); + it.effect("ProviderSessionManagerV2 retries release records that failed to persist", () => Effect.gen(function* () { const state = yield* Ref.make(emptyState); diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index e82995d1db77..01b60b3f2605 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -642,6 +642,13 @@ export const layerWithOptions = ( ? "Provider session failed before this runtime request was resolved." : "Provider session was closed before this runtime request was resolved."; + // Shutdown recovery runs next and decides which of these questions can + // still be answered after the restart. + const leaveQuestionsToRecovery = + input.reason === "server_shutdown" && + input.entry.runtime.providerSession.capabilities.sessions + .questionsAnswerableAfterRestart === true; + const events: Array = []; for (const threadId of input.entry.attachedThreadIds) { const projection = yield* projectionStore.getThreadRecords( @@ -654,6 +661,7 @@ export const layerWithOptions = ( request.status === "pending" && request.responseCapability.type === "live" && request.responseCapability.providerSessionId === providerSessionId && + !(leaveQuestionsToRecovery && request.kind === "user_input") && DateTime.isLessThanOrEqualTo(request.createdAt, input.releasedAt), ); diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index 9db0564cced0..e372e24df32e 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -4576,6 +4576,244 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { }), ); + it.effect.each(["startup", "shutdown"] as const)( + "keeps a blocked Codex question answerable across %s and answers it once", + (trigger) => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const eventSink = yield* EventSink.EventSinkV2; + const outbox = yield* EffectOutbox.EffectOutboxV2; + const recovery = yield* ProviderRuntimeRecoveryService.ProviderRuntimeRecoveryService; + const threadId = ThreadId.make(`restart-question-${trigger}`); + const firstMessageId = MessageId.make(`${threadId}:message`); + yield* orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make(`${threadId}:create`), + threadId, + projectId: ProjectId.make(`${threadId}:project`), + title: "Question across restart", + modelSelection, + runtimeMode: "full-access", + interactionMode: "plan", + branch: null, + worktreePath: process.cwd(), + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make(`${threadId}:message`), + threadId, + messageId: firstMessageId, + text: "Ask me to choose Alpha or Beta.", + attachments: [], + dispatchMode: { type: "start_immediately" }, + }); + const initial = yield* orchestrator.getThreadProjection(threadId); + const run = initial.runs[0]!; + const now = yield* DateTime.now; + const providerThread = { + ...initial.providerThreads[0]!, + nativeThreadRef: { driver, nativeId: `${threadId}:native`, strength: "strong" as const }, + status: "active" as const, + updatedAt: now, + }; + const providerSessionId = providerThread.providerSessionId!; + const providerTurnId = ProviderTurnId.make(`${threadId}:turn`); + const requestId = RuntimeRequestId.make(`${threadId}:request`); + const nodeId = NodeId.make(`${threadId}:question-node`); + const itemId = TurnItemId.make(`${threadId}:question-item`); + yield* eventSink.write({ + commandId: CommandId.make(`${threadId}:waiting`), + events: [ + { + id: EventId.make(`${threadId}:run-running`), + type: "run.updated", + threadId, + runId: run.id, + occurredAt: now, + payload: { ...run, status: "running", startedAt: now }, + }, + { + id: EventId.make(`${threadId}:session`), + type: "provider-session.attached", + threadId, + occurredAt: now, + payload: { + id: providerSessionId, + driver, + providerInstanceId: modelSelection.instanceId, + status: "ready", + cwd: process.cwd(), + model: modelSelection.model, + capabilities: CodexProviderCapabilitiesV2, + createdAt: now, + updatedAt: now, + lastError: null, + }, + }, + { + id: EventId.make(`${threadId}:provider-thread`), + type: "provider-thread.updated", + threadId, + occurredAt: now, + payload: providerThread, + }, + { + id: EventId.make(`${threadId}:turn`), + type: "provider-turn.updated", + threadId, + runId: run.id, + occurredAt: now, + payload: { + id: providerTurnId, + providerThreadId: providerThread.id, + nodeId: run.rootNodeId!, + runAttemptId: run.activeAttemptId, + nativeTurnRef: { driver, nativeId: `${threadId}:native-turn`, strength: "strong" }, + ordinal: 1, + status: "running", + startedAt: now, + completedAt: null, + }, + }, + { + id: EventId.make(`${threadId}:question-node`), + type: "node.updated", + threadId, + runId: run.id, + nodeId, + occurredAt: now, + payload: { + id: nodeId, + threadId, + runId: run.id, + parentNodeId: run.rootNodeId, + rootNodeId: run.rootNodeId!, + kind: "user_input_request", + status: "waiting", + countsForRun: false, + providerThreadId: providerThread.id, + providerTurnId, + nativeItemRef: { driver, nativeId: "call_question", strength: "strong" }, + runtimeRequestId: requestId, + checkpointScopeId: null, + startedAt: now, + completedAt: null, + }, + }, + { + id: EventId.make(`${threadId}:question-request`), + type: "runtime-request.updated", + threadId, + runId: run.id, + nodeId, + occurredAt: now, + payload: { + id: requestId, + nodeId, + providerTurnId, + nativeRequestRef: { driver, nativeId: "call_question", strength: "strong" }, + kind: "user_input", + status: "pending", + responseCapability: { type: "live", providerSessionId }, + createdAt: now, + resolvedAt: null, + }, + }, + { + id: EventId.make(`${threadId}:question-item`), + type: "turn-item.updated", + threadId, + runId: run.id, + nodeId, + occurredAt: now, + payload: { + id: itemId, + type: "user_input_request", + threadId, + runId: run.id, + nodeId, + providerThreadId: providerThread.id, + providerTurnId, + nativeItemRef: { driver, nativeId: "call_question", strength: "strong" }, + parentItemId: null, + ordinal: 1, + status: "waiting", + title: null, + startedAt: now, + completedAt: null, + updatedAt: now, + requestId, + questions: [ + { id: "choice", header: "Choice", question: "Alpha or Beta?", options: [] }, + ], + }, + }, + ], + }); + + yield* recovery.prepareForShutdown; + yield* recovery.reconcile(trigger); + yield* recovery.reconcile("startup"); + const restarted = yield* orchestrator.getThreadProjection(threadId); + const pending = restarted.runtimeRequests.find((request) => request.id === requestId); + assert.equal(pending?.status, "pending"); + assert.deepEqual(pending?.responseCapability, { type: "message" }); + assert.equal(restarted.nodes.find((node) => node.id === nodeId)?.status, "waiting"); + assert.equal(restarted.turnItems.find((item) => item.id === itemId)?.status, "waiting"); + assert.deepEqual( + restarted.runs.map((candidate) => candidate.status), + ["cancelled"], + ); + assert.isTrue(Option.isNone(yield* outbox.get(`effect:restart-continuation:${run.id}`))); + assert.deepEqual( + restarted.messages.map((message) => message.id), + [firstMessageId], + ); + + const answer = (key: string, choice: string) => + orchestrator + .dispatch({ + type: "runtime-request.respond", + commandId: CommandId.make(`${threadId}:answer:${key}`), + threadId, + requestId, + answers: { choice }, + }) + .pipe(Effect.result); + const results = yield* Effect.all([answer("alpha", "Alpha"), answer("beta", "Beta")], { + concurrency: "unbounded", + }); + assert.equal(results.filter((result) => result._tag === "Success").length, 1); + const answered = yield* orchestrator.getThreadProjection(threadId); + const resolved = answered.runtimeRequests.find((request) => request.id === requestId); + assert.equal(resolved?.status, "resolved"); + const reply = answered.messages.find((message) => message.id !== firstMessageId); + assert.equal(reply?.text, `Alpha or Beta?\n${resolved?.answers?.choice}`); + assert.lengthOf(answered.messages, 2); + assert.lengthOf(answered.runs, 2); + const resumed = answered.runs.find((candidate) => candidate.userMessageId === reply?.id); + assert.equal(resumed?.providerThreadId, providerThread.id); + assert.deepEqual( + answered.providerThreads.find((candidate) => candidate.id === providerThread.id) + ?.nativeThreadRef, + providerThread.nativeThreadRef, + ); + + yield* recovery.reconcile("startup"); + const afterStaleRecovery = yield* orchestrator.getThreadProjection(threadId); + assert.deepEqual( + afterStaleRecovery.runtimeRequests.find((request) => request.id === requestId), + resolved, + ); + assert.lengthOf(afterStaleRecovery.messages, 2); + assert.lengthOf(afterStaleRecovery.runs, 2); + }), + ); + it.effect("edits and removes queued runs", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; diff --git a/docs/user/updating.md b/docs/user/updating.md index 706d8fad5e10..dfad8d7c7dcc 100644 --- a/docs/user/updating.md +++ b/docs/user/updating.md @@ -53,8 +53,10 @@ A thread that finished its turn to wait on a monitor, subagent, or background ta is woken and told that this work stopped. A background shell, such as a dev server, does not wake it; the thread learns about it with your next message. A thread is not resumed while it waits for your approval, answer, or secret, or -while it is archived, snoozed, or settled. After three automatic resumes in a row, -the thread waits for your next message. +while it is archived, snoozed, or settled. A Codex question stays open across the +restart, and answering it continues that conversation. Approvals, secrets, and +questions from other providers or subagents end with the restart and need a new message. +After three automatic resumes in a row, the thread waits for your next message. Updates from the previous orchestration system preserve conversation transcripts but cannot carry every kind of runtime history forward. Read [Threads from older T3 Code versions](./thread-migration.md) diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 881a316f59ae..87f12232da0d 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -213,6 +213,11 @@ export const OrchestrationV2SessionCapabilities = Schema.Struct({ supportsProviderSwitchingViaHandoff: Schema.Boolean, supportsRuntimeModeSwitchInSession: Schema.Boolean, pendingRequestsSurviveRestart: Schema.Boolean, + /** + * The provider resumes a native conversation from a follow-up message, so a question cut off by + * a server restart can still be answered as a reply. + */ + questionsAnswerableAfterRestart: Schema.optional(Schema.Boolean), }); export type OrchestrationV2SessionCapabilities = typeof OrchestrationV2SessionCapabilities.Type; From f3d95159fe14d732ae6b653bf78ba2ff0425e08e Mon Sep 17 00:00:00 2001 From: hahasagined <3049408+hahasagined@users.noreply.github.com> Date: Thu, 22 Oct 2026 14:15:16 +0000 Subject: [PATCH 2/2] fix(server): close cards of questions that restart recovery cancels On a graceful shutdown the session manager now leaves Codex questions to recovery. When recovery cancels one whose run it does not cut (the run is missing, or kept to replay its checkpoint), the question card and its node stayed waiting. Recovery now closes the card and node of every request it ends, unless the run cleanup already does. Questions kept answerable stay open. Co-Authored-By: Claude Opus 5.5 --- .../ProviderRuntimeRecoveryService.test.ts | 58 ++++++++++++++++--- .../ProviderRuntimeRecoveryService.ts | 42 ++++++++++++++ 2 files changed, 93 insertions(+), 7 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts index 80382409b1e5..fe25f5e7eab1 100644 --- a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts @@ -203,7 +203,10 @@ const questionProjection = () => { }; type QuestionProjection = ReturnType; -const recoveredQuestion = (projection: QuestionProjection) => { +const recoveryEvents = ( + projection: QuestionProjection, + trigger: "startup" | "shutdown" = "startup", +) => { let committed: Parameters[0] | undefined; const layer = ProviderRuntimeRecovery.layer.pipe( Layer.provide(ServerSettings.layerTest()), @@ -222,24 +225,43 @@ const recoveredQuestion = (projection: QuestionProjection) => { }), IdAllocator.layer, Layer.mock(EffectOutbox.EffectOutboxV2)({ + // A waiting run with a checkpoint still to capture is replayed, not cut. + listByCommandId: () => + Effect.succeed( + projection.runs.map((run) => ({ + request: { type: "checkpoint.capture", runId: run.id }, + status: "pending", + })) as never, + ), reconcileAfterProcessLoss: Effect.succeed({ requeued: 0, cancelled: 0 }), }), ), ), ); return Effect.gen(function* () { - yield* (yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService).reconcile("startup"); - return committed?.events.flatMap((event) => - event.type === "runtime-request.updated" ? [event.payload] : [], - ); + yield* (yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService).reconcile(trigger); + return committed?.events ?? []; }).pipe(Effect.provide(layer)); }; +const recoveredQuestion = (projection: QuestionProjection) => + recoveryEvents(projection).pipe( + Effect.map((events) => + events.flatMap((event) => (event.type === "runtime-request.updated" ? [event.payload] : [])), + ), + ); + it.effect("keeps a root Codex question answerable as a reply after restart", () => Effect.gen(function* () { - const [request] = (yield* recoveredQuestion(questionProjection())) ?? []; + const events = yield* recoveryEvents(questionProjection()); + const [request] = events.flatMap((event) => + event.type === "runtime-request.updated" ? [event.payload] : [], + ); assert.equal(request?.status, "pending"); assert.deepEqual(request?.responseCapability, { type: "message" }); + assert.isFalse( + events.some((event) => event.type === "turn-item.updated" || event.type === "node.updated"), + ); }), ); @@ -299,12 +321,34 @@ it.effect.each<[string, (projection: QuestionProjection) => QuestionProjection]> ], ])("expires the question of %s after restart", ([, change]) => Effect.gen(function* () { - const [request] = (yield* recoveredQuestion(change(questionProjection()))) ?? []; + const [request] = yield* recoveredQuestion(change(questionProjection())); assert.equal(request?.status, "expired"); assert.equal(request?.responseCapability.type, "not_resumable"); }), ); +it.effect.each<[string, (projection: QuestionProjection) => QuestionProjection]>([ + ["whose run is missing", (projection) => ({ ...projection, runs: [] })], + [ + "whose run is kept for a replayable checkpoint", + (projection) => ({ + ...projection, + runs: projection.runs.map((run) => ({ ...run, status: "waiting" })), + }), + ], +])("closes the card of a question %s when shutdown cancels it", ([, change]) => + Effect.gen(function* () { + const events = yield* recoveryEvents(change(questionProjection()), "shutdown"); + const request = events.find((event) => event.type === "runtime-request.updated"); + assert.equal(request?.payload.status, "cancelled"); + const item = events.find((event) => event.type === "turn-item.updated"); + assert.equal(item?.payload.status, "cancelled"); + const node = events.find((event) => event.type === "node.updated"); + assert.equal(node?.payload.status, "cancelled"); + assert.isFalse(events.some((event) => event.type === "run.updated")); + }), +); + it.effect( "preserves a replayable waiting run while cancelling its process-bound background work", () => { diff --git a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts index 9b0ad01baa34..cf995cac33ab 100644 --- a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts +++ b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts @@ -4,6 +4,7 @@ import { CommandId, type OrchestrationV2DomainEvent, type ProviderThreadId, + type RunId, type OrchestrationV2RestartCancelledBackgroundWork, type OrchestrationV2Subagent, type OrchestrationV2ThreadProjection, @@ -555,6 +556,47 @@ export const make = Effect.gen(function* () { }); } const recoveredNonterminalRunIds = new Set(runs.map((run) => run.id)); + // A request ended here whose run was not cut (missing, or kept to replay + // its checkpoint) would otherwise leave its card waiting. + const isLeftOpen = (status: string, runId: RunId | null) => + isNonterminalTurnItemStatus(status) && + (runId === null || !recoveredNonterminalRunIds.has(runId)); + for (const request of requests) { + const item = (projection.turnItems ?? []).find( + (candidate) => + (candidate.type === "approval_request" || candidate.type === "user_input_request") && + candidate.requestId === request.id, + ); + const providerInstanceId = + item === undefined + ? projection.thread.providerInstanceId + : resolveStaleBackgroundItemProviderInstanceId(item, projection); + if (item !== undefined && isLeftOpen(item.status, item.runId)) { + events.push({ + id: yield* allocateEventId(), + type: "turn-item.updated", + threadId: projection.thread.id, + ...(item.runId === null ? {} : { runId: item.runId }), + ...(item.nodeId === null ? {} : { nodeId: item.nodeId }), + providerInstanceId, + occurredAt: now, + payload: cancelledItem(item, now), + }); + } + const node = projection.nodes.find((candidate) => candidate.id === request.nodeId); + if (node !== undefined && isLeftOpen(node.status, node.runId)) { + events.push({ + id: yield* allocateEventId(), + type: "node.updated", + threadId: projection.thread.id, + ...(node.runId === null ? {} : { runId: node.runId }), + nodeId: node.id, + providerInstanceId, + occurredAt: now, + payload: { ...node, status: "cancelled", completedAt: now }, + }); + } + } const cancelledStaleNodeIds = new Set(); for (const item of projection.turnItems ?? []) { if (item.runId !== null && recoveredNonterminalRunIds.has(item.runId)) {