diff --git a/apps/server/src/orchestration-v2/Adapters/AcpRegistryAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpRegistryAdapterV2.test.ts
index 97fb6e636054f..5f43a45a05941 100644
--- a/apps/server/src/orchestration-v2/Adapters/AcpRegistryAdapterV2.test.ts
+++ b/apps/server/src/orchestration-v2/Adapters/AcpRegistryAdapterV2.test.ts
@@ -1436,6 +1436,142 @@ describe("AcpRegistryAdapterV2", () => {
}).pipe(Effect.provide(layerTest)),
);
+ 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* () {
+ const marker = (title: string) =>
+ claudeFrames.update(CLAUDE_ROOT_SESSION, {
+ sessionUpdate: "session_info_update",
+ title,
+ });
+ // 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" || sessionUpdate === "tool_call_update"
+ ? {
+ 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" },
+ },
+ },
+ ]
+ : [...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) =>
+ 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 === null));
+ 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("the agent process env names the thread it serves", () =>
Effect.gen(function* () {
const selfInvocation = yield* resolveSelfInvocation();
diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts
index ee35f0a61671a..96b5b8f2e7edf 100644
--- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts
+++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts
@@ -2510,6 +2510,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* (
status: "running",
startedAt: input.startedAt,
completedAt: null,
+ inputRecordedAt: null,
},
});
yield* rememberRootProviderThread(input.turnInput.providerThread);
@@ -4941,6 +4942,32 @@ 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,
+ },
+ });
// A run's first request can 429 before Codex pushes a codex snapshot.
const readUsageLimitReset = Effect.gen(function* () {
const sent = yield* Ref.get(rateLimitSnapshot);
@@ -4973,29 +5000,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),
);
@@ -5533,6 +5539,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/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: [],
diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts
index 9a7e4fcb64490..a44f7844c2a22 100644
--- a/apps/server/src/orchestration-v2/ProjectionStore.ts
+++ b/apps/server/src/orchestration-v2/ProjectionStore.ts
@@ -1,5 +1,9 @@
import type { ProjectThreadAwarenessV2Input } from "@t3tools/shared/agentAwareness";
-import { applyOrchestrationV2ThreadVisit, pendingGuestMessageCount } from "@t3tools/contracts";
+import {
+ applyOrchestrationV2ThreadVisit,
+ mergeProviderTurnInputRecordedAt,
+ pendingGuestMessageCount,
+} from "@t3tools/contracts";
import { applyThreadPullRequestLinkUpdate } from "../sourceControl/threadPullRequests.ts";
import { projectTurnItemForWire } from "./WireProjection.ts";
import * as Stream from "effect/Stream";
@@ -842,11 +846,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 }),
+ ...(inputRecordedAt === undefined ? {} : { inputRecordedAt }),
});
}
@@ -2745,7 +2754,7 @@ export const layer: Layer.Layer =
}
case "provider-turn.updated": {
const existingRows =
- event.payload.tokenUsage === 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/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts
index 6587e742c7cd0..9372be0a4a40b 100644
--- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts
+++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts
@@ -69,6 +69,7 @@ import * as RuntimePolicy from "./RuntimePolicy.ts";
import * as TurnItemPositionStore from "./TurnItemPositionStore.ts";
import {
isRestartNoteContinuation,
+ isRestartResendContinuation,
pendingRestartCancelledBackgroundWork,
restartCancelledBackgroundWorkNote,
} from "./RestartBackgroundNote.ts";
@@ -848,6 +849,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,
@@ -1737,11 +1742,14 @@ export const layer: Layer.Layer<
.join("\n\n");
const { restartContinuationOfRunId: _resumedRunId, ...promptedInput } = turnInput;
const text = goalCommand ? userText : withInlineContext(context, userText);
- const startInput = noteContinuation ? promptedInput : turnInput;
+ const startInput = promptedContinuation ? promptedInput : turnInput;
const { resume: _resume, ...contextInput } = startInput;
const { message: resumeMessage, resume: _resumeIntent, ...resumeInput } = startInput;
const operation =
- turnInput.resume && context === "" && session.resumeTurn !== undefined
+ !promptedContinuation &&
+ turnInput.resume &&
+ context === "" &&
+ session.resumeTurn !== undefined
? session.resumeTurn({ ...resumeInput, messageId: resumeMessage.messageId })
: start({
...(context === "" ? startInput : contextInput),
@@ -1811,7 +1819,7 @@ export const layer: Layer.Layer<
retryHandoff.length === 0 &&
sideChatHandoff === undefined &&
restartNote === "" &&
- !noteContinuation
+ !promptedContinuation
? session
: makeDeliverySession(session, startWithHandoffs);
return runExecution.startRootRun({
diff --git a/apps/server/src/orchestration-v2/RestartBackgroundNote.test.ts b/apps/server/src/orchestration-v2/RestartBackgroundNote.test.ts
index 90da48e32de16..5be3b2a24b65a 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 23a95bc2bd20f..645ebe8b47914 100644
--- a/apps/server/src/orchestration-v2/RestartBackgroundNote.ts
+++ b/apps/server/src/orchestration-v2/RestartBackgroundNote.ts
@@ -364,6 +364,49 @@ 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,
+ providerTurns: ReadonlyArray<
+ Pick
+ >,
+ runs: ReadonlyArray,
+): boolean {
+ 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 !== null)
+ )
+ return false;
+ const previousId: OrchestrationV2Run["restartContinuationOfRunId"] =
+ current.restartContinuationOfRunId;
+ current = runs.find(
+ (run) => run.id === previousId && run.providerThreadId === source.providerThreadId,
+ );
+ }
+ return true;
+}
+
+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, runs);
+}
+
/** 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 77af7d2680e05..dbbafb861224e 100644
--- a/apps/server/src/orchestration-v2/RestartContinuation.test.ts
+++ b/apps/server/src/orchestration-v2/RestartContinuation.test.ts
@@ -121,7 +121,6 @@ it("requires matching saved native state for an unfinished root run", () => {
},
],
},
- { ...projection, providerTurns: [] },
...[
"queued",
"preparing",
@@ -136,6 +135,30 @@ it("requires matching saved native state for an unfinished root run", () => {
assert.isUndefined(restartContinuationRun(invalid as OrchestrationV2ThreadProjection, now));
});
+it("recovers a running Codex request before its first provider turn", () => {
+ const projection = makeProjection();
+ assert.equal(restartContinuationRun({ ...projection, providerTurns: [] }, now)?.id, runId);
+ const claude = ProviderDriverKind.make("claudeAgent");
+ assert.isUndefined(
+ restartContinuationRun(
+ {
+ ...projection,
+ providerTurns: [],
+ providerThreads: projection.providerThreads.map((thread) => ({
+ ...thread,
+ driver: claude,
+ nativeThreadRef: { ...thread.nativeThreadRef!, driver: claude },
+ })),
+ providerSessions: projection.providerSessions.map((session) => ({
+ ...session,
+ driver: claude,
+ })),
+ },
+ now,
+ ),
+ );
+});
+
it("continues a live turn whose session the adapter never marked running", () => {
const projection = makeProjection();
const withSessionStatus = (status: string) =>
@@ -1247,6 +1270,118 @@ 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.each(["codex", "acpRegistry"] as const)(
+ "resends a %s request and its image before the provider accepted it",
+ (driverKind) =>
+ Effect.gen(function* () {
+ const cut = cutMidTurn();
+ const unconfirmed = cut.providerTurns.map((turn) => ({ ...turn, inputRecordedAt: null }));
+ for (const providerTurns of [
+ unconfirmed,
+ // Pre-turn compaction reports usage before Codex records the new request.
+ unconfirmed.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 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(
"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 e8db8055ab69a..e3c3345f8f60b 100644
--- a/apps/server/src/orchestration-v2/RestartContinuation.ts
+++ b/apps/server/src/orchestration-v2/RestartContinuation.ts
@@ -33,6 +33,7 @@ import {
neverStarted,
restartCancelledBackgroundWorkNote,
restartContinuationNote,
+ restartSourceInputUnrecorded,
} from "./RestartBackgroundNote.ts";
const CONTINUE_PROMPT =
@@ -220,14 +221,15 @@ export function restartContinuationRun(
)
return;
if (settled) return hasOpenAwaitedWork(projection, providerThread) ? run : undefined;
+ const attemptTurns = projection.providerTurns.filter(
+ (turn) =>
+ turn.providerThreadId === providerThread.id && turn.runAttemptId === run.activeAttemptId,
+ );
+ const codexBeforeTurn = providerThread.driver === "codex" && attemptTurns.length === 0;
if (
liveTurnRequired &&
- !projection.providerTurns.some(
- (turn) =>
- turn.providerThreadId === providerThread.id &&
- turn.runAttemptId === run.activeAttemptId &&
- turn.status === "running",
- )
+ !codexBeforeTurn &&
+ !attemptTurns.some((turn) => turn.status === "running")
)
return;
return run;
@@ -246,7 +248,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 (
@@ -318,12 +328,20 @@ export const continueRestartedRun = Effect.fn("RestartContinuation.continueResta
);
const noteText =
note.work.length === 0 ? undefined : restartCancelledBackgroundWorkNote(note.work);
+ 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 &&
+ (driver === "codex" || driver === "acpRegistry") &&
+ restartSourceInputUnrecorded(source, projection.providerTurns, projection.runs)
+ ? 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",
@@ -339,7 +357,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 e1ef5e5dc3fe3..80bc6078ab104 100644
--- a/apps/server/src/orchestration-v2/testkit/OrchestratorReplayRecovery.integration.test.ts
+++ b/apps/server/src/orchestration-v2/testkit/OrchestratorReplayRecovery.integration.test.ts
@@ -2,11 +2,13 @@ import * as NodeCrypto from "@effect/platform-node/NodeCrypto";
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,
@@ -28,6 +30,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 {
@@ -79,6 +82,75 @@ 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?: {
+ 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 =>
+ 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 === "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 } } }];
+ });
+ 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* () {
@@ -367,6 +439,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 instanceof Object,
+ 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 16301157a6a4e..2c68498f77c07 100644
--- a/packages/client-runtime/src/state/orchestrationV2Projection.test.ts
+++ b/packages/client-runtime/src/state/orchestrationV2Projection.test.ts
@@ -135,7 +135,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,
@@ -152,6 +152,7 @@ describe("applyOrchestrationV2ProjectionEvent", () => {
maxTokens: 200_000,
updatedAt: "2026-08-29T00:00:00.000Z",
},
+ inputRecordedAt: now,
};
const projection = { ...emptyProjection, providerTurns: [running] };
const event = {
@@ -165,6 +166,7 @@ describe("applyOrchestrationV2ProjectionEvent", () => {
status: "completed",
completedAt: now,
tokenUsage: undefined,
+ inputRecordedAt: undefined,
},
} as OrchestrationV2DomainEvent;
@@ -172,6 +174,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 37da408a1c35b..88041174e210c 100644
--- a/packages/client-runtime/src/state/orchestrationV2Projection.ts
+++ b/packages/client-runtime/src/state/orchestrationV2Projection.ts
@@ -2,6 +2,7 @@ import { applyThreadPullRequestLinkUpdate } from "../threadPullRequests.ts";
import {
applyOrchestrationV2ThreadVisit,
isQueuedProviderThreadPlaceholder,
+ mergeProviderTurnInputRecordedAt,
} from "@t3tools/contracts";
import type {
OrchestrationV2DomainEvent,
@@ -394,22 +395,22 @@ export function applyOrchestrationV2ProjectionEvent(
: base.thread,
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 = mergeProviderTurnInputRecordedAt(
+ current?.inputRecordedAt,
+ event.payload.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 44e9aa8290e9a..d1cbe86f91866 100644
--- a/packages/contracts/src/orchestrationV2.ts
+++ b/packages/contracts/src/orchestrationV2.ts
@@ -1237,6 +1237,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. 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),
});
@@ -2543,10 +2548,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.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 dc12bd7aa1e14..dced2e972d330 100644
--- a/packages/provider-acp/src/server/adapter.ts
+++ b/packages/provider-acp/src/server/adapter.ts
@@ -1463,6 +1463,7 @@ interface ActiveAcpTurn {
readonly nativeTurnId: string;
readonly startedAt: DateTime.Utc;
readonly completed: Deferred.Deferred;
+ inputRecordedAt?: DateTime.Utc;
readonly itemOrdinals: Map;
readonly outgoingPrompt?: ReadonlyArray | undefined;
consumedPromptMessageId?: string;
@@ -2034,6 +2035,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) => {
@@ -5288,6 +5291,47 @@ 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 does, since the agent may act before it writes text.
+ if (
+ context !== null &&
+ !context.finalized &&
+ context.inputRecordedAt === undefined &&
+ notification.sessionId === context.nativeThreadId &&
+ (yield* Deferred.isDone(context.promptDispatched)) &&
+ ((update.sessionUpdate === "user_message" &&
+ context.outgoingPrompt !== undefined &&
+ Equal.equals(update.content, context.outgoingPrompt)) ||
+ ((update.sessionUpdate === "agent_message_chunk" ||
+ update.sessionUpdate === "agent_thought_chunk" ||
+ update.sessionUpdate === "agent_message" ||
+ update.sessionUpdate === "agent_thought" ||
+ startsTool) &&
+ acpPostSettleWakeEvidence(notification, flavor)))
+ ) {
+ 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 &&
@@ -7899,6 +7943,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* (
status,
startedAt: context.startedAt,
completedAt,
+ inputRecordedAt: context.inputRecordedAt ?? null,
...(status === "running" || status === "pending"
? {}
: {