From 21a6d4536b1812be723fad77121a8bbc781a139e Mon Sep 17 00:00:00 2001 From: Nathan Wong <20381195+thelogicmatrix@users.noreply.github.com> Date: Thu, 29 Oct 2026 15:17:57 +0800 Subject: [PATCH 1/7] fix(server): resend Codex requests restart recovery would drop Restart continuation resumes Codex natively with empty input, which assumes the cut turn's request is already in Codex's own history. Codex records a turn's input only after prompt hooks and pre-turn compaction, so a restart in that window resumed a turn with no user message. Codex now confirms the input: the adapter records `inputRecordedAt` on the provider turn when Codex completes the root turn's userMessage item, and the server and client projections keep it across later turn updates. When no native turn of a cut attempt has that confirmation, the continuation carries the original text, attachments and context, and turn start sends it as a prompt instead of resuming natively. Confirmed turns, including a goal's later turns, still resume natively. Other providers are unchanged. Fixes #26384 Co-Authored-By: Claude Opus 5.5 --- .../Adapters/CodexAdapterV2.ts | 58 +++--- .../src/orchestration-v2/ProjectionStore.ts | 6 +- .../ProviderTurnStartService.ts | 9 +- .../orchestration-v2/RestartBackgroundNote.ts | 30 +++ .../RestartContinuation.test.ts | 66 +++++++ .../orchestration-v2/RestartContinuation.ts | 30 ++- ...estratorReplayRecovery.integration.test.ts | 174 ++++++++++++++++++ .../state/orchestrationV2Projection.test.ts | 5 +- .../src/state/orchestrationV2Projection.ts | 17 +- packages/contracts/src/orchestrationV2.ts | 3 + 10 files changed, 354 insertions(+), 44 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 58cd4a0c0c17b..ffd3ee162d9d9 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -4336,6 +4336,33 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( }).pipe(Effect.orDie), ); + const updateRunningProviderTurn = ( + context: ActiveCodexTurnContext, + nativeTurnId: string, + update: Pick, + ) => + emitProviderEvent({ + type: "provider_turn.updated", + driver: CODEX_PROVIDER, + threadId: context.projectionThreadId, + providerTurn: { + id: context.providerTurnId, + providerThreadId: context.providerThread.id, + nodeId: context.providerNodeId, + runAttemptId: context.subagent === null ? context.input.attemptId : null, + nativeTurnRef: { + driver: CODEX_PROVIDER, + nativeId: nativeTurnId, + strength: "strong", + }, + ordinal: context.providerTurnOrdinal, + status: "running", + startedAt: context.startedAt, + completedAt: null, + ...update, + }, + }); + yield* client.handleServerNotification("thread/tokenUsage/updated", (payload) => Effect.gen(function* () { accumulateCodexTurnTokenUsage( @@ -4348,29 +4375,8 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( return; } const now = yield* DateTime.now; - yield* emitProviderEvent({ - type: "provider_turn.updated", - driver: CODEX_PROVIDER, - threadId: context.projectionThreadId, - providerTurn: { - id: context.providerTurnId, - providerThreadId: context.providerThread.id, - nodeId: context.providerNodeId, - runAttemptId: context.subagent === null ? context.input.attemptId : null, - nativeTurnRef: { - driver: CODEX_PROVIDER, - nativeId: payload.turnId, - strength: "strong", - }, - ordinal: context.providerTurnOrdinal, - status: "running", - startedAt: context.startedAt, - completedAt: null, - tokenUsage: codexProviderTurnTokenUsage( - payload.tokenUsage, - DateTime.formatIso(now), - ), - }, + yield* updateRunningProviderTurn(context, payload.turnId, { + tokenUsage: codexProviderTurnTokenUsage(payload.tokenUsage, DateTime.formatIso(now)), }); }).pipe(Effect.orDie), ); @@ -4819,6 +4825,12 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( if (yield* emitSubagentUserMessage(context, payload.item)) { return; } + // Restart recovery resumes natively only after Codex recorded the turn's input. + if (!settled && context.subagent === null) { + yield* updateRunningProviderTurn(context, payload.turnId, { + inputRecordedAt: yield* DateTime.now, + }); + } } if (payload.item.type === "commandExecution") { diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 2a21bfae4bd32..892ea692470a2 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -666,6 +666,9 @@ function upsertProviderTurn( ...((next.tokenUsage ?? current?.tokenUsage) === undefined ? {} : { tokenUsage: next.tokenUsage ?? current?.tokenUsage }), + ...((next.inputRecordedAt ?? current?.inputRecordedAt) === undefined + ? {} + : { inputRecordedAt: next.inputRecordedAt ?? current?.inputRecordedAt }), }); } @@ -2271,7 +2274,8 @@ export const layer: Layer.Layer = } case "provider-turn.updated": { const existingRows = - event.payload.tokenUsage === undefined + event.payload.tokenUsage === undefined || + event.payload.inputRecordedAt === undefined ? yield* sql` SELECT payload_json FROM orchestration_v2_projection_provider_turns diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index 94ba406b106f5..9b2991fb4be84 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -57,6 +57,7 @@ import * as RunExecutionService from "./RunExecutionService.ts"; import * as RuntimePolicy from "./RuntimePolicy.ts"; import { isRestartNoteContinuation, + isRestartResendContinuation, pendingRestartCancelledBackgroundWork, restartCancelledBackgroundWorkNote, } from "./RestartBackgroundNote.ts"; @@ -771,6 +772,10 @@ export const layer: Layer.Layer< projection.providerTurns, projection.attempts, ); + // A request the source turn never got to a model is resent as a prompt, not resumed natively. + const promptedContinuation = + noteContinuation || + isRestartResendContinuation(run, projection.runs, projection.providerTurns); const restartCancelledWork = pendingRestartCancelledBackgroundWork({ runs: projection.runs, providerTurns: projection.providerTurns, @@ -1416,7 +1421,7 @@ export const layer: Layer.Layer< .join("\n\n"); const { restartContinuationOfRunId: _resumedRunId, ...promptedInput } = turnInput; yield* start({ - ...(noteContinuation ? promptedInput : turnInput), + ...(promptedContinuation ? promptedInput : turnInput), message: { ...turnInput.message, text: @@ -1464,7 +1469,7 @@ export const layer: Layer.Layer< effectiveHandoffs.length === 0 && retryHandoff.length === 0 && restartNote === "" && - !noteContinuation + !promptedContinuation ? session : makeDeliverySession(session, startWithHandoffs); return runExecution.startRootRun({ diff --git a/apps/server/src/orchestration-v2/RestartBackgroundNote.ts b/apps/server/src/orchestration-v2/RestartBackgroundNote.ts index ba3830f65453d..be1f2b32a28d0 100644 --- a/apps/server/src/orchestration-v2/RestartBackgroundNote.ts +++ b/apps/server/src/orchestration-v2/RestartBackgroundNote.ts @@ -236,6 +236,36 @@ export function isRestartNoteContinuation( ); } +/** + * Without the provider's confirmation, the native history may lack the request: Codex runs prompt + * hooks and pre-turn compaction first. Any native turn of the attempt counts, so a goal's later turns resume natively. + */ +export function restartSourceInputUnrecorded( + source: Pick, + providerTurns: ReadonlyArray< + Pick + >, +): boolean { + return ( + source.activeAttemptId == null || + !providerTurns.some( + (turn) => turn.runAttemptId === source.activeAttemptId && turn.inputRecordedAt !== undefined, + ) + ); +} + +export function isRestartResendContinuation( + run: Pick, + runs: ReadonlyArray, + providerTurns: Parameters[1], +): boolean { + const source = + run.restartContinuationOfRunId === undefined + ? undefined + : runs.find((candidate) => candidate.id === run.restartContinuationOfRunId); + return source !== undefined && restartSourceInputUnrecorded(source, providerTurns); +} + /** Derive delivery from later completed turns on the same provider thread, excluding compactions, resumes, and rolled-back runs. */ export function pendingRestartCancelledBackgroundWork(input: { readonly runs: ReadonlyArray; diff --git a/apps/server/src/orchestration-v2/RestartContinuation.test.ts b/apps/server/src/orchestration-v2/RestartContinuation.test.ts index 061fc54a7f112..49a23fcb77db6 100644 --- a/apps/server/src/orchestration-v2/RestartContinuation.test.ts +++ b/apps/server/src/orchestration-v2/RestartContinuation.test.ts @@ -970,6 +970,72 @@ it.effect("does not continue a cut /compact or /logout turn", () => }), ); +const cutRequest = { + id: MessageId.make("message:user"), + text: "find the cheapest return flights", + attachments: [ + { + type: "image" as const, + id: "attachment:dates", + name: "dates.png", + mimeType: "image/png", + sizeBytes: 1, + }, + ], + context: { version: 1 as const, records: [] }, +}; + +it.effect("resends a Codex request the restart cut before Codex recorded it", () => + Effect.gen(function* () { + const cut = cutMidTurn(); + for (const providerTurns of [ + cut.providerTurns, + // Pre-turn compaction reports usage before Codex records the new request. + cut.providerTurns.map((turn) => ({ ...turn, tokenUsage: { usedTokens: 1 } })), + ]) { + const dispatches = yield* continuationDispatches({ + ...cut, + messages: [cutRequest], + providerTurns, + } as unknown as OrchestrationV2ThreadProjection); + assert.deepEqual( + dispatches.map((command) => [command.text, command.attachments, command.context]), + [[cutRequest.text, cutRequest.attachments, cutRequest.context]], + ); + } + }), +); + +it.effect("resumes natively once Codex recorded the request, and for other providers", () => + Effect.gen(function* () { + const cut = cutMidTurn(); + const inputRecordedAt = DateTime.makeUnsafe("2026-10-03T11:59:00.000Z"); + const recorded = { + ...cut, + messages: [cutRequest], + providerTurns: cut.providerTurns.map((turn) => ({ ...turn, inputRecordedAt })), + } as unknown as OrchestrationV2ThreadProjection; + assert.deepEqual(yield* continuationTexts(recorded), ["Continue where you left off."]); + // A goal's first native turn recorded the request; its cut follow-up turn records none. + const goalFollowUp = { + ...cut, + messages: [{ ...cutRequest, text: "/goal find the cheapest return flights" }], + providerTurns: [ + { ...cut.providerTurns[0]!, ordinal: 1, status: "completed", inputRecordedAt }, + { ...cut.providerTurns[0]!, id: "turn:goal-follow-up", ordinal: 2 }, + ], + } as unknown as OrchestrationV2ThreadProjection; + assert.deepEqual(yield* continuationTexts(goalFollowUp), ["Continue where you left off."]); + const claude = ProviderDriverKind.make("claudeAgent"); + const otherDriver = { + ...cut, + messages: [cutRequest], + providerThreads: cut.providerThreads.map((thread) => ({ ...thread, driver: claude })), + } as unknown as OrchestrationV2ThreadProjection; + assert.deepEqual(yield* continuationTexts(otherDriver), ["Continue where you left off."]); + }), +); + it.effect( "carries the note forward when its continuation was cut before reaching the provider", () => diff --git a/apps/server/src/orchestration-v2/RestartContinuation.ts b/apps/server/src/orchestration-v2/RestartContinuation.ts index c81faa6cf47cb..054eae7abdb7b 100644 --- a/apps/server/src/orchestration-v2/RestartContinuation.ts +++ b/apps/server/src/orchestration-v2/RestartContinuation.ts @@ -26,6 +26,7 @@ import { neverStarted, restartCancelledBackgroundWorkNote, restartContinuationNote, + restartSourceInputUnrecorded, } from "./RestartBackgroundNote.ts"; const CONTINUE_PROMPT = "Continue where you left off."; @@ -188,7 +189,15 @@ export const continueRestartedRun = Effect.fn("RestartContinuation.continueResta const messageId = MessageId.make(`message:restart-continuation:${input.sourceRunId}`); const projection = yield* threads.getThreadRecords( input.threadId, - ["messages", "runs", "providerTurns", "attempts", "runtimeRequests", "turnItems"], + [ + "messages", + "runs", + "providerThreads", + "providerTurns", + "attempts", + "runtimeRequests", + "turnItems", + ], { messageIds: [messageId], turnItemTypes: ["secret_request"] }, ); if ( @@ -267,12 +276,18 @@ export const continueRestartedRun = Effect.fn("RestartContinuation.continueResta ); const noteText = note.work.length === 0 ? undefined : restartCancelledBackgroundWorkNote(note.work); + // Codex resumes natively with no input, so a request it never recorded must be sent again. + const resent = + !note.settled && + sourceMessage !== undefined && + projection.providerThreads.find((thread) => thread.id === source.providerThreadId)?.driver === + "codex" && + restartSourceInputUnrecorded(source, projection.providerTurns) + ? sourceMessage + : undefined; + const prompt = resent?.text ?? CONTINUE_PROMPT; const text = - noteText === undefined - ? CONTINUE_PROMPT - : note.settled - ? noteText - : `${noteText}\n\n${CONTINUE_PROMPT}`; + noteText === undefined ? prompt : note.settled ? noteText : `${noteText}\n\n${prompt}`; yield* threads .dispatch({ type: "message.dispatch", @@ -288,7 +303,8 @@ export const continueRestartedRun = Effect.fn("RestartContinuation.continueResta : "T3 Code restarted and resumed this turn", detail: text, }, - attachments: [], + attachments: resent?.attachments ?? [], + ...(resent?.context === undefined ? {} : { context: resent.context }), modelSelection: source.modelSelection, dispatchMode: { type: "start_immediately" }, createdBy: "agent", diff --git a/apps/server/src/orchestration-v2/testkit/OrchestratorReplayRecovery.integration.test.ts b/apps/server/src/orchestration-v2/testkit/OrchestratorReplayRecovery.integration.test.ts index d17b407506018..830b09a45699e 100644 --- a/apps/server/src/orchestration-v2/testkit/OrchestratorReplayRecovery.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/OrchestratorReplayRecovery.integration.test.ts @@ -1,11 +1,13 @@ import * as NodeServices from "@effect/platform-node/NodeServices"; import { assert, describe, it } from "@effect/vitest"; import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Path from "effect/Path"; import * as Schema from "effect/Schema"; import * as SqlClient from "effect/sql/SqlClient"; +import * as Stream from "effect/Stream"; import * as CodexReplay from "effect-codex-app-server/replay"; import { type OrchestrationV2ThreadProjection, @@ -27,6 +29,7 @@ import { import * as CursorAdapterV2Testkit from "../Adapters/CursorAdapterV2.testkit.ts"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import * as EffectWorker from "../EffectWorker.ts"; +import * as Orchestrator from "../Orchestrator.ts"; import * as SqlitePersistence from "../../persistence/Sqlite.ts"; import { provideDeterministicTestRuntime } from "./DeterministicRuntime.ts"; import { @@ -78,6 +81,49 @@ const readCodexTranscript = Effect.fn("readCodexRecoveryTranscript")(function* ( ); return yield* decodeCodexTranscript(materializeReplayTranscriptWorkspace(transcript, workspace)); }); +/** The first runtime stops once Codex starts the turn, before it records the request; the second resumes it. */ +const readCodexCutTurnTranscripts = Effect.fn("readCodexCutTurnTranscripts")(function* ( + workspace: string, +) { + const recorded = materializeReplayTranscriptWorkspace( + yield* readRawTranscript( + new URL("./fixtures/provider_thread_resume/codex_transcript.ndjson", import.meta.url), + ), + workspace, + ); + const { entries } = recorded; + type Frame = { readonly params?: Record }; + const frameOf = (entry: (typeof entries)[number] | undefined): Frame | undefined => + entry === undefined || entry.type === "runtime_exit" ? undefined : (entry.frame as Frame); + const firstStart = entries.findIndex( + (entry) => entry.type === "expect_outbound" && entry.label === "turn/start", + ); + const firstStarted = entries.findIndex( + (entry) => entry.type === "emit_inbound" && entry.label === "turn/started", + ); + const secondRuntime = entries.findIndex( + (entry, index) => + index > firstStarted && entry.type === "expect_outbound" && entry.label === "initialize", + ); + const request = frameOf(entries[firstStart])?.params?.input; + const cutTurnId = frameOf(entries[firstStarted])?.params?.turn?.id; + if (firstStart < 0 || firstStarted < 0 || secondRuntime < 0 || cutTurnId === undefined) { + throw new Error("provider_thread_resume must record two Codex runtimes."); + } + const resumed = entries.slice(secondRuntime).flatMap((entry): Array => { + if (entry.type === "runtime_exit") return [entry]; + const frame = frameOf(entry); + if (entry.label === "thread/tokenUsage/updated" && frame?.params?.turnId === cutTurnId) { + return []; + } + if (entry.type !== "expect_outbound" || entry.label !== "turn/start") return [entry]; + return [{ ...entry, frame: { ...frame, params: { ...frame?.params, input: request } } }]; + }); + return { + cut: yield* decodeCodexTranscript({ ...recorded, entries: entries.slice(0, firstStarted + 1) }), + resumed: yield* decodeCodexTranscript({ ...recorded, entries: resumed }), + }; +}); const decodePromptPair = Schema.decodeUnknownEffect(Schema.Tuple([Schema.String, Schema.String])); const readClaudeSubagentResumeTranscript = Effect.fn("readClaudeSubagentResumeTranscript")( function* () { @@ -365,6 +411,134 @@ describe("orchestrator replay recovery", () => { ), ); + it.effect("resends a Codex request a restart cut before Codex recorded it", () => + Effect.scoped( + Effect.gen(function* () { + const workspace = yield* checkpointWorkspace("provider_thread_resume"); + const transcripts = yield* readCodexCutTurnTranscripts(workspace); + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const tempDir = yield* Effect.acquireRelease( + fs.makeTempDirectory({ prefix: "t3-orchestration-v2-cut-turn-" }), + (directory) => fs.remove(directory, { recursive: true, force: true }).pipe(Effect.orDie), + ); + const snapshotPath = path.join(tempDir, "restart-state.sqlite"); + const materialized = yield* materializeFixtureInput({ + scenario: "provider_thread_resume", + fixtureInput: { steps: [{ type: "message", text: PROVIDER_THREAD_RESUME_FIRST_PROMPT }] }, + driver: ProviderDriverKind.make("codex"), + modelSelection: CODEX_MODEL_SELECTION, + }); + const threadId = materialized.projectionThreadIds[0]; + const idleIndex = materialized.steps.findIndex((step) => step.type === "await_thread_idle"); + if (threadId === undefined || idleIndex < 0) { + return yield* Effect.die("cut-turn scenario needs one thread and an idle step"); + } + const harnessFor = (transcript: CodexReplay.CodexAppServerReplayTranscript) => ({ + ...CodexOrchestratorReplayHarness, + makeProviderAdapterRegistryLayer: () => CodexAdapterV2Testkit.layer({ transcript }), + }); + const scenario = ( + name: string, + transcript: CodexReplay.CodexAppServerReplayTranscript, + steps: typeof materialized.steps, + ) => ({ + name: `provider_thread_resume/codex:${name}`, + transcript, + commands: steps.flatMap((step) => (step.type === "dispatch" ? [step.command] : [])), + steps, + projectionThreadIds: materialized.projectionThreadIds, + runtimePolicyOverride: { cwd: workspace }, + }); + const layerDatabase = SqlitePersistence.layerFromPath( + path.join(tempDir, "state.sqlite"), + ).pipe(Layer.provide(NodeServices.layer)); + + const cut = scenario("cut", transcripts.cut, materialized.steps.slice(0, idleIndex)); + yield* Effect.scoped( + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + // Codex started the turn and has not yet confirmed recording the request. + const turnStarted = yield* orchestrator.streamStoredEvents.pipe( + Stream.filter( + ({ event }) => + event.type === "provider-turn.updated" && event.payload.status === "running", + ), + Stream.runHead, + Effect.forkScoped, + ); + yield* runOrchestratorV2Scenario(cut); + yield* Fiber.join(turnStarted); + const sql = yield* SqlClient.SqlClient; + yield* sql`VACUUM INTO ${snapshotPath}`; + }).pipe( + Effect.provide( + Layer.merge( + layerProviderReplay(cut, harnessFor(transcripts.cut), { + databaseLayer: layerDatabase, + }), + layerDatabase, + ), + ), + ), + ); + + const resumed = scenario( + "resumed", + transcripts.resumed, + materialized.steps.slice(idleIndex), + ); + const result = yield* Effect.scoped( + Effect.gen(function* () { + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + yield* worker.drain(); + return yield* runOrchestratorV2Scenario(resumed); + }).pipe( + Effect.provide( + layerProviderReplay(resumed, harnessFor(transcripts.resumed), { + databaseLayer: SqlitePersistence.layerFromPath(snapshotPath).pipe( + Layer.provide(NodeServices.layer), + ), + recoverOnStartup: true, + continueThreadsAfterServerUpdate: true, + }), + ), + ), + ); + + const projection = projectionFor(result, transcripts.resumed.scenario); + assert.deepEqual( + projection.runs.map((run) => [run.status, run.restartContinuationOfRunId]), + [ + ["cancelled", undefined], + ["completed", projection.runs[0]?.id], + ], + ); + // The replay only accepts the resumed turn/start when its input is the cut request. + assert.deepEqual( + projection.messages + .filter((message) => message.id === projection.runs[1]?.userMessageId) + .map((message) => message.text), + [PROVIDER_THREAD_RESUME_FIRST_PROMPT], + ); + // Codex confirmed the resent request, and later usage updates kept that confirmation. + assert.deepEqual( + projection.providerTurns.map((turn) => [ + turn.inputRecordedAt !== undefined, + turn.tokenUsage !== undefined, + ]), + [ + [false, false], + [true, true], + ], + ); + }).pipe( + provideDeterministicTestRuntime, + Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer)), + ), + ), + ); + it.effect( "resumes a provider-native Cursor thread after recreating the orchestrator runtime", () => diff --git a/packages/client-runtime/src/state/orchestrationV2Projection.test.ts b/packages/client-runtime/src/state/orchestrationV2Projection.test.ts index f6ba3e9f0c1a7..88f6d404b2cde 100644 --- a/packages/client-runtime/src/state/orchestrationV2Projection.test.ts +++ b/packages/client-runtime/src/state/orchestrationV2Projection.test.ts @@ -134,7 +134,7 @@ describe("applyOrchestrationV2ProjectionEvent", () => { expect(next?.thread.pullRequests?.[0]).toBe(links[0]); }); - it("keeps live token usage when the terminal provider turn omits it", () => { + it("keeps live token usage and input confirmation when the terminal provider turn omits them", () => { const providerTurnId = ProviderTurnId.make("provider-turn-reducer"); const running = { id: providerTurnId, @@ -151,6 +151,7 @@ describe("applyOrchestrationV2ProjectionEvent", () => { maxTokens: 200_000, updatedAt: "2026-08-29T00:00:00.000Z", }, + inputRecordedAt: now, }; const projection = { ...emptyProjection, providerTurns: [running] }; const event = { @@ -164,6 +165,7 @@ describe("applyOrchestrationV2ProjectionEvent", () => { status: "completed", completedAt: now, tokenUsage: undefined, + inputRecordedAt: undefined, }, } as OrchestrationV2DomainEvent; @@ -171,6 +173,7 @@ describe("applyOrchestrationV2ProjectionEvent", () => { expect(next?.providerTurns[0]?.status).toBe("completed"); expect(next?.providerTurns[0]?.tokenUsage).toEqual(running.tokenUsage); + expect(next?.providerTurns[0]?.inputRecordedAt).toEqual(now); }); it("applies thread lifecycle payloads instead of leaving stale metadata", () => { diff --git a/packages/client-runtime/src/state/orchestrationV2Projection.ts b/packages/client-runtime/src/state/orchestrationV2Projection.ts index fbba7d1bd9dd4..9703342951656 100644 --- a/packages/client-runtime/src/state/orchestrationV2Projection.ts +++ b/packages/client-runtime/src/state/orchestrationV2Projection.ts @@ -347,22 +347,19 @@ export function applyOrchestrationV2ProjectionEvent( }; case "provider-thread.updated": return { ...base, providerThreads: upsertEntity(base.providerThreads, event.payload) }; - case "provider-turn.updated": + case "provider-turn.updated": { + const current = base.providerTurns.find((turn) => turn.id === event.payload.id); + const tokenUsage = event.payload.tokenUsage ?? current?.tokenUsage; + const inputRecordedAt = event.payload.inputRecordedAt ?? current?.inputRecordedAt; return { ...base, providerTurns: upsertEntity(base.providerTurns, { ...event.payload, - ...((event.payload.tokenUsage ?? - base.providerTurns.find((turn) => turn.id === event.payload.id)?.tokenUsage) === - undefined - ? {} - : { - tokenUsage: - event.payload.tokenUsage ?? - base.providerTurns.find((turn) => turn.id === event.payload.id)?.tokenUsage, - }), + ...(tokenUsage === undefined ? {} : { tokenUsage }), + ...(inputRecordedAt === undefined ? {} : { inputRecordedAt }), }), }; + } case "runtime-request.updated": return { ...base, runtimeRequests: upsertEntity(base.runtimeRequests, event.payload) }; case "message.updated": diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 498f81a0e848a..0c8275327ee2c 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -1156,6 +1156,8 @@ export const OrchestrationV2ProviderTurn = Schema.Struct({ completedAt: Schema.NullOr(Schema.DateTimeUtc), tokenUsage: Schema.optional(OrchestrationV2ProviderTurnTokenUsage), turnTokenUsage: Schema.optional(TurnTokenUsage), + /** When the provider confirmed it recorded the turn's user input in its native history. */ + inputRecordedAt: Schema.optional(Schema.DateTimeUtc), }); export type OrchestrationV2ProviderTurn = typeof OrchestrationV2ProviderTurn.Type; @@ -2336,6 +2338,7 @@ export const OrchestrationV2ProviderTurnJson = OrchestrationV2ProviderTurn.mapFi ...fields, startedAt: Schema.NullOr(Schema.DateTimeUtcFromString), completedAt: Schema.NullOr(Schema.DateTimeUtcFromString), + inputRecordedAt: Schema.optional(Schema.DateTimeUtcFromString), })); export type OrchestrationV2ProviderTurnJson = typeof OrchestrationV2ProviderTurnJson.Type; From 4ed56474999b4f941aa285370ed3d872d7f1e20e Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Sat, 31 Oct 2026 22:34:25 +0000 Subject: [PATCH 2/7] test(server): type codex restart replay frames --- .../OrchestratorReplayRecovery.integration.test.ts | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/testkit/OrchestratorReplayRecovery.integration.test.ts b/apps/server/src/orchestration-v2/testkit/OrchestratorReplayRecovery.integration.test.ts index 830b09a45699e..7d77038b17719 100644 --- a/apps/server/src/orchestration-v2/testkit/OrchestratorReplayRecovery.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/OrchestratorReplayRecovery.integration.test.ts @@ -92,7 +92,13 @@ const readCodexCutTurnTranscripts = Effect.fn("readCodexCutTurnTranscripts")(fun workspace, ); const { entries } = recorded; - type Frame = { readonly params?: Record }; + type Frame = { + readonly params?: { + readonly input?: unknown; + readonly turn?: { readonly id?: string }; + readonly turnId?: string; + }; + }; const frameOf = (entry: (typeof entries)[number] | undefined): Frame | undefined => entry === undefined || entry.type === "runtime_exit" ? undefined : (entry.frame as Frame); const firstStart = entries.findIndex( From 628147266014051fe76875877528aa89ecfa9d84 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Sat, 31 Oct 2026 22:41:35 +0000 Subject: [PATCH 3/7] fix(server): preserve codex input confirmation across restarts --- .../RestartBackgroundNote.test.ts | 17 +++++++++++ .../orchestration-v2/RestartBackgroundNote.ts | 30 ++++++++++++++----- .../RestartContinuation.test.ts | 4 ++- .../orchestration-v2/RestartContinuation.ts | 2 +- ...estratorReplayRecovery.integration.test.ts | 20 +++++++++++++ 5 files changed, 63 insertions(+), 10 deletions(-) diff --git a/apps/server/src/orchestration-v2/RestartBackgroundNote.test.ts b/apps/server/src/orchestration-v2/RestartBackgroundNote.test.ts index 24269dfeaff24..41a2cf83b9055 100644 --- a/apps/server/src/orchestration-v2/RestartBackgroundNote.test.ts +++ b/apps/server/src/orchestration-v2/RestartBackgroundNote.test.ts @@ -13,6 +13,7 @@ import { cancelledTurnItemWork, isAwaitedWork, isRestartNoteContinuation, + isRestartResendContinuation, pendingRestartCancelledBackgroundWork, restartContinuationNote, mergeRestartCancelledBackgroundWork, @@ -44,6 +45,22 @@ const turnFor = (source: OrchestrationV2Run) => ({ status: "completed" as const, }); +it("keeps native restart continuations native across another restart", () => { + const source = run(1, codexThread); + const resumed = run(2, codexThread, { restartContinuationOfRunId: source.id }); + const next = run(3, codexThread, { restartContinuationOfRunId: resumed.id }); + const confirmed = { + ...turnFor(source), + inputRecordedAt: DateTime.makeUnsafe("2026-10-03T10:00:00.000Z"), + }; + assert.isFalse(isRestartResendContinuation(next, [source, resumed, next], [confirmed])); + assert.isTrue(isRestartResendContinuation(next, [source, resumed, next], [])); + const unrelated = run(2, codexThread); + assert.isTrue(isRestartResendContinuation(next, [source, unrelated, next], [confirmed])); + const switched = { ...source, providerThreadId: claudeThread }; + assert.isTrue(isRestartResendContinuation(next, [switched, resumed, next], [confirmed])); +}); + it("keeps the note for the provider thread that lost the work across a provider switch", () => { const root = run(1, claudeThread, { restartCancelledBackgroundWork: lost }); const onCodex = run(2, codexThread); diff --git a/apps/server/src/orchestration-v2/RestartBackgroundNote.ts b/apps/server/src/orchestration-v2/RestartBackgroundNote.ts index cf8111345b5fc..f946e2e180f79 100644 --- a/apps/server/src/orchestration-v2/RestartBackgroundNote.ts +++ b/apps/server/src/orchestration-v2/RestartBackgroundNote.ts @@ -282,20 +282,34 @@ export function isRestartNoteContinuation( /** * Without the provider's confirmation, the native history may lack the request: Codex runs prompt - * hooks and pre-turn compaction first. Any native turn of the attempt counts, so a goal's later turns resume natively. + * hooks and pre-turn compaction first. Native continuations inherit the source request's confirmation. */ export function restartSourceInputUnrecorded( - source: Pick, + source: OrchestrationV2Run, providerTurns: ReadonlyArray< Pick >, + runs: ReadonlyArray, ): boolean { - return ( - source.activeAttemptId == null || - !providerTurns.some( - (turn) => turn.runAttemptId === source.activeAttemptId && turn.inputRecordedAt !== undefined, + const visited = new Set(); + let current: OrchestrationV2Run | undefined = source; + while (current !== undefined && !visited.has(current.id)) { + visited.add(current.id); + const attemptId = current.activeAttemptId; + if ( + attemptId !== null && + providerTurns.some( + (turn) => turn.runAttemptId === attemptId && turn.inputRecordedAt !== undefined, + ) ) - ); + return false; + const previousId: OrchestrationV2Run["restartContinuationOfRunId"] = + current.restartContinuationOfRunId; + current = runs.find( + (run) => run.id === previousId && run.providerThreadId === source.providerThreadId, + ); + } + return true; } export function isRestartResendContinuation( @@ -307,7 +321,7 @@ export function isRestartResendContinuation( run.restartContinuationOfRunId === undefined ? undefined : runs.find((candidate) => candidate.id === run.restartContinuationOfRunId); - return source !== undefined && restartSourceInputUnrecorded(source, providerTurns); + return source !== undefined && restartSourceInputUnrecorded(source, providerTurns, runs); } /** Derive delivery from later completed turns on the same provider thread, excluding compactions, resumes, and rolled-back runs. */ diff --git a/apps/server/src/orchestration-v2/RestartContinuation.test.ts b/apps/server/src/orchestration-v2/RestartContinuation.test.ts index 41ff8102ed37e..ebb0da183eca7 100644 --- a/apps/server/src/orchestration-v2/RestartContinuation.test.ts +++ b/apps/server/src/orchestration-v2/RestartContinuation.test.ts @@ -1137,7 +1137,9 @@ it.effect("resumes natively once Codex recorded the request, and for other provi // A goal's first native turn recorded the request; its cut follow-up turn records none. const goalFollowUp = { ...cut, - messages: [{ ...cutRequest, text: "/goal find the cheapest return flights" }], + messages: [ + { ...cutRequest, text: "/goal find the cheapest return flights", attachments: [] }, + ], providerTurns: [ { ...cut.providerTurns[0]!, ordinal: 1, status: "completed", inputRecordedAt }, { ...cut.providerTurns[0]!, id: "turn:goal-follow-up", ordinal: 2 }, diff --git a/apps/server/src/orchestration-v2/RestartContinuation.ts b/apps/server/src/orchestration-v2/RestartContinuation.ts index f55edac42a431..45ba18822180d 100644 --- a/apps/server/src/orchestration-v2/RestartContinuation.ts +++ b/apps/server/src/orchestration-v2/RestartContinuation.ts @@ -313,7 +313,7 @@ export const continueRestartedRun = Effect.fn("RestartContinuation.continueResta sourceMessage !== undefined && projection.providerThreads.find((thread) => thread.id === source.providerThreadId)?.driver === "codex" && - restartSourceInputUnrecorded(source, projection.providerTurns) + restartSourceInputUnrecorded(source, projection.providerTurns, projection.runs) ? sourceMessage : undefined; const prompt = resent?.text ?? CONTINUE_PROMPT; diff --git a/apps/server/src/orchestration-v2/testkit/OrchestratorReplayRecovery.integration.test.ts b/apps/server/src/orchestration-v2/testkit/OrchestratorReplayRecovery.integration.test.ts index 7d77038b17719..674a809736f35 100644 --- a/apps/server/src/orchestration-v2/testkit/OrchestratorReplayRecovery.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/OrchestratorReplayRecovery.integration.test.ts @@ -97,6 +97,7 @@ const readCodexCutTurnTranscripts = Effect.fn("readCodexCutTurnTranscripts")(fun readonly input?: unknown; readonly turn?: { readonly id?: string }; readonly turnId?: string; + readonly item?: { readonly type: string; readonly content?: unknown }; }; }; const frameOf = (entry: (typeof entries)[number] | undefined): Frame | undefined => @@ -122,6 +123,25 @@ const readCodexCutTurnTranscripts = Effect.fn("readCodexCutTurnTranscripts")(fun if (entry.label === "thread/tokenUsage/updated" && frame?.params?.turnId === cutTurnId) { return []; } + if (entry.type === "emit_inbound" && frame?.params?.item?.type === "userMessage") { + return [ + { + ...entry, + frame: { + ...frame, + params: { + ...frame.params, + item: { + ...frame.params.item, + content: [ + { type: "text", text: PROVIDER_THREAD_RESUME_FIRST_PROMPT, text_elements: [] }, + ], + }, + }, + }, + }, + ]; + } if (entry.type !== "expect_outbound" || entry.label !== "turn/start") return [entry]; return [{ ...entry, frame: { ...frame, params: { ...frame?.params, input: request } } }]; }); From 04a5559cda3254f10a2ebf7a160b875bb84f133b Mon Sep 17 00:00:00 2001 From: Nathan Wong <20381195+thelogicmatrix@users.noreply.github.com> Date: Tue, 3 Nov 2026 13:04:39 +0800 Subject: [PATCH 4/7] fix(server): preserve unaccepted ACP requests across restarts --- .../Adapters/AcpRegistryAdapterV2.test.ts | 61 +++++++++++++++++++ .../RestartContinuation.test.ts | 54 ++++++++++------ .../orchestration-v2/RestartContinuation.ts | 8 ++- packages/provider-acp/src/server/adapter.ts | 34 ++++++++++- 4 files changed, 134 insertions(+), 23 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/AcpRegistryAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpRegistryAdapterV2.test.ts index 49ac36c553214..76f0de92815fe 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpRegistryAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpRegistryAdapterV2.test.ts @@ -1244,6 +1244,67 @@ describe("AcpRegistryAdapterV2", () => { }).pipe(Effect.provide(layerTest), Effect.scoped), ); + it.effect.each(["user_message", "agent_message_chunk"] as const)( + "confirms input only after a root %s, and keeps it when the turn settles", + (sessionUpdate) => + Effect.gen(function* () { + const marker = (title: string) => + claudeFrames.update(CLAUDE_ROOT_SESSION, { + sessionUpdate: "session_info_update", + title, + }); + const session = yield* openClaudeReplaySession([ + ...claudeFrames.open, + claudeFrames.prompt(1)[0], + claudeFrames.update(CLAUDE_ROOT_SESSION, { + sessionUpdate: "usage_update", + used: 1, + size: 100, + }), + claudeFrames.update("unrelated-child", { + sessionUpdate: "agent_message_chunk", + content: { type: "text", text: "Earlier background work" }, + }), + claudeFrames.update(CLAUDE_ROOT_SESSION, { + sessionUpdate: "compaction_update", + compactionId: "before-input", + status: "running", + }), + marker("before acceptance"), + claudeFrames.update(CLAUDE_ROOT_SESSION, { + sessionUpdate, + ...(sessionUpdate === "user_message" + ? { messageId: "accepted", content: [{ type: "text", text: "test prompt" }] } + : { content: { type: "text", text: "Working on the request" } }), + }), + marker("after acceptance"), + ...claudeFrames.prompt(1).slice(1), + claudeFrames.idle("end_turn"), + claudeFrames.closeRequest, + claudeFrames.closed, + ]); + yield* session.runtime.startTurn(session.turnInput(1)); + const before = yield* session.throughMarker("before acceptance"); + const initial = before.flatMap((event) => + event.type === "provider_turn.updated" ? [event.providerTurn] : [], + ); + const acceptedEvents = yield* session.throughMarker("after acceptance"); + const after = yield* session.until((event) => event.type === "turn.terminal"); + const turns = [...acceptedEvents, ...after].flatMap((event) => + event.type === "provider_turn.updated" ? [event.providerTurn] : [], + ); + assert.isNotEmpty(initial); + assert.isTrue(initial.every((turn) => turn.inputRecordedAt === undefined)); + const accepted = acceptedEvents.find((event) => event.type === "provider_turn.updated"); + const acceptedTurn = + accepted?.type === "provider_turn.updated" ? accepted.providerTurn : undefined; + assert.isDefined(acceptedTurn?.inputRecordedAt); + const terminal = turns.at(-1)!; + assert.equal(terminal.status, "completed"); + assert.deepEqual(terminal.inputRecordedAt, acceptedTurn?.inputRecordedAt); + }).pipe(Effect.provide(layerTest), Effect.scoped), + ); + it.effect.each(["before", "after"] as const)( "reconciles a full prompt echo %s acknowledgment", (ordering) => diff --git a/apps/server/src/orchestration-v2/RestartContinuation.test.ts b/apps/server/src/orchestration-v2/RestartContinuation.test.ts index ebb0da183eca7..11eeafa25f43d 100644 --- a/apps/server/src/orchestration-v2/RestartContinuation.test.ts +++ b/apps/server/src/orchestration-v2/RestartContinuation.test.ts @@ -1103,25 +1103,31 @@ const cutRequest = { context: { version: 1 as const, records: [] }, }; -it.effect("resends a Codex request the restart cut before Codex recorded it", () => - Effect.gen(function* () { - const cut = cutMidTurn(); - for (const providerTurns of [ - cut.providerTurns, - // Pre-turn compaction reports usage before Codex records the new request. - cut.providerTurns.map((turn) => ({ ...turn, tokenUsage: { usedTokens: 1 } })), - ]) { - const dispatches = yield* continuationDispatches({ - ...cut, - messages: [cutRequest], - providerTurns, - } as unknown as OrchestrationV2ThreadProjection); - assert.deepEqual( - dispatches.map((command) => [command.text, command.attachments, command.context]), - [[cutRequest.text, cutRequest.attachments, cutRequest.context]], - ); - } - }), +it.effect.each(["codex", "acpRegistry"] as const)( + "resends a %s request and its image before the provider accepted it", + (driverKind) => + Effect.gen(function* () { + const cut = cutMidTurn(); + for (const providerTurns of [ + cut.providerTurns, + // Pre-turn compaction reports usage before Codex records the new request. + cut.providerTurns.map((turn) => ({ ...turn, tokenUsage: { usedTokens: 1 } })), + ]) { + const dispatches = yield* continuationDispatches({ + ...cut, + messages: [cutRequest], + providerTurns, + providerThreads: cut.providerThreads.map((thread) => ({ + ...thread, + driver: ProviderDriverKind.make(driverKind), + })), + } as unknown as OrchestrationV2ThreadProjection); + assert.deepEqual( + dispatches.map((command) => [command.text, command.attachments, command.context]), + [[cutRequest.text, cutRequest.attachments, cutRequest.context]], + ); + } + }), ); it.effect("resumes natively once Codex recorded the request, and for other providers", () => @@ -1134,6 +1140,16 @@ it.effect("resumes natively once Codex recorded the request, and for other provi providerTurns: cut.providerTurns.map((turn) => ({ ...turn, inputRecordedAt })), } as unknown as OrchestrationV2ThreadProjection; assert.deepEqual(yield* continuationTexts(recorded), ["Continue where you left off."]); + assert.deepEqual( + yield* continuationTexts({ + ...recorded, + providerThreads: recorded.providerThreads.map((thread) => ({ + ...thread, + driver: ProviderDriverKind.make("acpRegistry"), + })), + }), + ["Continue where you left off."], + ); // A goal's first native turn recorded the request; its cut follow-up turn records none. const goalFollowUp = { ...cut, diff --git a/apps/server/src/orchestration-v2/RestartContinuation.ts b/apps/server/src/orchestration-v2/RestartContinuation.ts index 45ba18822180d..172915019574f 100644 --- a/apps/server/src/orchestration-v2/RestartContinuation.ts +++ b/apps/server/src/orchestration-v2/RestartContinuation.ts @@ -307,12 +307,14 @@ export const continueRestartedRun = Effect.fn("RestartContinuation.continueResta ); const noteText = note.work.length === 0 ? undefined : restartCancelledBackgroundWorkNote(note.work); - // Codex resumes natively with no input, so a request it never recorded must be sent again. + const driver = projection.providerThreads.find( + (thread) => thread.id === source.providerThreadId, + )?.driver; + // Native history may lack a prompt the provider never acknowledged. const resent = !note.settled && sourceMessage !== undefined && - projection.providerThreads.find((thread) => thread.id === source.providerThreadId)?.driver === - "codex" && + (driver === "codex" || driver === "acpRegistry") && restartSourceInputUnrecorded(source, projection.providerTurns, projection.runs) ? sourceMessage : undefined; diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index bd2552ab42183..f9f39935a28bd 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -1112,6 +1112,8 @@ interface ActiveAcpTurn { readonly nativeTurnId: string; readonly startedAt: DateTime.Utc; readonly completed: Deferred.Deferred; + readonly promptDispatched: Deferred.Deferred; + inputRecordedAt?: DateTime.Utc; readonly itemOrdinals: Map; readonly outgoingPrompt?: ReadonlyArray | undefined; consumedPromptMessageId?: string; @@ -4606,6 +4608,31 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( ) { const context = yield* Ref.get(activeTurn); const update = notification.update; + // Session metadata and child output do not acknowledge this root prompt. + if ( + context !== null && + !context.finalized && + context.inputRecordedAt === undefined && + notification.sessionId === context.nativeThreadId && + (yield* Deferred.isDone(context.promptDispatched)) && + (update.sessionUpdate === "user_message" || + update.sessionUpdate === "user_message_chunk" || + (update.sessionUpdate !== "compaction_update" && + update.sessionUpdate !== "compaction_summary_chunk" && + acpRootSessionUpdateIngestsOutput(notification))) + ) { + context.inputRecordedAt = yield* DateTime.now; + const turn = providerTurnPayload(context, "running", null); + yield* Ref.update(providerTurns, (current) => + new Map(current).set(String(turn.id), turn), + ); + yield* emitProviderEvent({ + type: "provider_turn.updated", + driver, + threadId: context.input.threadId, + providerTurn: turn, + }); + } if ( followsAgentWorkState && flavor.extractSubagentSessionEvent !== undefined && @@ -7043,6 +7070,9 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( status, startedAt: context.startedAt, completedAt, + ...(context.inputRecordedAt === undefined + ? {} + : { inputRecordedAt: context.inputRecordedAt }), ...(status === "running" || status === "pending" ? {} : { @@ -7572,6 +7602,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( nativeTurnId, startedAt, completed, + promptDispatched: yield* Deferred.make(), itemOrdinals: new Map(), outgoingPrompt: (yield* Ref.get(activeSessionSetup))?.initializeResult.protocolVersion === 2 @@ -7739,7 +7770,8 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( return; } const promptGeneration = yield* Ref.get(runtimeCallbackGeneration); - yield* runtime.prompt({ prompt: promptParts!.prompt }).pipe( + const promptOptions = { dispatched: context.promptDispatched }; + yield* runtime.prompt({ prompt: promptParts!.prompt }, promptOptions).pipe( Effect.tap(() => Ref.update(promptInstructionStates, (current) => { if (promptParts?.instructionState === undefined) return current; From a6e111fb8aa0f8731698ab122657026e6eca7245 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 13 Nov 2026 00:13:15 +0000 Subject: [PATCH 5/7] fix(acp): a root tool call confirms the prompt was accepted An ACP agent can start and finish a tool before it writes any text, so restart recovery resent a request whose action had already run. A new root tool_call now counts as acceptance; updates to earlier tools still do not. Co-Authored-By: Claude Opus 5.5 --- .../Adapters/AcpRegistryAdapterV2.test.ts | 171 +++++++++++------- packages/provider-acp/src/server/adapter.ts | 6 +- 2 files changed, 109 insertions(+), 68 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/AcpRegistryAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpRegistryAdapterV2.test.ts index 351999ad9df41..ff2cbe703b37f 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpRegistryAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpRegistryAdapterV2.test.ts @@ -1392,7 +1392,7 @@ describe("AcpRegistryAdapterV2", () => { }).pipe(Effect.provide(layerTest), Effect.scoped), ); - it.effect.each(["user_message", "agent_message_chunk"] as const)( + it.effect.each(["user_message", "agent_message_chunk", "tool_call"] as const)( "confirms input only after a root %s, and keeps it when the turn settles", (sessionUpdate) => Effect.gen(function* () { @@ -1401,72 +1401,111 @@ describe("AcpRegistryAdapterV2", () => { sessionUpdate: "session_info_update", title, }); - const session = yield* openClaudeReplaySession([ - ...claudeFrames.open, - claudeFrames.prompt(1)[0], - claudeFrames.update(CLAUDE_ROOT_SESSION, { - sessionUpdate: "usage_update", - used: 1, - size: 100, - }), - claudeFrames.update("unrelated-child", { - sessionUpdate: "agent_message_chunk", - content: { type: "text", text: "Earlier background work" }, - }), - claudeFrames.update(CLAUDE_ROOT_SESSION, { - sessionUpdate: "compaction_update", - compactionId: "before-input", - status: "running", - }), - claudeFrames.update(CLAUDE_ROOT_SESSION, { - sessionUpdate: "user_message_chunk", - content: { - type: "text", - text: 'completed', - }, - }), - claudeFrames.update(CLAUDE_ROOT_SESSION, { - sessionUpdate: "user_message", - messageId: "previous-task-notice", - content: [{ type: "text", text: "Previous task completed." }], - }), - claudeFrames.update(CLAUDE_ROOT_SESSION, { - sessionUpdate: "tool_call_update", - toolCallId: "previous-task-tool", - status: "completed", - }), - marker("before acceptance"), - claudeFrames.update(CLAUDE_ROOT_SESSION, { - sessionUpdate, - ...(sessionUpdate === "user_message" - ? { - messageId: "accepted", - content: [ - { - type: "text", - text: t3AcpPromptWithInstructions({ - prompt: "test prompt", - state: { - interactionMode: "default", - hasT3Mcp: false, - hasDeviceTools: false, - }, - }), - }, - { - type: "text", - text: buildRuntimeInstructions({ harness: "acpRegistry", model: "default" }), + // ACP v2 has no tool_call start (tools start with tool_call_update), so that row runs on v1. + const v1 = sessionUpdate === "tool_call"; + const open = claudeFrames.open.map((entry) => + v1 && entry.frame.method === "initialize" && entry.type === "emit_inbound" + ? { + ...entry, + frame: { + ...entry.frame, + result: { protocolVersion: 1, agentCapabilities: {}, authMethods: [] }, + }, + } + : entry, + ); + const session = yield* openClaudeReplaySession( + [ + ...open, + claudeFrames.prompt(1)[0], + claudeFrames.update(CLAUDE_ROOT_SESSION, { + sessionUpdate: "usage_update", + used: 1, + size: 100, + }), + claudeFrames.update("unrelated-child", { + sessionUpdate: "agent_message_chunk", + content: { type: "text", text: "Earlier background work" }, + }), + claudeFrames.update(CLAUDE_ROOT_SESSION, { + sessionUpdate: "compaction_update", + compactionId: "before-input", + status: "running", + }), + claudeFrames.update(CLAUDE_ROOT_SESSION, { + sessionUpdate: "user_message_chunk", + content: { + type: "text", + text: 'completed', + }, + }), + claudeFrames.update(CLAUDE_ROOT_SESSION, { + sessionUpdate: "user_message", + messageId: "previous-task-notice", + content: [{ type: "text", text: "Previous task completed." }], + }), + claudeFrames.update(CLAUDE_ROOT_SESSION, { + sessionUpdate: "tool_call_update", + toolCallId: "previous-task-tool", + status: "completed", + }), + marker("before acceptance"), + claudeFrames.update(CLAUDE_ROOT_SESSION, { + sessionUpdate, + ...(sessionUpdate === "user_message" + ? { + messageId: "accepted", + content: [ + { + type: "text", + text: t3AcpPromptWithInstructions({ + prompt: "test prompt", + state: { + interactionMode: "default", + hasT3Mcp: false, + hasDeviceTools: false, + }, + }), + }, + { + type: "text", + text: buildRuntimeInstructions({ + harness: "acpRegistry", + model: "default", + }), + }, + ], + } + : sessionUpdate === "tool_call" + ? { + toolCallId: "append-action-log", + title: "Append to actions.log", + kind: "execute", + status: "pending", + rawInput: { command: "echo done >> actions.log" }, + } + : { content: { type: "text", text: "Working on the request" } }), + }), + marker("after acceptance"), + ...(v1 + ? [ + { + type: "emit_inbound", + frame: { + kind: "response", + method: "session/prompt", + result: { stopReason: "end_turn" }, }, - ], - } - : { content: { type: "text", text: "Working on the request" } }), - }), - marker("after acceptance"), - ...claudeFrames.prompt(1).slice(1), - claudeFrames.idle("end_turn"), - claudeFrames.closeRequest, - claudeFrames.closed, - ]); + }, + ] + : [...claudeFrames.prompt(1).slice(1), claudeFrames.idle("end_turn")]), + claudeFrames.closeRequest, + claudeFrames.closed, + ], + undefined, + !v1, + v1 ? "devin" : undefined, + ); yield* session.runtime.startTurn(session.turnInput(1)); const before = yield* session.throughMarker("before acceptance"); const initial = before.flatMap((event) => diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index d032c09d81e0b..9ea9ad7490912 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -5061,7 +5061,8 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( ) { const context = yield* Ref.get(activeTurn); const update = notification.update; - // Session metadata and child output do not acknowledge this root prompt. + // Session metadata, child output, and updates to earlier tools do not acknowledge this + // root prompt; a new root tool call does, since the agent may act before it writes text. if ( context !== null && !context.finalized && @@ -5074,7 +5075,8 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( ((update.sessionUpdate === "agent_message_chunk" || update.sessionUpdate === "agent_thought_chunk" || update.sessionUpdate === "agent_message" || - update.sessionUpdate === "agent_thought") && + update.sessionUpdate === "agent_thought" || + update.sessionUpdate === "tool_call") && acpPostSettleWakeEvidence(notification, flavor))) ) { context.inputRecordedAt = yield* DateTime.now; From b2b7f847b5ac781968ffcb6555d23ed15060be23 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 13 Nov 2026 00:27:20 +0000 Subject: [PATCH 6/7] fix(server): resend only requests the provider explicitly left unconfirmed Turns from servers that predate acceptance tracking have no inputRecordedAt, which read as unconfirmed, so the first restart after upgrading resent requests that may already have run. Adapters now stamp inputRecordedAt as null when they create a turn, and restart recovery resends only when every turn in the chain is explicitly unconfirmed; turns without the field keep the cautious continuation prompt. ACP v2 starts a tool with its first tool_call_update, so a pending or in-progress update for a tool the session has not seen before now also confirms the prompt. Updates to earlier tools still do not. Co-Authored-By: Claude Opus 5.5 --- .../Adapters/AcpRegistryAdapterV2.test.ts | 6 +- .../Adapters/CodexAdapterV2.ts | 1 + .../src/orchestration-v2/ProjectionStore.ts | 13 +- .../orchestration-v2/RestartBackgroundNote.ts | 5 +- .../RestartContinuation.test.ts | 113 +++++++++++------- ...estratorReplayRecovery.integration.test.ts | 2 +- .../src/state/orchestrationV2Projection.ts | 6 +- packages/contracts/src/orchestrationV2.ts | 15 ++- packages/provider-acp/src/server/adapter.ts | 21 +++- 9 files changed, 117 insertions(+), 65 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/AcpRegistryAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpRegistryAdapterV2.test.ts index ff2cbe703b37f..1bfc54217b8cb 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpRegistryAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpRegistryAdapterV2.test.ts @@ -1392,7 +1392,7 @@ describe("AcpRegistryAdapterV2", () => { }).pipe(Effect.provide(layerTest), Effect.scoped), ); - it.effect.each(["user_message", "agent_message_chunk", "tool_call"] as const)( + it.effect.each(["user_message", "agent_message_chunk", "tool_call", "tool_call_update"] as const)( "confirms input only after a root %s, and keeps it when the turn settles", (sessionUpdate) => Effect.gen(function* () { @@ -1476,7 +1476,7 @@ describe("AcpRegistryAdapterV2", () => { }, ], } - : sessionUpdate === "tool_call" + : sessionUpdate === "tool_call" || sessionUpdate === "tool_call_update" ? { toolCallId: "append-action-log", title: "Append to actions.log", @@ -1517,7 +1517,7 @@ describe("AcpRegistryAdapterV2", () => { event.type === "provider_turn.updated" ? [event.providerTurn] : [], ); assert.isNotEmpty(initial); - assert.isTrue(initial.every((turn) => turn.inputRecordedAt === undefined)); + assert.isTrue(initial.every((turn) => turn.inputRecordedAt === null)); const accepted = acceptedEvents.find((event) => event.type === "provider_turn.updated"); const acceptedTurn = accepted?.type === "provider_turn.updated" ? accepted.providerTurn : undefined; diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 5dc3d8d7a11b3..ce13468c34000 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -2337,6 +2337,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( status: "running", startedAt: input.startedAt, completedAt: null, + inputRecordedAt: null, }, }); yield* rememberRootProviderThread(input.turnInput.providerThread); diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 2b9f9ef783b9c..ba2cc9278c6b7 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -1,5 +1,5 @@ import type { ProjectThreadAwarenessV2Input } from "@t3tools/shared/agentAwareness"; -import { pendingGuestMessageCount } from "@t3tools/contracts"; +import { mergeProviderTurnInputRecordedAt, pendingGuestMessageCount } from "@t3tools/contracts"; import { applyThreadPullRequestLinkUpdate } from "../sourceControl/threadPullRequests.ts"; import { projectTurnItemForWire } from "./WireProjection.ts"; import * as Stream from "effect/Stream"; @@ -794,14 +794,16 @@ function upsertProviderTurn( next: OrchestrationV2ProviderTurn, ): Array { const current = turns.find((turn) => turn.id === next.id); + const inputRecordedAt = mergeProviderTurnInputRecordedAt( + current?.inputRecordedAt, + next.inputRecordedAt, + ); return upsertById(turns, { ...next, ...((next.tokenUsage ?? current?.tokenUsage) === undefined ? {} : { tokenUsage: next.tokenUsage ?? current?.tokenUsage }), - ...((next.inputRecordedAt ?? current?.inputRecordedAt) === undefined - ? {} - : { inputRecordedAt: next.inputRecordedAt ?? current?.inputRecordedAt }), + ...(inputRecordedAt === undefined ? {} : { inputRecordedAt }), }); } @@ -2682,8 +2684,7 @@ export const layer: Layer.Layer = } case "provider-turn.updated": { const existingRows = - event.payload.tokenUsage === undefined || - event.payload.inputRecordedAt === undefined + event.payload.tokenUsage === undefined || event.payload.inputRecordedAt == null ? yield* sql` SELECT payload_json FROM orchestration_v2_projection_provider_turns diff --git a/apps/server/src/orchestration-v2/RestartBackgroundNote.ts b/apps/server/src/orchestration-v2/RestartBackgroundNote.ts index 6764a0b977e9e..01e00a8ff98db 100644 --- a/apps/server/src/orchestration-v2/RestartBackgroundNote.ts +++ b/apps/server/src/orchestration-v2/RestartBackgroundNote.ts @@ -307,6 +307,7 @@ export function isRestartNoteContinuation( /** * Without the provider's confirmation, the native history may lack the request: Codex runs prompt * hooks and pre-turn compaction first. Native continuations inherit the source request's confirmation. + * Turns from servers that did not track it (no inputRecordedAt) count as possibly recorded. */ export function restartSourceInputUnrecorded( source: OrchestrationV2Run, @@ -322,9 +323,7 @@ export function restartSourceInputUnrecorded( const attemptId = current.activeAttemptId; if ( attemptId !== null && - providerTurns.some( - (turn) => turn.runAttemptId === attemptId && turn.inputRecordedAt !== undefined, - ) + providerTurns.some((turn) => turn.runAttemptId === attemptId && turn.inputRecordedAt !== null) ) return false; const previousId: OrchestrationV2Run["restartContinuationOfRunId"] = diff --git a/apps/server/src/orchestration-v2/RestartContinuation.test.ts b/apps/server/src/orchestration-v2/RestartContinuation.test.ts index 13c0074af7d92..de323d57ec1f4 100644 --- a/apps/server/src/orchestration-v2/RestartContinuation.test.ts +++ b/apps/server/src/orchestration-v2/RestartContinuation.test.ts @@ -1137,10 +1137,11 @@ it.effect.each(["codex", "acpRegistry"] as const)( (driverKind) => Effect.gen(function* () { const cut = cutMidTurn(); + const unconfirmed = cut.providerTurns.map((turn) => ({ ...turn, inputRecordedAt: null })); for (const providerTurns of [ - cut.providerTurns, + unconfirmed, // Pre-turn compaction reports usage before Codex records the new request. - cut.providerTurns.map((turn) => ({ ...turn, tokenUsage: { usedTokens: 1 } })), + unconfirmed.map((turn) => ({ ...turn, tokenUsage: { usedTokens: 1 } })), ]) { const dispatches = yield* continuationDispatches({ ...cut, @@ -1159,47 +1160,73 @@ it.effect.each(["codex", "acpRegistry"] as const)( }), ); -it.effect("resumes natively once Codex recorded the request, and for other providers", () => - Effect.gen(function* () { - const cut = cutMidTurn(); - const inputRecordedAt = DateTime.makeUnsafe("2026-10-03T11:59:00.000Z"); - const recorded = { - ...cut, - messages: [cutRequest], - providerTurns: cut.providerTurns.map((turn) => ({ ...turn, inputRecordedAt })), - } as unknown as OrchestrationV2ThreadProjection; - const recordedAcp = { - ...recorded, - providerThreads: recorded.providerThreads.map((thread) => ({ - ...thread, - driver: ProviderDriverKind.make("acpRegistry"), - })), - }; - // A goal's first native turn recorded the request; its cut follow-up turn records none. - const goalFollowUp = { - ...cut, - messages: [ - { ...cutRequest, text: "/goal find the cheapest return flights", attachments: [] }, - ], - providerTurns: [ - { ...cut.providerTurns[0]!, ordinal: 1, status: "completed", inputRecordedAt }, - { ...cut.providerTurns[0]!, id: "turn:goal-follow-up", ordinal: 2 }, - ], - } as unknown as OrchestrationV2ThreadProjection; - const claude = ProviderDriverKind.make("claudeAgent"); - const otherDriver = { - ...cut, - messages: [cutRequest], - providerThreads: cut.providerThreads.map((thread) => ({ ...thread, driver: claude })), - } as unknown as OrchestrationV2ThreadProjection; - for (const projection of [recorded, recordedAcp, goalFollowUp, otherDriver]) { - const commands = yield* continuationDispatches(projection); - assert.lengthOf(commands, 1); - assert.notInclude(commands[0]!.text, cutRequest.text); - assert.deepEqual(commands[0]!.attachments, []); - assert.isUndefined(commands[0]!.context); - } - }), +it.effect( + "resumes natively once recorded, for turns from before tracking, and for other providers", + () => + Effect.gen(function* () { + const cut = cutMidTurn(); + const inputRecordedAt = DateTime.makeUnsafe("2026-10-03T11:59:00.000Z"); + const recorded = { + ...cut, + messages: [cutRequest], + providerTurns: cut.providerTurns.map((turn) => ({ ...turn, inputRecordedAt })), + } as unknown as OrchestrationV2ThreadProjection; + const recordedAcp = { + ...recorded, + providerThreads: recorded.providerThreads.map((thread) => ({ + ...thread, + driver: ProviderDriverKind.make("acpRegistry"), + })), + }; + // A goal's first native turn recorded the request; its cut follow-up turn records none. + const goalFollowUp = { + ...cut, + messages: [ + { ...cutRequest, text: "/goal find the cheapest return flights", attachments: [] }, + ], + providerTurns: [ + { ...cut.providerTurns[0]!, ordinal: 1, status: "completed", inputRecordedAt }, + { + ...cut.providerTurns[0]!, + id: "turn:goal-follow-up", + ordinal: 2, + inputRecordedAt: null, + }, + ], + } as unknown as OrchestrationV2ThreadProjection; + // Servers before acceptance tracking left the field off, so the request may already have run. + const legacy = { + ...cut, + messages: [cutRequest], + } as unknown as OrchestrationV2ThreadProjection; + const legacyAcp = { + ...legacy, + providerThreads: legacy.providerThreads.map((thread) => ({ + ...thread, + driver: ProviderDriverKind.make("acpRegistry"), + })), + }; + const claude = ProviderDriverKind.make("claudeAgent"); + const otherDriver = { + ...cut, + messages: [cutRequest], + providerThreads: cut.providerThreads.map((thread) => ({ ...thread, driver: claude })), + } as unknown as OrchestrationV2ThreadProjection; + for (const projection of [ + recorded, + recordedAcp, + goalFollowUp, + legacy, + legacyAcp, + otherDriver, + ]) { + const commands = yield* continuationDispatches(projection); + assert.lengthOf(commands, 1); + assert.notInclude(commands[0]!.text, cutRequest.text); + assert.deepEqual(commands[0]!.attachments, []); + assert.isUndefined(commands[0]!.context); + } + }), ); it.effect( diff --git a/apps/server/src/orchestration-v2/testkit/OrchestratorReplayRecovery.integration.test.ts b/apps/server/src/orchestration-v2/testkit/OrchestratorReplayRecovery.integration.test.ts index ccc5afa785f22..479c674c7ede9 100644 --- a/apps/server/src/orchestration-v2/testkit/OrchestratorReplayRecovery.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/OrchestratorReplayRecovery.integration.test.ts @@ -555,7 +555,7 @@ describe("orchestrator replay recovery", () => { // Codex confirmed the resent request, and later usage updates kept that confirmation. assert.deepEqual( projection.providerTurns.map((turn) => [ - turn.inputRecordedAt !== undefined, + turn.inputRecordedAt instanceof Object, turn.tokenUsage !== undefined, ]), [ diff --git a/packages/client-runtime/src/state/orchestrationV2Projection.ts b/packages/client-runtime/src/state/orchestrationV2Projection.ts index acbacb0388f72..4be25f61316b1 100644 --- a/packages/client-runtime/src/state/orchestrationV2Projection.ts +++ b/packages/client-runtime/src/state/orchestrationV2Projection.ts @@ -1,4 +1,5 @@ import { applyThreadPullRequestLinkUpdate } from "../threadPullRequests.ts"; +import { mergeProviderTurnInputRecordedAt } from "@t3tools/contracts"; import type { OrchestrationV2DomainEvent, OrchestrationV2Run, @@ -380,7 +381,10 @@ export function applyOrchestrationV2ProjectionEvent( case "provider-turn.updated": { const current = base.providerTurns.find((turn) => turn.id === event.payload.id); const tokenUsage = event.payload.tokenUsage ?? current?.tokenUsage; - const inputRecordedAt = event.payload.inputRecordedAt ?? current?.inputRecordedAt; + const inputRecordedAt = mergeProviderTurnInputRecordedAt( + current?.inputRecordedAt, + event.payload.inputRecordedAt, + ); return { ...base, providerTurns: upsertEntity(base.providerTurns, { diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 371b79356e30c..64d623a1997b6 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -1216,8 +1216,11 @@ export const OrchestrationV2ProviderTurn = Schema.Struct({ completedAt: Schema.NullOr(Schema.DateTimeUtc), tokenUsage: Schema.optional(OrchestrationV2ProviderTurnTokenUsage), turnTokenUsage: Schema.optional(TurnTokenUsage), - /** When the provider confirmed it recorded the turn's user input in its native history. */ - inputRecordedAt: Schema.optional(Schema.DateTimeUtc), + /** + * When the provider confirmed it recorded the turn's user input in its native history. Null while + * unconfirmed; absent on turns from servers that did not track it, whose acceptance is unknown. + */ + inputRecordedAt: Schema.optional(Schema.NullOr(Schema.DateTimeUtc)), /** The model the provider reported running this turn's root agent, which can differ from the requested one. */ model: Schema.optional(TrimmedNonEmptyString), }); @@ -2492,11 +2495,17 @@ export const OrchestrationV2ContextTransferJson = OrchestrationV2ContextTransfer ); export type OrchestrationV2ContextTransferJson = typeof OrchestrationV2ContextTransferJson.Type; +/** A confirmation survives later updates that omit it or still report the turn as unconfirmed. */ +export const mergeProviderTurnInputRecordedAt = ( + current: T | null | undefined, + next: T | null | undefined, +): T | null | undefined => next ?? current ?? (next === null ? null : current); + export const OrchestrationV2ProviderTurnJson = OrchestrationV2ProviderTurn.mapFields((fields) => ({ ...fields, startedAt: Schema.NullOr(Schema.DateTimeUtcFromString), completedAt: Schema.NullOr(Schema.DateTimeUtcFromString), - inputRecordedAt: Schema.optional(Schema.DateTimeUtcFromString), + inputRecordedAt: Schema.optional(Schema.NullOr(Schema.DateTimeUtcFromString)), })); export type OrchestrationV2ProviderTurnJson = typeof OrchestrationV2ProviderTurnJson.Type; diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index 9ea9ad7490912..472ac658b826a 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -1952,6 +1952,8 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( input.runtimePolicy; const events = yield* Queue.unbounded(); const activeTurn = yield* Ref.make(null); + // Spans turns, so a late update to an earlier turn's tool is not taken for a new one. + const seenToolCallIds = new Set(); // Keyed by the agent's steer id; an injection can arrive before its send returns. const steerInjections = new Map>(); const steerInjection = (steerId: string) => { @@ -5061,8 +5063,19 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( ) { const context = yield* Ref.get(activeTurn); const update = notification.update; + // ACP v2 starts a tool with its first tool_call_update rather than a tool_call. + let startsTool = false; + if (update.sessionUpdate === "tool_call" || update.sessionUpdate === "tool_call_update") { + const key = `${notification.sessionId}\u0000${update.toolCallId}`; + startsTool = + !seenToolCallIds.has(key) && + (update.sessionUpdate === "tool_call" || + update.status === "pending" || + update.status === "in_progress"); + seenToolCallIds.add(key); + } // Session metadata, child output, and updates to earlier tools do not acknowledge this - // root prompt; a new root tool call does, since the agent may act before it writes text. + // root prompt; a new root tool does, since the agent may act before it writes text. if ( context !== null && !context.finalized && @@ -5076,7 +5089,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( update.sessionUpdate === "agent_thought_chunk" || update.sessionUpdate === "agent_message" || update.sessionUpdate === "agent_thought" || - update.sessionUpdate === "tool_call") && + startsTool) && acpPostSettleWakeEvidence(notification, flavor))) ) { context.inputRecordedAt = yield* DateTime.now; @@ -7645,9 +7658,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( status, startedAt: context.startedAt, completedAt, - ...(context.inputRecordedAt === undefined - ? {} - : { inputRecordedAt: context.inputRecordedAt }), + inputRecordedAt: context.inputRecordedAt ?? null, ...(status === "running" || status === "pending" ? {} : { From dd79155cc1d8e2df969fb68db9949fc452ea8827 Mon Sep 17 00:00:00 2001 From: Nathan Wong <20381195+thelogicmatrix@users.noreply.github.com> Date: Sun, 15 Nov 2026 14:10:32 +0800 Subject: [PATCH 7/7] test(server): include provider threads in recovery executor fixture --- apps/server/src/orchestration-v2/EffectWorker.test.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/apps/server/src/orchestration-v2/EffectWorker.test.ts b/apps/server/src/orchestration-v2/EffectWorker.test.ts index 94444eb8e5c69..56820f16d62f4 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.test.ts +++ b/apps/server/src/orchestration-v2/EffectWorker.test.ts @@ -813,6 +813,7 @@ it.effect.each([3, 4])( }, ], messages: [], + providerThreads: [], providerTurns: [], runtimeRequests: [], turnItems: [],