fix(server): resend unaccepted Codex and ACP requests after restart

apps/server/src/orchestration-v2/Adapters/AcpRegistryAdapterV2.test.ts

@@ -1308,6 +1308,103 @@ 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",
+          }),
+          claudeFrames.update(CLAUDE_ROOT_SESSION, {
+            sessionUpdate: "user_message_chunk",
+            content: {
+              type: "text",
+              text: '<monitor-event id="previous-task">completed</monitor-event>',
+            },
+          }),
+          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" }),
+                    },
+                  ],
+                }
+              : { 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("the agent process env names the thread it serves", () =>
     Effect.gen(function* () {
       const selfInvocation = yield* resolveSelfInvocation();

apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts

@@ -4742,6 +4742,32 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* (
           }).pipe(Effect.orDie),
         );
 
+        const updateRunningProviderTurn = (
+          context: ActiveCodexTurnContext,
+          nativeTurnId: string,
+          update: Pick<OrchestrationV2ProviderTurn, "tokenUsage" | "inputRecordedAt">,
+        ) =>
+          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);
@@ -4774,29 +4800,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),
         );
@@ -5316,6 +5321,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") {

apps/server/src/orchestration-v2/ProjectionStore.ts

@@ -778,6 +778,9 @@ function upsertProviderTurn(
     ...((next.tokenUsage ?? current?.tokenUsage) === undefined
       ? {}
       : { tokenUsage: next.tokenUsage ?? current?.tokenUsage }),
+    ...((next.inputRecordedAt ?? current?.inputRecordedAt) === undefined
+      ? {}
+      : { inputRecordedAt: next.inputRecordedAt ?? current?.inputRecordedAt }),
   });
 }
 
@@ -2620,7 +2623,8 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
             }
             case "provider-turn.updated": {
               const existingRows =
-                event.payload.tokenUsage === undefined
+                event.payload.tokenUsage === undefined ||
+                event.payload.inputRecordedAt === undefined
                   ? yield* sql<PayloadRow>`
                     SELECT payload_json
                     FROM orchestration_v2_projection_provider_turns

apps/server/src/orchestration-v2/ProviderTurnStartService.ts

@@ -65,6 +65,7 @@ import * as RuntimePolicy from "./RuntimePolicy.ts";
 import * as TurnItemPositionStore from "./TurnItemPositionStore.ts";
 import {
   isRestartNoteContinuation,
+  isRestartResendContinuation,
   pendingRestartCancelledBackgroundWork,
   restartCancelledBackgroundWorkNote,
 } from "./RestartBackgroundNote.ts";
@@ -883,6 +884,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,
@@ -1758,7 +1763,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: goalCommand ? userText : withInlineContext(context, userText),
@@ -1821,7 +1826,7 @@ export const layer: Layer.Layer<
         retryHandoff.length === 0 &&
         sideChatHandoff === undefined &&
         restartNote === "" &&
-        !noteContinuation
+        !promptedContinuation
           ? session
           : makeDeliverySession(session, startWithHandoffs);
       return runExecution.startRootRun({

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);

apps/server/src/orchestration-v2/RestartBackgroundNote.ts

@@ -304,6 +304,50 @@ 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.
+ */
+export function restartSourceInputUnrecorded(
+  source: OrchestrationV2Run,
+  providerTurns: ReadonlyArray<
+    Pick<OrchestrationV2ProviderTurn, "runAttemptId" | "inputRecordedAt">
+  >,
+  runs: ReadonlyArray<OrchestrationV2Run>,
+): boolean {
+  const visited = new Set<string>();
+  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(
+  run: Pick<OrchestrationV2Run, "restartContinuationOfRunId">,
+  runs: ReadonlyArray<OrchestrationV2Run>,
+  providerTurns: Parameters<typeof restartSourceInputUnrecorded>[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<OrchestrationV2Run>;