From fbed573c8b8bc58df264ccb86437276db32013e4 Mon Sep 17 00:00:00 2001 From: Abhishek Date: Thu, 29 Oct 2026 18:16:53 +0530 Subject: [PATCH 1/6] fix(server): Codex questions the agent waits on put the thread in waiting for input Codex asks with request_user_input_async and then idles in the interruptible clock.sleep tool until the answer arrives. T3 Code keeps async questions non-blocking while their run is live, so these threads stayed "Working" everywhere and no waiting-for-input push was sent. The Codex adapter now marks the turn's async questions `agentWaiting` while a sleep item runs and clears it when the sleep ends, and the thread shell treats an awaited question as blocking. The event sink drops a provider's pending repeat of a question that is already answered, so the sleep ending after an answer cannot reopen it. Co-Authored-By: Claude Opus 5.5 --- .../Adapters/CodexAdapterV2.test.ts | 27 ++++++++++++--- .../Adapters/CodexAdapterV2.ts | 34 +++++++++++++++++++ apps/server/src/orchestration-v2/EventSink.ts | 16 +++++++-- .../orchestration-v2/ProjectionStore.test.ts | 5 ++- .../src/orchestration-v2/ProjectionStore.ts | 4 ++- .../src/orchestration-v2/runtimeLayer.test.ts | 26 ++++++++++++++ docs/internals/providers.md | 4 +++ docs/user/providers-codex.md | 3 +- packages/contracts/src/orchestrationV2.ts | 6 +++- 9 files changed, 114 insertions(+), 11 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 69fca8c0825b8..13fe56207da3f 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -1535,7 +1535,7 @@ describe("CodexAdapterV2 post-settle continuation", () => { event.type === "message.updated" && event.message.role === "assistant", ); - it.effect("keeps an asynchronous Codex question actionable after the turn completes", () => + it.effect("marks an asynchronous Codex question awaited while the agent sleeps on it", () => Effect.scoped( Effect.gen(function* () { const nativeThreadId = "async-question-thread"; @@ -1573,6 +1573,20 @@ describe("CodexAdapterV2 post-settle continuation", () => { }, }, }, + ...(["item/started", "item/completed"] as const).map((method) => ({ + type: "emit_inbound" as const, + label: `sleep ${method}`, + frame: { + method, + params: { + threadId: nativeThreadId, + turnId: nativeTurnId, + startedAtMs: 1782622441000, + completedAtMs: 1782622501000, + item: { type: "sleep", id: "sleep-item", durationMs: 60_000 }, + }, + }, + })), { type: "emit_inbound", label: "usage", @@ -1612,9 +1626,14 @@ describe("CodexAdapterV2 post-settle continuation", () => { const requests = harness.events.flatMap((event) => event.type === "runtime_request.updated" ? [event.runtimeRequest] : [], ); - assert.lengthOf(requests, 1); - assert.equal(requests[0]?.status, "pending"); - assert.deepEqual(requests[0]?.responseCapability, { type: "message" }); + assert.deepEqual( + requests.map((request) => [request.status, request.responseCapability]), + [ + ["pending", { type: "message" }], + ["pending", { type: "message", agentWaiting: true }], + ["pending", { type: "message" }], + ], + ); const questionItem = harness.events.find( (event) => event.type === "turn_item.updated" && event.turnItem.type === "user_input_request", diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 58cd4a0c0c17b..0aa9ac44eefec 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -1131,6 +1131,7 @@ interface ActiveCodexTurnContext { readonly runningCompactionItemIds: Set; // Modern approval requests omit paths; keep the kinds without retaining patch bodies. readonly fileChangeKindsByItem: Map; + readonly asyncQuestions: Map; readonly itemPositions: Map< string, { readonly ordinal: number; readonly startedAt: DateTime.Utc } @@ -1961,6 +1962,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( startedAt: input.startedAt, itemPositions: new Map(), fileChangeKindsByItem: new Map(), + asyncQuestions: new Map(), runningCompactionItemIds: new Set(), }; const nativeThreadId = input.turnInput.providerThread.nativeThreadRef?.nativeId; @@ -2560,6 +2562,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( startedAt: turn.startedAt, itemPositions: new Map(), fileChangeKindsByItem: new Map(), + asyncQuestions: new Map(), runningCompactionItemIds: new Set(), }; beginTurnTokenUsage(activeContext); @@ -4639,6 +4642,25 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( }).pipe(Effect.orDie, turnTerminalizationPermit.withPermits(1)), ); + // Codex agents wait on an async question's answer with the interruptible clock.sleep tool. + const markAsyncQuestionsAwaited = Effect.fn("CodexAdapterV2.markAsyncQuestionsAwaited")( + function* (context: ActiveCodexTurnContext, agentWaiting: boolean) { + for (const request of context.asyncQuestions.values()) { + yield* emitProviderEvent({ + type: "runtime_request.updated", + driver: CODEX_PROVIDER, + threadId: context.projectionThreadId, + runtimeRequest: { + ...request, + responseCapability: agentWaiting + ? { type: "message", agentWaiting } + : { type: "message" }, + }, + }); + } + }, + ); + yield* client.handleServerNotification("item/started", (payload) => Effect.gen(function* () { const context = yield* awaitActiveTurn(payload.turnId); @@ -4675,6 +4697,10 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( } yield* completeProviderRetry(context, yield* DateTime.now); + if (payload.item.type === "sleep") { + yield* markAsyncQuestionsAwaited(context, true); + return; + } if (payload.item.type !== "reasoning") { yield* resolveItemPosition( context, @@ -4800,6 +4826,11 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( } const { context, settled } = resolved; + if (payload.item.type === "sleep") { + yield* markAsyncQuestionsAwaited(context, false); + return; + } + if (payload.item.type === "reasoning") { yield* completeReasoning(payload.turnId, payload.item); return; @@ -5015,6 +5046,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( threadId: artifacts.node.threadId, runtimeRequest: artifacts.request, }); + context.asyncQuestions.set(String(artifacts.request.id), artifacts.request); yield* Ref.update(finalAnswerItemIdsByTurn, (current) => { const ids = current.get(payload.turnId); if (!ids?.has(payload.item.id)) return current; @@ -5853,6 +5885,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( return false; } input.context.fileChangeKindsByItem.clear(); + input.context.asyncQuestions.clear(); const providerRetry = yield* Ref.modify(providerRetries, (current) => { const retry = current.get(input.context.providerTurnId); if (retry === undefined) { @@ -6334,6 +6367,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( startedAt: now, itemPositions: new Map(), fileChangeKindsByItem: new Map(), + asyncQuestions: new Map(), runningCompactionItemIds: new Set(), }; const providerTurn = { diff --git a/apps/server/src/orchestration-v2/EventSink.ts b/apps/server/src/orchestration-v2/EventSink.ts index b210d3c9840ae..c73bf5a3dcfcd 100644 --- a/apps/server/src/orchestration-v2/EventSink.ts +++ b/apps/server/src/orchestration-v2/EventSink.ts @@ -284,23 +284,31 @@ const layerBase: Layer.Layer< }); // A user can answer after terminal normalization reads the pending request. - // Recheck inside the write transaction so stale cleanup cannot erase answers. + // Recheck inside the write transaction so stale cleanup cannot erase answers, + // and so a provider repeating a pending question cannot reopen it. const guardUserInputCancellations = /* This helper runs for individual orchestration events; the enclosing operation owns the span. */ Effect.fnUntraced(function* (events: ReadonlyArray) { const staleRequests = new Set(); const staleNodes = new Set(); + const settledRequests = new Set(); for (const event of events) { if ( event.type !== "runtime-request.updated" || event.payload.kind !== "user_input" || - event.payload.status !== "cancelled" + (event.payload.status !== "cancelled" && event.payload.status !== "pending") ) continue; const current = yield* projectionStore.getRuntimeRequest( event.threadId, event.payload.id, ); + if (event.payload.status === "pending") { + if (current !== undefined && current.status !== "pending") { + settledRequests.add(event.payload.id); + } + continue; + } if ( current?.status !== "pending" || current.kind !== "user_input" || @@ -314,7 +322,9 @@ const layerBase: Layer.Layer< return events.filter((event) => { switch (event.type) { case "runtime-request.updated": - return event.payload.status !== "cancelled" || !staleRequests.has(event.payload.id); + return event.payload.status === "pending" + ? !settledRequests.has(event.payload.id) + : event.payload.status !== "cancelled" || !staleRequests.has(event.payload.id); case "node.updated": return event.payload.status !== "cancelled" || !staleNodes.has(event.payload.id); case "turn-item.updated": diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 2af87ecf4feb7..acfd6f191b362 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -3709,10 +3709,13 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { const requestId = RuntimeRequestId.make("request:projection-shell-question"); for (const [responseCapability, blocking] of [ [{ type: "message" }, false], + [{ type: "message", agentWaiting: true }, true], [{ type: "live", providerSessionId: ProviderSessionId.make("session:question") }, true], ] as const) { yield* projectionStore.apply({ - id: EventId.make(`event:projection-shell-question:${responseCapability.type}`), + id: EventId.make( + `event:projection-shell-question:${responseCapability.type}:${blocking}`, + ), type: "runtime-request.updated", threadId, nodeId: rootNodeId, diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 2a21bfae4bd32..8949316d7d615 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -1358,7 +1358,9 @@ function pendingRuntimeRequestSummary( id: request.id, kind: request.kind, createdAt: request.createdAt, - blocking: request.responseCapability.type !== "message", + blocking: + request.responseCapability.type !== "message" || + request.responseCapability.agentWaiting === true, }; } diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index 65c0c040e0daf..80d36ee82d188 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -1450,6 +1450,32 @@ it.layer(layerTest)("OrchestrationV2LayerLive", (it) => { .pipe(Effect.result); assert.equal(duplicate._tag, "Failure"); assert.equal((yield* orchestrator.getThreadProjection(threadId)).messages.length, 1); + + // Codex repeats the question as pending when the sleep the answer interrupted ends. + const answeredRequest = answered.runtimeRequests[0]!; + yield* eventSink.write({ + guardPendingUserInputCancellations: true, + commandId: CommandId.make("runtime-async-question-repeat"), + events: [ + { + id: EventId.make("runtime-async-question-repeat"), + type: "runtime-request.updated", + threadId, + nodeId, + occurredAt: yield* DateTime.now, + payload: { + ...answeredRequest, + status: "pending", + resolvedAt: null, + responseCapability: { type: "message" }, + }, + }, + ], + }); + assert.deepEqual( + (yield* orchestrator.getThreadProjection(threadId)).runtimeRequests, + answered.runtimeRequests, + ); }), ); diff --git a/docs/internals/providers.md b/docs/internals/providers.md index 42f0a70101b3a..58527bf393968 100644 --- a/docs/internals/providers.md +++ b/docs/internals/providers.md @@ -109,6 +109,10 @@ no pending RPC response to send. The `user_input_request` turn items and runtime requests with `responseCapability: { type: "message" }`. Their execution nodes do not block the run. Web, desktop, and mobile use their normal question panels, and requests remain pending after a turn finishes, a provider exits, or the server restarts. +The agent waits for an answer by calling the interruptible `clock.sleep` tool, so the adapter sets +`agentWaiting` on the request while a sleep item runs; status surfaces then show the thread waiting +for input instead of Working. The event sink drops the request's later `pending` repeats once it +is answered. `runtime-request.respond` reads the persisted request and question item, validates required answers, and commits the resolution and a user message in one transaction. Repeating the same diff --git a/docs/user/providers-codex.md b/docs/user/providers-codex.md index bdc87902a63b2..c32152994ed7c 100644 --- a/docs/user/providers-codex.md +++ b/docs/user/providers-codex.md @@ -79,7 +79,8 @@ contains a copied Codex setup. Use a fresh shadow directory and sign in again. ## Answer questions while Codex works Codex can ask a question and keep working. Answer it in the thread's question -panel. The answer becomes a new message: it reaches the active turn, or starts +panel. While Codex stops to wait for your answer, the thread shows that it needs +input and you get the usual notification. The answer becomes a new message: it reaches the active turn, or starts another turn if Codex has finished. Unanswered questions survive reconnects. If you do not want to answer, dismiss the question from its panel. Dismissing closes it without sending anything to Codex. This requires a Codex version that diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 747d05e7ce4eb..3389938f389cb 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -1190,7 +1190,11 @@ export const OrchestrationV2RuntimeRequest = Schema.Struct({ providerSessionId: ProviderSessionId, supportsFeedback: Schema.optional(Schema.Boolean), }), - Schema.Struct({ type: Schema.Literal("message") }), + Schema.Struct({ + type: Schema.Literal("message"), + // The agent has stopped working to wait for this answer. + agentWaiting: Schema.optional(Schema.Boolean), + }), Schema.Struct({ type: Schema.Literal("not_resumable"), reason: Schema.String }), ]), createdAt: Schema.DateTimeUtc, From 47acb8b7eb9dd2a21699153959dd9dac9cc7dd8f Mon Sep 17 00:00:00 2001 From: Abhishek Date: Thu, 29 Oct 2026 22:02:47 +0530 Subject: [PATCH 2/6] fix(server): clear a Codex question's waiting flag when the turn ends mid-sleep Stop or a failure can finalize a Codex turn before its sleep item completes. The adapter then dropped its async questions without clearing `agentWaiting`, so the unanswered question stayed blocking and a later turn would show Input while the agent worked. Turn finalization now resets any awaited question, and the reset skips questions that are already in the requested state. Co-Authored-By: Claude Opus 5.5 --- .../Adapters/CodexAdapterV2.test.ts | 295 +++++++++--------- .../Adapters/CodexAdapterV2.ts | 23 +- 2 files changed, 167 insertions(+), 151 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 13fe56207da3f..9b73713a1a2fb 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -1535,164 +1535,171 @@ describe("CodexAdapterV2 post-settle continuation", () => { event.type === "message.updated" && event.message.role === "assistant", ); - it.effect("marks an asynchronous Codex question awaited while the agent sleeps on it", () => - Effect.scoped( - Effect.gen(function* () { - const nativeThreadId = "async-question-thread"; - const nativeTurnId = "async-question-turn"; - const usage = { - totalTokens: 15, - inputTokens: 10, - cachedInputTokens: 2, - outputTokens: 5, - reasoningOutputTokens: 1, - }; - const transcript = makeCodexReplayTranscript({ - scenario: "async-question-and-billed-usage", - entries: [ - ...codexReplayPreamble({ - nativeThreadId, - nativeTurnId, - prompt: "Continue while I decide.", - }), - { - type: "emit_inbound", - label: "question", - frame: { - method: "item/completed", - params: { - threadId: nativeThreadId, - turnId: nativeTurnId, - item: { - type: "agentMessage", - id: "async-question-item", - text: "Which branch?", - delivery: "async", - questions: [{ title: "Which branch?", options: ["main", "dev"] }], + it.effect.each(["completed", "interrupted"] as const)( + "marks an asynchronous Codex question awaited until its sleep ends or the turn is %s", + (terminalStatus) => + Effect.scoped( + Effect.gen(function* () { + const nativeThreadId = "async-question-thread"; + const nativeTurnId = "async-question-turn"; + const usage = { + totalTokens: 15, + inputTokens: 10, + cachedInputTokens: 2, + outputTokens: 5, + reasoningOutputTokens: 1, + }; + const transcript = makeCodexReplayTranscript({ + scenario: "async-question-and-billed-usage", + entries: [ + ...codexReplayPreamble({ + nativeThreadId, + nativeTurnId, + prompt: "Continue while I decide.", + }), + { + type: "emit_inbound", + label: "question", + frame: { + method: "item/completed", + params: { + threadId: nativeThreadId, + turnId: nativeTurnId, + item: { + type: "agentMessage", + id: "async-question-item", + text: "Which branch?", + delivery: "async", + questions: [{ title: "Which branch?", options: ["main", "dev"] }], + }, }, }, }, - }, - ...(["item/started", "item/completed"] as const).map((method) => ({ - type: "emit_inbound" as const, - label: `sleep ${method}`, - frame: { - method, - params: { - threadId: nativeThreadId, - turnId: nativeTurnId, - startedAtMs: 1782622441000, - completedAtMs: 1782622501000, - item: { type: "sleep", id: "sleep-item", durationMs: 60_000 }, + // An interrupted turn ends without completing the sleep item. + ...(terminalStatus === "completed" + ? (["item/started", "item/completed"] as const) + : (["item/started"] as const) + ).map((method) => ({ + type: "emit_inbound" as const, + label: `sleep ${method}`, + frame: { + method, + params: { + threadId: nativeThreadId, + turnId: nativeTurnId, + startedAtMs: 1782622441000, + completedAtMs: 1782622501000, + item: { type: "sleep", id: "sleep-item", durationMs: 60_000 }, + }, }, - }, - })), - { - type: "emit_inbound", - label: "usage", - frame: { - method: "thread/tokenUsage/updated", - params: { - threadId: nativeThreadId, - turnId: nativeTurnId, - tokenUsage: { total: usage, last: usage, modelContextWindow: 200_000 }, + })), + { + type: "emit_inbound", + label: "usage", + frame: { + method: "thread/tokenUsage/updated", + params: { + threadId: nativeThreadId, + turnId: nativeTurnId, + tokenUsage: { total: usage, last: usage, modelContextWindow: 200_000 }, + }, }, }, - }, - { - type: "emit_inbound", - label: "complete", - frame: { - method: "turn/completed", - params: { - threadId: nativeThreadId, - turn: makeCodexReplayTurn({ id: nativeTurnId, status: "completed" }), + { + type: "emit_inbound", + label: "complete", + frame: { + method: "turn/completed", + params: { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ id: nativeTurnId, status: terminalStatus }), + }, }, }, - }, - ], - }); - const harness = yield* makeCodexReplayHarness(transcript); - yield* harness.runtime.startTurn( - makeCodexTestTurnInput({ - threadId: harness.threadId, - providerThread: harness.providerThread, - now: yield* DateTime.now, - attemptId: RunAttemptId.make("async-question-attempt"), - text: "Continue while I decide.", - }), - ); - yield* harness.firstTerminal; - const requests = harness.events.flatMap((event) => - event.type === "runtime_request.updated" ? [event.runtimeRequest] : [], - ); - assert.deepEqual( - requests.map((request) => [request.status, request.responseCapability]), - [ - ["pending", { type: "message" }], - ["pending", { type: "message", agentWaiting: true }], - ["pending", { type: "message" }], - ], - ); - const questionItem = harness.events.find( - (event) => - event.type === "turn_item.updated" && event.turnItem.type === "user_input_request", - ); - assert.equal(questionItem?.type, "turn_item.updated"); - if ( - questionItem?.type === "turn_item.updated" && - questionItem.turnItem.type === "user_input_request" - ) { + ], + }); + const harness = yield* makeCodexReplayHarness(transcript); + yield* harness.runtime.startTurn( + makeCodexTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("async-question-attempt"), + text: "Continue while I decide.", + }), + ); + yield* harness.firstTerminal; + const requests = harness.events.flatMap((event) => + event.type === "runtime_request.updated" ? [event.runtimeRequest] : [], + ); assert.deepEqual( - questionItem.turnItem.questions[0]?.options.map((option) => option.label), - ["main", "dev"], + requests.map((request) => [request.status, request.responseCapability]), + [ + ["pending", { type: "message" }], + ["pending", { type: "message", agentWaiting: true }], + ["pending", { type: "message" }], + ], + ); + const questionItem = harness.events.find( + (event) => + event.type === "turn_item.updated" && event.turnItem.type === "user_input_request", + ); + assert.equal(questionItem?.type, "turn_item.updated"); + if ( + questionItem?.type === "turn_item.updated" && + questionItem.turnItem.type === "user_input_request" + ) { + assert.deepEqual( + questionItem.turnItem.questions[0]?.options.map((option) => option.label), + ["main", "dev"], + ); + assert.equal(questionItem.turnItem.responseMode, "message"); + } + const questionNode = harness.events.find( + (event) => event.type === "node.updated" && event.node.id === requests[0]?.nodeId, ); - assert.equal(questionItem.turnItem.responseMode, "message"); - } - const questionNode = harness.events.find( - (event) => event.type === "node.updated" && event.node.id === requests[0]?.nodeId, - ); - assert.equal( - questionNode?.type === "node.updated" && questionNode.node.countsForRun, - false, - ); - const contextReport = harness.events.find( - (event) => - event.type === "provider_turn.updated" && event.providerTurn.tokenUsage !== undefined, - ); - assert.equal(contextReport?.type, "provider_turn.updated"); - if (contextReport?.type === "provider_turn.updated") { assert.equal( - contextReport.providerTurn.runAttemptId, - RunAttemptId.make("async-question-attempt"), + questionNode?.type === "node.updated" && questionNode.node.countsForRun, + false, ); - assert.equal(contextReport.providerTurn.providerThreadId, harness.providerThread.id); - assert.equal(contextReport.providerTurn.tokenUsage?.usedTokens, 15); - assert.equal(contextReport.providerTurn.tokenUsage?.maxTokens, 200_000); - } - const completed = harness.events.find( - (event) => - event.type === "provider_turn.updated" && event.providerTurn.status === "completed", - ); - assert.equal(completed?.type, "provider_turn.updated"); - if (completed?.type === "provider_turn.updated") { - assert.deepEqual(completed.providerTurn.turnTokenUsage, { - usageStatus: "complete", - usageScope: "main_agent", - hasSubagents: false, - inputTokens: 10, - cachedInputTokens: 2, - outputTokens: 5, - reasoningTokens: 1, - }); - } - assert.isEmpty(assistantMessages(harness.events)); - }).pipe( - Effect.provide( - Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + const contextReport = harness.events.find( + (event) => + event.type === "provider_turn.updated" && event.providerTurn.tokenUsage !== undefined, + ); + assert.equal(contextReport?.type, "provider_turn.updated"); + if (contextReport?.type === "provider_turn.updated") { + assert.equal( + contextReport.providerTurn.runAttemptId, + RunAttemptId.make("async-question-attempt"), + ); + assert.equal(contextReport.providerTurn.providerThreadId, harness.providerThread.id); + assert.equal(contextReport.providerTurn.tokenUsage?.usedTokens, 15); + assert.equal(contextReport.providerTurn.tokenUsage?.maxTokens, 200_000); + } + const completed = harness.events.find( + (event) => + event.type === "provider_turn.updated" && + event.providerTurn.status === terminalStatus, + ); + assert.equal(completed?.type, "provider_turn.updated"); + if (completed?.type === "provider_turn.updated") { + assert.deepEqual(completed.providerTurn.turnTokenUsage, { + usageStatus: terminalStatus === "completed" ? "complete" : "partial", + usageScope: "main_agent", + hasSubagents: false, + inputTokens: 10, + cachedInputTokens: 2, + outputTokens: 5, + reasoningTokens: 1, + }); + } + assert.isEmpty(assistantMessages(harness.events)); + }).pipe( + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), ), ), - ), ); it.effect("stops asking for commands the user allowed for the session", () => diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 0aa9ac44eefec..49ac06611cf29 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -4645,17 +4645,24 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( // Codex agents wait on an async question's answer with the interruptible clock.sleep tool. const markAsyncQuestionsAwaited = Effect.fn("CodexAdapterV2.markAsyncQuestionsAwaited")( function* (context: ActiveCodexTurnContext, agentWaiting: boolean) { - for (const request of context.asyncQuestions.values()) { + for (const [id, request] of context.asyncQuestions) { + if ( + request.responseCapability.type === "message" && + (request.responseCapability.agentWaiting === true) === agentWaiting + ) + continue; + const runtimeRequest: OrchestrationV2RuntimeRequest = { + ...request, + responseCapability: agentWaiting + ? { type: "message", agentWaiting } + : { type: "message" }, + }; + context.asyncQuestions.set(id, runtimeRequest); yield* emitProviderEvent({ type: "runtime_request.updated", driver: CODEX_PROVIDER, threadId: context.projectionThreadId, - runtimeRequest: { - ...request, - responseCapability: agentWaiting - ? { type: "message", agentWaiting } - : { type: "message" }, - }, + runtimeRequest, }); } }, @@ -5885,6 +5892,8 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( return false; } input.context.fileChangeKindsByItem.clear(); + // Stop or failure can end the turn before the sleep item completes. + yield* markAsyncQuestionsAwaited(input.context, false); input.context.asyncQuestions.clear(); const providerRetry = yield* Ref.modify(providerRetries, (current) => { const retry = current.get(input.context.providerTurnId); From 6c0d0580047642b560e912372c618b90e5e140b0 Mon Sep 17 00:00:00 2001 From: Abhishek Date: Thu, 29 Oct 2026 22:27:20 +0530 Subject: [PATCH 3/6] fix(server): server recovery clears a Codex question's waiting flag A server restart or shutdown while Codex sleeps on an async question loses the adapter that would clear `agentWaiting`. Recovery keeps message questions pending, so the question stayed blocking and the thread showed Input while a restart continuation or the next turn worked. Recovery now clears the flag and leaves the question pending and answerable. Co-Authored-By: Claude Opus 5.5 --- .../ProviderRuntimeRecoveryService.test.ts | 27 ++++++++++++++++--- .../ProviderRuntimeRecoveryService.ts | 8 +++++- docs/internals/providers.md | 5 ++-- 3 files changed, 34 insertions(+), 6 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts index 2fd9b13a35c33..314a4dc61fed1 100644 --- a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts @@ -85,6 +85,7 @@ it.effect("expires orphaned runtime requests before command readiness", () => { it.effect("preserves async questions across startup and shutdown", () => { const threadId = ThreadId.make("async-recovery-thread"); const nodeId = NodeId.make("async-recovery-node"); + const awaitedRequestId = RuntimeRequestId.make("async-recovery-awaited-request"); const projection = { thread: { id: threadId }, runtimeRequests: [ @@ -94,6 +95,12 @@ it.effect("preserves async questions across startup and shutdown", () => { status: "pending", responseCapability: { type: "message" }, }, + { + id: awaitedRequestId, + nodeId, + status: "pending", + responseCapability: { type: "message", agentWaiting: true }, + }, ], providerSessions: [], providerThreads: [], @@ -101,7 +108,10 @@ it.effect("preserves async questions across startup and shutdown", () => { nodes: [], turnItems: [], } as unknown as OrchestrationV2ThreadProjection; - const commitCommand = vi.fn(() => Effect.die("an async question needs no process-loss write")); + const commitCommand = vi.fn( + (_input: Parameters[0]) => + Effect.succeed({ committed: true, cancelledEffectCount: 0 } as never), + ); const layer = ProviderRuntimeRecovery.layer.pipe( Layer.provide(ServerSettings.layerTest()), Layer.provide( @@ -126,8 +136,19 @@ it.effect("preserves async questions across startup and shutdown", () => { return Effect.gen(function* () { const recovery = yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService; assert.equal((yield* recovery.reconcile("startup")).closedRequests, 0); - assert.equal((yield* recovery.reconcile("startup")).closedRequests, 0); - assert.isFalse(commitCommand.mock.calls.length > 0); + assert.equal((yield* recovery.reconcile("shutdown")).closedRequests, 0); + // The agent that was waiting on a question is gone, so it no longer blocks. + for (const [input] of commitCommand.mock.calls) { + assert.deepEqual( + input.events.map((event) => + event.type === "runtime-request.updated" + ? [event.payload.id, event.payload.status, event.payload.responseCapability] + : event.type, + ), + [[awaitedRequestId, "pending", { type: "message" }]], + ); + } + assert.lengthOf(commitCommand.mock.calls, 2); }).pipe(Effect.provide(layer)); }); diff --git a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts index c344357972268..cf5819f5a1fcc 100644 --- a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts +++ b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts @@ -248,6 +248,12 @@ export const make = Effect.gen(function* () { ) && !isProviderNativeSubagentThread(projection.thread), ); + // The agent that was waiting on these questions is gone; they stay answerable. + const awaitedQuestions = pendingRequests.filter( + (request) => + request.responseCapability.type === "message" && + request.responseCapability.agentWaiting === true, + ); const recoveredRequestIds = new Set(questions.map((request) => request.id)); const messageRequestNodeIds = new Set( pendingRequests @@ -356,7 +362,7 @@ export const make = Effect.gen(function* () { payload: { ...run, queueHeld: true }, }); } - for (const question of questions) { + for (const question of [...questions, ...awaitedQuestions]) { events.push({ id: yield* allocateEventId(), type: "runtime-request.updated", diff --git a/docs/internals/providers.md b/docs/internals/providers.md index 58527bf393968..a04d1c75a5e37 100644 --- a/docs/internals/providers.md +++ b/docs/internals/providers.md @@ -111,8 +111,9 @@ Their execution nodes do not block the run. Web, desktop, and mobile use their n panels, and requests remain pending after a turn finishes, a provider exits, or the server restarts. The agent waits for an answer by calling the interruptible `clock.sleep` tool, so the adapter sets `agentWaiting` on the request while a sleep item runs; status surfaces then show the thread waiting -for input instead of Working. The event sink drops the request's later `pending` repeats once it -is answered. +for input instead of Working. The turn ending and server recovery both clear the flag, since no +agent is left waiting. The event sink drops the request's later `pending` repeats once it is +answered. `runtime-request.respond` reads the persisted request and question item, validates required answers, and commits the resolution and a user message in one transaction. Repeating the same From 37702890b5aa17835d7508004b9ea96d590b6340 Mon Sep 17 00:00:00 2001 From: Abhishek Date: Thu, 29 Oct 2026 23:16:50 +0530 Subject: [PATCH 4/6] fix(server): recovery cannot reopen a question answered during shutdown A runtime-request.respond already past the shutdown gate can answer a question after recovery reads the thread. Recovery then committed its stale pending payload over the answer. commitCommand now accepts the in-transaction user-input guard the other event sink writes use, and recovery enables it. A commit whose events the guard drops entirely still records its receipt. The recovery test moves to the real event sink and projections, and answers a question between recovery's read and its commit. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/orchestration-v2/EventSink.ts | 14 ++- .../ProjectionRecovery.test.ts | 112 ++++++++++++++++++ .../ProviderRuntimeRecoveryService.test.ts | 27 +---- .../ProviderRuntimeRecoveryService.ts | 2 + 4 files changed, 127 insertions(+), 28 deletions(-) diff --git a/apps/server/src/orchestration-v2/EventSink.ts b/apps/server/src/orchestration-v2/EventSink.ts index c73bf5a3dcfcd..f56f02c279f68 100644 --- a/apps/server/src/orchestration-v2/EventSink.ts +++ b/apps/server/src/orchestration-v2/EventSink.ts @@ -148,6 +148,7 @@ export interface EventSinkV2Shape { }; /** Like writeWithEffects' updateDelegatedCompletionOnly, for this run's run.updated only. */ readonly updateDelegatedCompletionOnlyRunId?: RunId; + readonly guardPendingUserInputCancellations?: boolean; }) => Effect.Effect< { readonly receipt: CommandReceiptStore.CommandReceiptV2; @@ -662,19 +663,24 @@ const layerBase: Layer.Layer< } const runId = input.updateDelegatedCompletionOnlyRunId; - const normalized = yield* normalizeEvents( + const events = runId === undefined ? input.events - : yield* updateDelegatedCompletionOnly(input.events, (id) => id === runId), + : yield* updateDelegatedCompletionOnly(input.events, (id) => id === runId); + const normalized = yield* normalizeEvents( + input.guardPendingUserInputCancellations === true + ? yield* guardUserInputCancellations(events) + : events, ); const storedEvents = yield* eventStore.append({ commandId: input.commandId, events: normalized, }); - // A Stop that changes nothing can still queue the stop of its delegated tasks. + // A Stop that changes nothing can still queue the stop of its delegated tasks, + // and the user-input guard can drop every event. const sequence = storedEvents.at(-1)?.sequence ?? - (input.effects.length === 0 + (input.effects.length === 0 && input.guardPendingUserInputCancellations !== true ? undefined : yield* eventStore.latestSequence({ threadId: input.threadId })); if (sequence === undefined) { diff --git a/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts b/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts index 4df71fd5a9d63..d456cebc5db51 100644 --- a/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts @@ -52,6 +52,62 @@ const layerRecoveryTest = ProviderRuntimeRecovery.layer.pipe( ), Layer.provide(ServerSettings.layerTest()), ); +// Answers a question right after recovery reads its thread, like a respond already in flight at shutdown. +const layerAnswerDuringRecoveryTest = (answeredRequestId: RuntimeRequestId) => + ProviderRuntimeRecovery.layer.pipe( + Layer.provide( + Layer.effect( + ProjectionStore.ProjectionStoreV2, + Effect.gen(function* () { + const projections = yield* ProjectionStore.ProjectionStoreV2; + const eventSink = yield* EventSink.EventSinkV2; + return ProjectionStore.ProjectionStoreV2.of({ + ...projections, + getRuntimeRecoveryProjection: (threadId) => + projections.getRuntimeRecoveryProjection(threadId).pipe( + Effect.tap((projection) => + Effect.forEach( + projection.runtimeRequests.filter( + (request) => request.id === answeredRequestId, + ), + (request) => + eventSink.write({ + events: [ + { + id: EventId.make(`event:${request.id}:answered`), + type: "runtime-request.updated", + threadId, + nodeId: request.nodeId, + occurredAt: request.createdAt, + payload: { + ...request, + status: "resolved", + resolvedAt: request.createdAt, + }, + }, + ], + }), + ).pipe(Effect.orDie), + ), + ), + }); + }), + ), + ), + Layer.provideMerge( + EventSink.layer.pipe( + Layer.provideMerge( + Layer.mergeAll( + EventStore.layer, + ProjectionStore.layer, + EffectOutbox.layer, + IdAllocator.layer, + ).pipe(Layer.provideMerge(SqlitePersistence.layerMemory)), + ), + ), + ), + Layer.provide(ServerSettings.layerTest()), + ); const providerInstanceId = ProviderInstanceId.make("codex"); const modelSelection = { instanceId: providerInstanceId, model: "gpt-5.4" }; const driver = ProviderDriverKind.make("codex"); @@ -210,6 +266,62 @@ it.effect.each(["startup", "shutdown"] as const)( }).pipe(Effect.provide(layerRecoveryTest)), ); +it.effect( + "shutdown stops awaiting async questions without reopening one answered meanwhile", + () => { + const answeredRequestId = RuntimeRequestId.make("request:recovery:awaited-answered"); + return Effect.gen(function* () { + const projections = yield* ProjectionStore.ProjectionStoreV2; + const eventSink = yield* EventSink.EventSinkV2; + const recovery = yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService; + const now = yield* DateTime.now; + const threads = [] as Array; + for (const requestId of [ + RuntimeRequestId.make("request:recovery:awaited"), + answeredRequestId, + ]) { + const threadId = yield* createThread(requestId); + const nodeId = NodeId.make(`node:${requestId}`); + yield* eventSink.write({ + events: [ + { + id: EventId.make(`event:${requestId}`), + type: "runtime-request.updated", + threadId, + nodeId, + occurredAt: now, + payload: { + id: requestId, + nodeId, + providerTurnId: null, + nativeRequestRef: null, + kind: "user_input", + status: "pending", + responseCapability: { type: "message", agentWaiting: true }, + createdAt: now, + resolvedAt: null, + }, + }, + ], + }); + threads.push([threadId, requestId]); + } + yield* recovery.reconcile("shutdown"); + const requests = [] as Array; + for (const [threadId, requestId] of threads) { + const request = (yield* projections.getThreadProjection(threadId)).runtimeRequests.find( + (candidate) => candidate.id === requestId, + ); + requests.push([request?.status, request?.responseCapability]); + } + assert.deepEqual(requests, [ + ["pending", { type: "message" }], + ["resolved", { type: "message", agentWaiting: true }], + ]); + }).pipe(Effect.provide(layerAnswerDuringRecoveryTest(answeredRequestId))); + }, +); + it.effect("stops a session its thread detached, which kept its live status", () => Effect.gen(function* () { const projections = yield* ProjectionStore.ProjectionStoreV2; diff --git a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts index 314a4dc61fed1..2fd9b13a35c33 100644 --- a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts @@ -85,7 +85,6 @@ it.effect("expires orphaned runtime requests before command readiness", () => { it.effect("preserves async questions across startup and shutdown", () => { const threadId = ThreadId.make("async-recovery-thread"); const nodeId = NodeId.make("async-recovery-node"); - const awaitedRequestId = RuntimeRequestId.make("async-recovery-awaited-request"); const projection = { thread: { id: threadId }, runtimeRequests: [ @@ -95,12 +94,6 @@ it.effect("preserves async questions across startup and shutdown", () => { status: "pending", responseCapability: { type: "message" }, }, - { - id: awaitedRequestId, - nodeId, - status: "pending", - responseCapability: { type: "message", agentWaiting: true }, - }, ], providerSessions: [], providerThreads: [], @@ -108,10 +101,7 @@ it.effect("preserves async questions across startup and shutdown", () => { nodes: [], turnItems: [], } as unknown as OrchestrationV2ThreadProjection; - const commitCommand = vi.fn( - (_input: Parameters[0]) => - Effect.succeed({ committed: true, cancelledEffectCount: 0 } as never), - ); + const commitCommand = vi.fn(() => Effect.die("an async question needs no process-loss write")); const layer = ProviderRuntimeRecovery.layer.pipe( Layer.provide(ServerSettings.layerTest()), Layer.provide( @@ -136,19 +126,8 @@ it.effect("preserves async questions across startup and shutdown", () => { return Effect.gen(function* () { const recovery = yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService; assert.equal((yield* recovery.reconcile("startup")).closedRequests, 0); - assert.equal((yield* recovery.reconcile("shutdown")).closedRequests, 0); - // The agent that was waiting on a question is gone, so it no longer blocks. - for (const [input] of commitCommand.mock.calls) { - assert.deepEqual( - input.events.map((event) => - event.type === "runtime-request.updated" - ? [event.payload.id, event.payload.status, event.payload.responseCapability] - : event.type, - ), - [[awaitedRequestId, "pending", { type: "message" }]], - ); - } - assert.lengthOf(commitCommand.mock.calls, 2); + assert.equal((yield* recovery.reconcile("startup")).closedRequests, 0); + assert.isFalse(commitCommand.mock.calls.length > 0); }).pipe(Effect.provide(layer)); }); diff --git a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts index cf5819f5a1fcc..69a2d69be951a 100644 --- a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts +++ b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts @@ -704,6 +704,8 @@ export const make = Effect.gen(function* () { acceptedAt: now, events, effects, + // A respond already past the shutdown gate can answer after this read. + guardPendingUserInputCancellations: true, cancelUnsettledEffects: { effectTypes: EffectOutbox.PROCESS_BOUND_EFFECT_TYPES, reason: detail, From fdf693e516d44ffc6bf78b8483a9c3bd58b99753 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Fri, 30 Oct 2026 13:32:59 +0000 Subject: [PATCH 5/6] fix(server): restore awaited codex questions after restart --- .../Adapters/CodexAdapterV2.test.ts | 94 +++++++++++++------ .../Adapters/CodexAdapterV2.ts | 6 +- .../ProviderTurnStartService.ts | 20 +++- .../RunExecutionService.test.ts | 52 ++++++++++ .../orchestration-v2/RunExecutionService.ts | 12 +++ .../src/server/ProviderAdapter.ts | 1 + 6 files changed, 150 insertions(+), 35 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 9b73713a1a2fb..119c7516e4ed3 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -25,6 +25,7 @@ import { ProviderTurnId, RunAttemptId, RunId, + RuntimeRequestId, ThreadId, TurnItemId, } from "@t3tools/contracts"; @@ -1535,9 +1536,13 @@ describe("CodexAdapterV2 post-settle continuation", () => { event.type === "message.updated" && event.message.role === "assistant", ); - it.effect.each(["completed", "interrupted"] as const)( - "marks an asynchronous Codex question awaited until its sleep ends or the turn is %s", - (terminalStatus) => + it.effect.each([ + { terminalStatus: "completed", restored: false }, + { terminalStatus: "interrupted", restored: false }, + { terminalStatus: "completed", restored: true }, + ] as const)( + "marks an asynchronous Codex question awaited until the turn is $terminalStatus (restored: $restored)", + ({ terminalStatus, restored }) => Effect.scoped( Effect.gen(function* () { const nativeThreadId = "async-question-thread"; @@ -1557,24 +1562,28 @@ describe("CodexAdapterV2 post-settle continuation", () => { nativeTurnId, prompt: "Continue while I decide.", }), - { - type: "emit_inbound", - label: "question", - frame: { - method: "item/completed", - params: { - threadId: nativeThreadId, - turnId: nativeTurnId, - item: { - type: "agentMessage", - id: "async-question-item", - text: "Which branch?", - delivery: "async", - questions: [{ title: "Which branch?", options: ["main", "dev"] }], - }, - }, - }, - }, + ...(restored + ? [] + : [ + { + type: "emit_inbound", + label: "question", + frame: { + method: "item/completed", + params: { + threadId: nativeThreadId, + turnId: nativeTurnId, + item: { + type: "agentMessage", + id: "async-question-item", + text: "Which branch?", + delivery: "async", + questions: [{ title: "Which branch?", options: ["main", "dev"] }], + }, + }, + }, + } as const, + ]), // An interrupted turn ends without completing the sleep item. ...(terminalStatus === "completed" ? (["item/started", "item/completed"] as const) @@ -1619,32 +1628,55 @@ describe("CodexAdapterV2 post-settle continuation", () => { ], }); const harness = yield* makeCodexReplayHarness(transcript); - yield* harness.runtime.startTurn( - makeCodexTestTurnInput({ + const now = yield* DateTime.now; + const restoredRequest: OrchestrationV2RuntimeRequest = { + id: RuntimeRequestId.make("restored-question"), + nodeId: NodeId.make("restored-question-node"), + providerTurnId: null, + nativeRequestRef: { + driver: CodexAdapterV2.CODEX_DRIVER_KIND, + nativeId: "async:prior-question", + strength: "strong", + }, + kind: "user_input", + status: "pending", + responseCapability: { type: "message" }, + createdAt: now, + resolvedAt: null, + }; + const turnInput = { + ...makeCodexTestTurnInput({ threadId: harness.threadId, providerThread: harness.providerThread, - now: yield* DateTime.now, + now, attemptId: RunAttemptId.make("async-question-attempt"), text: "Continue while I decide.", }), - ); + pendingUserInputRequests: restored ? [restoredRequest] : [], + }; + yield* harness.runtime.startTurn(turnInput); yield* harness.firstTerminal; const requests = harness.events.flatMap((event) => event.type === "runtime_request.updated" ? [event.runtimeRequest] : [], ); assert.deepEqual( requests.map((request) => [request.status, request.responseCapability]), - [ - ["pending", { type: "message" }], - ["pending", { type: "message", agentWaiting: true }], - ["pending", { type: "message" }], - ], + restored + ? [ + ["pending", { type: "message", agentWaiting: true }], + ["pending", { type: "message" }], + ] + : [ + ["pending", { type: "message" }], + ["pending", { type: "message", agentWaiting: true }], + ["pending", { type: "message" }], + ], ); const questionItem = harness.events.find( (event) => event.type === "turn_item.updated" && event.turnItem.type === "user_input_request", ); - assert.equal(questionItem?.type, "turn_item.updated"); + assert.equal(questionItem?.type, restored ? undefined : "turn_item.updated"); if ( questionItem?.type === "turn_item.updated" && questionItem.turnItem.type === "user_input_request" diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 49ac06611cf29..12fd4944fdeda 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -1962,7 +1962,11 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( startedAt: input.startedAt, itemPositions: new Map(), fileChangeKindsByItem: new Map(), - asyncQuestions: new Map(), + asyncQuestions: new Map( + (input.turnInput.pendingUserInputRequests ?? []) + .filter((request) => request.nativeRequestRef?.driver === CODEX_PROVIDER) + .map((request) => [String(request.id), request]), + ), runningCompactionItemIds: new Set(), }; const nativeThreadId = input.turnInput.providerThread.nativeThreadRef?.nativeId; diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index 94ba406b106f5..332c7ff6399fe 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -624,9 +624,10 @@ export const layer: Layer.Layer< runs: current.runs, turnItems: current.turnItems, }); - const inheritedBackgroundTurnItems = yield* projectionStore - .getRuntimeRecoveryProjection(projection.thread.id) - .pipe(Effect.map(selectInheritedBackgroundItems)); + const runtimeProjection = yield* projectionStore.getRuntimeRecoveryProjection( + projection.thread.id, + ); + const inheritedBackgroundTurnItems = selectInheritedBackgroundItems(runtimeProjection); const providerSessionId = providerThread.providerSessionId; const runControls = makeRunControls({ threadId: projection.thread.id, @@ -1467,6 +1468,11 @@ export const layer: Layer.Layer< !noteContinuation ? session : makeDeliverySession(session, startWithHandoffs); + const providerTurnIds = new Set( + projection.providerTurns + .filter((turn) => turn.providerThreadId === runningProviderThread.id) + .map((turn) => turn.id), + ); return runExecution.startRootRun({ commandId: CommandId.make(`command:effect:provider-turn.start:${run.id}`), appThread: projection.thread, @@ -1476,6 +1482,14 @@ export const layer: Layer.Layer< rootNode: runningRootNode, checkpointScope, providerThread: runningProviderThread, + pendingUserInputRequests: runtimeProjection.runtimeRequests.filter( + (request) => + request.kind === "user_input" && + request.status === "pending" && + request.responseCapability.type === "message" && + request.providerTurnId !== null && + providerTurnIds.has(request.providerTurnId), + ), attempt: runningAttempt, attemptId: attempt.id, loadInheritedBackgroundTurnItems: runControls.loadInheritedBackgroundTurnItems, diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index 394cc1527cb7d..875d2b5bd2dba 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -23,6 +23,7 @@ import { ProviderTurnId, RunAttemptId, RunId, + RuntimeRequestId, ServerSettingsError, ThreadId, TurnItemId, @@ -498,6 +499,57 @@ it("does not route a superseded attempt through a reused provider thread", () => providerTurnId: null, }); assert.isFalse(RunExecutionService.routeProviderEvent(oldTurnEvent, newAttempt, newState)[0]); + + const requestId = RuntimeRequestId.make("request:shared-runtime:restart"); + const restoredState = RunExecutionService.makeProviderEventRoutingState({ + identity: newAttempt, + providerTurnId: null, + inheritedUserInputRequestIds: [requestId], + }); + const requestEvent: Extract< + ProviderAdapter.ProviderAdapterV2Event, + { type: "runtime_request.updated" } + > = { + type: "runtime_request.updated", + driver, + threadId, + runtimeRequest: { + id: requestId, + nodeId: NodeId.make("node:shared-runtime:question"), + providerTurnId: oldTurnEvent.providerTurn.id, + nativeRequestRef: null, + kind: "user_input", + status: "pending", + responseCapability: { type: "message", agentWaiting: true }, + createdAt: DateTime.makeUnsafe("2026-10-30T00:00:00Z"), + resolvedAt: null, + }, + }; + assert.isFalse(RunExecutionService.routeProviderEvent(requestEvent, newAttempt, newState)[0]); + assert.isTrue(RunExecutionService.routeProviderEvent(requestEvent, newAttempt, restoredState)[0]); + assert.isFalse( + RunExecutionService.routeProviderEvent( + { + ...requestEvent, + threadId: ThreadId.make("thread:other"), + }, + newAttempt, + restoredState, + )[0], + ); + assert.isFalse( + RunExecutionService.routeProviderEvent( + { + ...requestEvent, + runtimeRequest: { + ...requestEvent.runtimeRequest, + id: RuntimeRequestId.make("request:other"), + }, + }, + newAttempt, + restoredState, + )[0], + ); }); it("routes only exact same-thread background items inherited from settled runs", () => { diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index 3e3f37a15a936..4b4b38665fda2 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -15,12 +15,14 @@ import { type OrchestrationV2ProviderTurn, type OrchestrationV2Run, type OrchestrationV2RunAttempt, + type OrchestrationV2RuntimeRequest, type OrchestrationV2Subagent, type OrchestrationV2TurnItem, type ProviderSessionId, type ProviderThreadId, type ProviderTurnId, type RunAttemptId, + type RuntimeRequestId, type ThreadId, type TurnItemId, } from "@t3tools/contracts"; @@ -62,6 +64,7 @@ export interface ProviderEventRoutingState { readonly ownedProviderThreadIds: ReadonlySet; readonly ownedProviderTurns: ReadonlyMap; readonly inheritedBackgroundTurnItems: ReadonlyMap; + readonly inheritedUserInputRequestIds: ReadonlySet; readonly rootProviderTurnId: ProviderTurnId | null; } @@ -327,6 +330,7 @@ function finalProviderThreadStatus( export function makeProviderEventRoutingState(input: { readonly identity: ProviderEventRouteIdentity; readonly inheritedBackgroundTurnItems?: ReadonlyArray; + readonly inheritedUserInputRequestIds?: ReadonlyArray; readonly providerTurnId: ProviderTurnId | null; readonly relatedThreadIds?: ReadonlyArray; readonly relatedProviderThreadIds?: ReadonlyArray; @@ -346,6 +350,7 @@ export function makeProviderEventRoutingState(input: { inheritedBackgroundTurnItems: new Map( (input.inheritedBackgroundTurnItems ?? []).map((item) => [item.id, item.runId]), ), + inheritedUserInputRequestIds: new Set(input.inheritedUserInputRequestIds), rootProviderTurnId: input.providerTurnId, }; } @@ -503,6 +508,8 @@ export function routeProviderEvent( case "runtime_request.updated": return [ (event.threadId !== undefined && ownsChildThread(event.threadId)) || + (event.threadId === input.threadId && + state.inheritedUserInputRequestIds.has(event.runtimeRequest.id)) || (event.runtimeRequest.providerTurnId !== null && state.ownedProviderTurns.has(event.runtimeRequest.providerTurnId)), state, @@ -561,6 +568,7 @@ export interface RunExecutionServiceV2StartRootRunInput { readonly attemptId: RunAttemptId; readonly providerTurnOrdinal: number; readonly nativeThreadHasTurns?: boolean; + readonly pendingUserInputRequests?: ReadonlyArray; readonly loadInheritedBackgroundTurnItems?: () => Effect.Effect< ReadonlyArray, unknown @@ -1018,6 +1026,9 @@ export const layer: Layer.Layer< makeProviderEventRoutingState({ identity: routeIdentity, inheritedBackgroundTurnItems, + inheritedUserInputRequestIds: (input.pendingUserInputRequests ?? []).map( + (request) => request.id, + ), providerTurnId: input.attempt.providerTurnId, ...(input.relatedThreadIds === undefined ? {} @@ -1629,6 +1640,7 @@ export const layer: Layer.Layer< runId: input.run.id, runOrdinal: input.run.ordinal, providerTurnOrdinal: input.providerTurnOrdinal, + pendingUserInputRequests: input.pendingUserInputRequests ?? [], ...(input.nativeThreadHasTurns === undefined ? {} : { nativeThreadHasTurns: input.nativeThreadHasTurns }), diff --git a/packages/provider-core/src/server/ProviderAdapter.ts b/packages/provider-core/src/server/ProviderAdapter.ts index 9dc7ec1f62285..6a2d3e8634078 100644 --- a/packages/provider-core/src/server/ProviderAdapter.ts +++ b/packages/provider-core/src/server/ProviderAdapter.ts @@ -419,6 +419,7 @@ export interface ProviderAdapterV2TurnInput { readonly runId: RunId; readonly runOrdinal: number; readonly nativeThreadHasTurns?: boolean; + readonly pendingUserInputRequests?: ReadonlyArray; readonly providerTurnOrdinal: number; readonly restartContinuationOfRunId?: RunId; readonly attemptId: RunAttemptId; From 74857379aa2807aa69ade797332689ddb01ef842 Mon Sep 17 00:00:00 2001 From: Abhishek Date: Wed, 4 Nov 2026 12:41:56 +0530 Subject: [PATCH 6/6] fix(server): a repeated Codex async question completion is ignored Runtime request ids are random, so a repeated item/completed for an async question the turn already tracks added a second pending question. That copy was not marked awaited, so a sleeping turn could fall back to Working. The adapter now ignores the repeat, including for questions restored after restart. Co-Authored-By: Claude Opus 5.5 --- .../Adapters/CodexAdapterV2.test.ts | 78 +++++++++---------- .../Adapters/CodexAdapterV2.ts | 7 +- 2 files changed, 45 insertions(+), 40 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index 92ef72c64ce18..1198c26d2a382 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -1851,6 +1851,40 @@ describe("CodexAdapterV2 post-settle continuation", () => { outputTokens: 5, reasoningOutputTokens: 1, }; + const question = (label: string) => + ({ + type: "emit_inbound", + label, + frame: { + method: "item/completed", + params: { + threadId: nativeThreadId, + turnId: nativeTurnId, + item: { + type: "agentMessage", + id: restored ? "prior-question" : "async-question-item", + text: "Which branch?", + delivery: "async", + questions: [{ title: "Which branch?", options: ["main", "dev"] }], + }, + }, + }, + }) as const; + const sleep = (method: "item/started" | "item/completed") => + ({ + type: "emit_inbound", + label: `sleep ${method}`, + frame: { + method, + params: { + threadId: nativeThreadId, + turnId: nativeTurnId, + startedAtMs: 1782622441000, + completedAtMs: 1782622501000, + item: { type: "sleep", id: "sleep-item", durationMs: 60_000 }, + }, + }, + }) as const; const transcript = makeCodexReplayTranscript({ scenario: "async-question-and-billed-usage", entries: [ @@ -1859,46 +1893,12 @@ describe("CodexAdapterV2 post-settle continuation", () => { nativeTurnId, prompt: "Continue while I decide.", }), - ...(restored - ? [] - : [ - { - type: "emit_inbound", - label: "question", - frame: { - method: "item/completed", - params: { - threadId: nativeThreadId, - turnId: nativeTurnId, - item: { - type: "agentMessage", - id: "async-question-item", - text: "Which branch?", - delivery: "async", - questions: [{ title: "Which branch?", options: ["main", "dev"] }], - }, - }, - }, - } as const, - ]), + ...(restored ? [] : [question("question")]), + sleep("item/started"), + // A repeated completion of a tracked question must not reopen or replace it. + question("repeated question"), // An interrupted turn ends without completing the sleep item. - ...(terminalStatus === "completed" - ? (["item/started", "item/completed"] as const) - : (["item/started"] as const) - ).map((method) => ({ - type: "emit_inbound" as const, - label: `sleep ${method}`, - frame: { - method, - params: { - threadId: nativeThreadId, - turnId: nativeTurnId, - startedAtMs: 1782622441000, - completedAtMs: 1782622501000, - item: { type: "sleep", id: "sleep-item", durationMs: 60_000 }, - }, - }, - })), + ...(terminalStatus === "completed" ? [sleep("item/completed")] : []), { type: "emit_inbound", label: "usage", diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index cffe66e657507..4762cf42cadb6 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -5249,10 +5249,15 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( } if (payload.item.delivery === "async" && payload.item.questions?.length) { + const nativeRequestId = `async:${payload.item.id}`; + // Request ids are random, so a repeated completion would add a second question. + for (const request of context.asyncQuestions.values()) { + if (request.nativeRequestRef?.nativeId === nativeRequestId) return; + } const artifacts = yield* buildUserInputRequestArtifacts({ context, nativeItemId: payload.item.id, - nativeRequestId: `async:${payload.item.id}`, + nativeRequestId, responseMode: "message", questions: payload.item.questions.map((question, index) => ({ id: String(index),