diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 0b9d8de53331..8400b00b39fa 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -240,6 +240,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..fe25f5e7eab1 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,222 @@ 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 recoveryEvents = ( + projection: QuestionProjection, + trigger: "startup" | "shutdown" = "startup", +) => { + 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)({ + // 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(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 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"), + ); + }), +); + +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.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 f84346192569..59110bf75326 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, @@ -151,6 +152,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], + interruptedRunIds: 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 || !interruptedRunIds.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, @@ -224,16 +280,24 @@ export const make = Effect.gen(function* () { } runs.push(run); } + const interruptedRunIds = new Set(runs.map((run) => run.id)); + const answerableRequests = projection.runtimeRequests.filter((request) => + answerableAfterRestart(projection, request, interruptedRunIds), + ); 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([ @@ -330,6 +394,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(), @@ -475,10 +549,49 @@ export const make = Effect.gen(function* () { payload: { ...task, completionWake: "always", updatedAt: now }, }); } - 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 || !cutRunIds.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)) { + if (item.runId !== null && cutRunIds.has(item.runId)) { continue; } if (!isBackgroundCapableTurnItemType(item.type)) { diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index a64997a8dd7b..e33c4603b933 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, @@ -2560,6 +2561,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 a9daa1995bc5..bd3b743c0a30 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -614,6 +614,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( @@ -626,6 +633,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 31e8c04f30f5..625c1550097b 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -5066,6 +5066,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 cb36732b5747..d4891057e05f 100644 --- a/docs/user/updating.md +++ b/docs/user/updating.md @@ -54,8 +54,10 @@ command or monitor the agent said it was waiting on is woken and told that this stopped. Other background shells and monitors, such as a dev server and its watchdog, do not wake it; the thread learns about them 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 8aa4788a3213..95c84a7f6ad4 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -216,6 +216,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;