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

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: '<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",
+                        }),
+                      },
+                    ],
+                  }
+                : 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();

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<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);
@@ -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") {

apps/server/src/orchestration-v2/EffectWorker.test.ts

@@ -813,6 +813,7 @@ it.effect.each([3, 4])(
                 },
               ],
               messages: [],
+              providerThreads: [],
               providerTurns: [],
               runtimeRequests: [],
               turnItems: [],

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<OrchestrationV2ProviderTurn> {
   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<ProjectionStoreV2, never, SqlClient.SqlClient> =
             }
             case "provider-turn.updated": {
               const existingRows =
-                event.payload.tokenUsage === undefined
+                event.payload.tokenUsage === undefined || event.payload.inputRecordedAt == null
                   ? yield* sql<PayloadRow>`
                     SELECT payload_json
                     FROM orchestration_v2_projection_provider_turns

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({

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

@@ -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<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 !== 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<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>;