fix(recovery): deliver a steered message the restart lost before the model read it (#19962)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work
> - A user can steer a running Codex or Pi turn: a comment goes into the
live turn instead of waiting for the next run
> - The provider acknowledges the steer at once, but the model reads it
only at its next step, after the in-flight tool call
> - A control-plane restart in that window kills the provider with the
message in memory. Restart recovery starts a continuation turn that does
not carry the message. The message is lost and no error shows
> - The restart chaos harness showed this in 6 of 6 steered runs (Codex,
K1, K2 and K6, 2 samples each): steer acknowledged, run succeeded, no
reply to the steered message
> - This pull request makes restart recovery deliver every acknowledged
steer that the model did not read, exactly once
> - The benefit is that manual Steer no longer loses messages on
restart, and the planned automatic steering can ship on top of it

## Linked Issues or Issue Description

**What happened?**
A comment steered into a running Codex turn got no reply when a
control-plane restart (SIGKILL, SIGTERM or crash loop) happened during
the agent's tool call. The run succeeded, and the steered comment had no
effect.

**Steps to reproduce**
1. Start a Codex task that runs a 40-second command.
2. During the command, post a comment and click Steer. The provider
acknowledges it.
3. Restart the control plane before the command ends.
4. Recovery resumes the turn with a continuation. The agent finishes the
task and ignores the comment.

**Expected behavior**
The agent reads and handles the steered comment once, after recovery.

**Root cause**
The provider keeps acknowledged steering input in memory until its next
model step. The restart continuation
(`paperclip.native-continuation.v1`) only told the model to continue the
current request, so the acknowledged input was gone.

## What Changed

- `unreadSteeredMessageIds` (runner package): derives
acknowledged-but-unread steers from run-log events in order. A
`steering_acknowledgement` item adds the message id (once per id). A
`userMessage` item in the same turn reads the oldest unread steer. Codex
records each steered input as one `userMessage` item when it enters the
conversation. A `userMessage` item in a later turn reads all earlier
steers, because the continuation carries them. `turn.completed` reads
all steers.
- `readUnreadSteeredMessages` (server): reads those facts from
`heartbeat_run_events` (only transport-validated events with a
`sourceEventId`). It also reads the comment or interaction text for the
same company and issue. The run log is the durable record of both
"acknowledged" and "read", so no new column is needed.
- The restart continuation now includes `steeredMessages` and one
instruction to handle them, oldest first. The runtime reads them through
a lazy `getUnreadSteeredMessages` option, only when a restart
continuation starts. Native recovery wires it for every run.
- `queuedInteractionSteeringBody` is extracted, so recovery rebuilds a
steered interaction response with the same text the steer sent.
- Fake Codex gets `--consume-steer`, which emits the `userMessage` item
for an accepted steer, like real Codex.

## Verification

- Red/green live test with real runnerd and fake Codex
(`runnerd-restart-continuation.test.ts`, "delivers a steered message the
restart lost exactly once"): start a held turn, steer it, SIGKILL runner
and provider, restore and recover.
- "unread": the continuation carries the message exactly once. Without
the runtime change this test fails.
- "read" (`--consume-steer`): the continuation does not carry it again.
- Both: after the continuation completes, a later restart has nothing to
deliver.
- `steering-delivery.test.ts`: ordering, dedupe, initial-input,
later-turn and completed-turn rules.
- `unread-steered-messages.test.ts` (embedded Postgres): ignores events
without a source id and messages from other issues. It reads interaction
text, ignores model-output items as read evidence, and returns nothing
after the continuation input.
- `pnpm vitest run src/steering-delivery.test.ts
src/live/runnerd-restart-continuation.test.ts` (runner) and `pnpm vitest
run src/services/native-runtime/unread-steered-messages.test.ts`
(server): pass on the rebased revision.
- `issue-queued-comments-routes.test.ts` and
`native-session-runtime.test.ts -t restart`: pass.
`native-session-resume.test.ts` "archives a damaged prior epoch…" fails
locally on plain `origin/master` too (local runnerd binary); CI runs it
with a fresh build.
- Restart chaos (Codex steer mode, K1/K2/K6) runs after this lands in
the harness slot. Before: 0 of 6 steered messages answered.

## Risks

- Duplicate delivery: a provider that records steered input without a
`userMessage` item, or a `userMessage` event lost in the crash window
before persistence, makes recovery send the message again. The model
then sees it twice. That is safer than losing it, and the instruction
asks the model to handle each one once.
- Pi (ACPX) steering has no `userMessage` read evidence today, so a Pi
steer that the model already read before a restart may arrive again in
the continuation.
- Only the restart continuation carries the messages. Other recovery
paths start after a completed turn, which already read every steer.
- No schema or migration change. A session saved by an earlier release
resumes normally; its run log already holds the acknowledgement events.

## Model Used

- Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, via Claude Code with
tool use (shell, file editing, test execution).

## Checklist

- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [ ] All Paperclip CI gates are green
- [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-codex-app-server.rs

@@ -717,6 +717,9 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
         .iter()
         .any(|value| value == "--hold-first-durable-turn");
     let durable_tool_ids = args.iter().any(|value| value == "--durable-tool-ids");
+    // Real Codex records each turn's input, and accepted steering input when it
+    // enters the conversation at the next model step, as userMessage items.
+    let consume_steer = args.iter().any(|value| value == "--consume-steer");
     let expected_canonical_task_context_file =
         argument(&args, "--expected-canonical-task-context-file");
     let emit_tool_call = args.iter().any(|value| value == "--emit-tool-call");
@@ -1833,6 +1836,19 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
                     "method": "turn/started",
                     "params": {"turn": {"id": provider_turn_id}}
                 }))?;
+                if consume_steer {
+                    // Real Codex records the turn's own input as its first userMessage item.
+                    for method in ["item/started", "item/completed"] {
+                        send(json!({
+                            "method": method,
+                            "params": {
+                                "threadId": state.thread_id,
+                                "turnId": provider_turn_id,
+                                "item": {"id": format!("{provider_turn_id}-input"), "type": "userMessage", "content": []}
+                            }
+                        }))?;
+                    }
+                }
                 if args.iter().any(|value| value == "--account-notifications") {
                     send(json!({"method":"account/updated", "params":{
                         "authMode":"chatgpt", "planType":"pro"
@@ -2262,6 +2278,18 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
             }
             "turn/steer" => {
                 send(json!({"id": id, "result": {"accepted": true}}))?;
+                if consume_steer {
+                    for method in ["item/started", "item/completed"] {
+                        send(json!({
+                            "method": method,
+                            "params": {
+                                "threadId": state.thread_id,
+                                "turnId": state.active_turn_id,
+                                "item": {"id": "steered-user-message-1", "type": "userMessage", "content": []}
+                            }
+                        }))?;
+                    }
+                }
                 if linger_after_turn_start {
                     send(json!({
                         "method": "item/completed",

packages/paperclip-runner/src/index.ts

@@ -34,6 +34,7 @@ export {
   summarizeCodexTurnDiff,
 } from "./drivers/codex/codex-turn-diff.js";
 export * from "./native-session-runtime.js";
+export * from "./steering-delivery.js";
 export {
   DurablePrpControlPlane,
   RUNNER_PRP_PROTOCOL_MIN,

packages/paperclip-runner/src/live/runnerd-restart-continuation.test.ts

@@ -7,6 +7,7 @@ import type { NativeExecutionInputV1 } from "../contracts/native-execution.js";
 import type { PersistedNativeSession } from "../contracts/native-session-backend.js";
 import type { PrpEvent } from "../protocol/replay-contract.js";
 import { executeNativeSession } from "../native-session-runtime.js";
+import { unreadSteeredMessageIds } from "../steering-delivery.js";
 import { createRunnerdCodexTransport, defaultCapabilityRunnerdBinary } from "./runnerd-codex-transport.js";
 
 it.each(["none", "before-start", "after-start"] as const)("completes after the real runner and provider are killed (second crash: %s)", async (crash) => {
@@ -309,3 +310,115 @@ it("finalizes the durable proposed result when the runner and provider die befor
     await rm(root, { recursive: true, force: true });
   }
 }, 40_000);
+
+it.each(["unread", "read"] as const)("delivers a steered message the restart lost exactly once (%s before the kill)", async (state) => {
+  // A user steers the live turn during a tool call, then the controller,
+  // runner and provider die before the model's next step ("unread"), or after
+  // the provider recorded the message ("read"). Recovery reads the durable run
+  // log: the unread message must reach the continuation once; the read one
+  // must not reach it again.
+  const root = await mkdtemp(join(tmpdir(), "runner-restart-steer-"));
+  const identity = { runId: "run-steer", sessionId: "session-steer", companyId: "company-steer", issueId: "issue-steer", agentId: "agent-steer" };
+  const execution: NativeExecutionInputV1 = {
+    schema: "paperclip.native-execution-input.v1",
+    binding: { runId: identity.runId, companyId: identity.companyId, issueId: identity.issueId, agentId: identity.agentId, executionWorkspaceId: "workspace-steer" },
+    session: { normalizedSessionId: identity.sessionId, driverKind: "codex_app_server", protocolVersion: 1, lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null } },
+    task: { identifier: "TEST-4", title: "Finish checks", description: null, prompt: "Watch the already running checks and finish the request.", workMode: "standard" },
+    workspace: { cwd: root, repoUrl: null, repoRef: null, branchName: null },
+    provider: { kind: "codex", model: null },
+    completionContract: { id: "contract-steer", sha256: "digest", schemaVersion: "paperclip.completion-contract.v1", contract: {
+      revision: "1", objective: "Finish checks", criteria: [{ id: "checks", requirement: "Check the outcome" }],
+    } },
+    interactionResponses: [], credentialBindings: [],
+  };
+  const commentId = "0b6f8c62-5d1a-4c55-9a7e-2f1c3b4d5e6f";
+  const steeredText = "HELLO steer-nonce-7f3a";
+  const options = {
+    runnerBinary: defaultCapabilityRunnerdBinary(),
+    codexCommand: resolve(import.meta.dirname, "../../runner/target/debug/fake-codex-app-server"),
+    codexArgs: ["--state-file", join(root, "fake-state.json"), "--call-log", join(root, "calls.log"), "--durable-turn-ids", "--hold-first-durable-turn",
+      ...(state === "read" ? ["--consume-steer"] : [])],
+    stateDirectory: root,
+    prpIdentity: { runnerInstanceId: "runner-steer", environmentLeaseId: "lease-steer", runId: identity.runId, normalizedSessionId: identity.sessionId, turnId: "turn-steer", itemId: "item-steer" },
+  };
+  const first = createRunnerdCodexTransport(options);
+  let restored: ReturnType<typeof createRunnerdCodexTransport> | undefined;
+  const alive = (pid: number | null) => {
+    if (!pid) return false;
+    try { process.kill(pid, 0); return true; } catch { return false; }
+  };
+  const kill = (pid: number | null) => {
+    if (!pid) return;
+    try { process.kill(pid, "SIGKILL"); } catch (error) { if ((error as NodeJS.ErrnoException).code !== "ESRCH") throw error; }
+  };
+  // The server's run-log reader, over the events the control plane persisted.
+  const unread = (events: PrpEvent[]) => unreadSteeredMessageIds(events.map(event => ({
+    eventType: event.eventType, turnId: event.turnId, itemId: event.itemId,
+    kind: typeof (event.payload as { kind?: unknown }).kind === "string" ? (event.payload as { kind: string }).kind : null,
+  })));
+  try {
+    const backend = createRunnerdNativeSessionBackend(execution, { runnerInstanceId: "runner-steer", transportFactory: () => first.transport });
+    const session = await backend.openSession({ identity, workingDirectory: root });
+    await session.startTurn({ message: { role: "user", text: execution.task.prompt } });
+    const started = await session.snapshot();
+    expect(started.activeTurnId).toBe("provider-turn-1");
+    expect((await session.capabilities()).steering).toBe(true);
+    await session.steer!({ turnId: started.activeTurnId!, message: { role: "user", text: steeredText }, correlationId: commentId });
+    const persisted = await session.snapshot();
+    const durableEvents: PrpEvent[] = [];
+    const initialEvents = session.events()[Symbol.asyncIterator]();
+    const kinds = () => durableEvents.map(event => (event.payload as { kind?: unknown }).kind);
+    // "read": the turn's input and the steered message are both userMessage items.
+    while (!kinds().includes("steering_acknowledgement") || (state === "read" && durableEvents.filter(event => event.eventType === "item.started" && (event.payload as { kind?: unknown }).kind === "userMessage").length < 2)) {
+      const event = await initialEvents.next();
+      if (event.done) throw new Error("initial event history ended early");
+      durableEvents.push(event.value);
+    }
+    expect(unread(durableEvents)).toEqual(state === "unread" ? [commentId] : []);
+    const owner = first.evidence();
+    await first.detachControllerForRestart();
+    kill(owner.runnerPid);
+    kill(owner.providerPid);
+    await vi.waitFor(() => { expect(alive(owner.runnerPid)).toBe(false); expect(alive(owner.providerPid)).toBe(false); });
+    const providerState = JSON.parse(await readFile(join(root, "fake-state.json"), "utf8"));
+    providerState.activeTurnId = null;
+    await writeFile(join(root, "fake-state.json"), JSON.stringify(providerState));
+    restored = createRunnerdCodexTransport({ ...options, resumeDynamicTools: [], resumeActiveTurnId: persisted.activeTurnId });
+    const recovering = createRunnerdNativeSessionBackend(execution, { runnerInstanceId: "runner-steer", transportFactory: () => restored!.transport });
+    const completion = await executeNativeSession({
+      input: execution, backend: recovering, persistedSession: persisted, resumeInterruptedTurn: true,
+      getUnreadSteeredMessages: async () => unread(durableEvents).map(id => ({ id, text: id === commentId ? steeredText : "" })),
+      runnerInstanceId: "runner-steer", controlPlaneInstanceId: "controller-steer", timeoutMs: 15_000,
+      controlPlane: {
+        async openRun() {},
+        async checkpointSession() {},
+        async appendEvent(event) { if (!("sourceSeq" in event)) throw new Error("Expected a durable PRP event"); durableEvents.push(event); return { cursor: durableEvents.length, highestContiguousSourceSeq: event.sourceSeq, disposition: "committed" }; },
+        async replayEvents(request) {
+          const source = durableEvents.filter(event => event.sourceInstanceId === request.sourceInstanceId);
+          return { events: source.filter(event => event.sourceSeq > request.afterSourceSeq), highestContiguousSourceSeq: source.at(-1)?.sourceSeq ?? 0 };
+        },
+        async completeRun() {},
+      },
+    });
+    expect(completion).toMatchObject({ turnId: "provider-turn-2", terminal: { runTerminalState: "succeeded" } });
+    const journal = JSON.parse(await readFile(join(root, "control-plane/control-plane-state.json"), "utf8"));
+    const submissions: Array<{ payload: { text: string } }> = journal.commands.filter((command: { type: string }) => command.type === "turn.start");
+    expect(submissions).toHaveLength(2);
+    const continuation = submissions[1]!.payload.text;
+    expect(continuation).toContain("Reconcile any unfinished tool or command");
+    expect(continuation.split(steeredText).length - 1).toBe(state === "unread" ? 1 : 0);
+    if (state === "unread") expect(JSON.parse(continuation).steeredMessages).toEqual([steeredText]);
+    // The continuation's completed turn is the durable read fact: a later
+    // restart has nothing left to deliver.
+    expect(unread(durableEvents)).toEqual([]);
+  } finally {
+    await restored?.transport.close().catch(() => undefined);
+    for (const bundle of [first, restored]) {
+      if (!bundle) continue;
+      const evidence = bundle.evidence();
+      kill(evidence.runnerPid);
+      kill(evidence.providerPid);
+    }
+    await rm(root, { recursive: true, force: true });
+  }
+}, 30_000);

packages/paperclip-runner/src/native-session-runtime.ts

@@ -304,6 +304,12 @@ export interface ExecuteNativeSessionOptions {
   resumeInterruptedTurn?: boolean;
   /** Read bounded task history only after recovery requires a fresh conversation. */
   getFreshSessionHandoff?: () => Promise<string | null>;
+  /**
+   * Steered messages the provider acknowledged but the model never read, read
+   * when a restart continuation starts. The provider lost them with the
+   * interrupted turn, so the continuation delivers them.
+   */
+  getUnreadSteeredMessages?: () => Promise<Array<{ id: string; text: string }>>;
   /** Durable launch intent, after cleanup admission and before provider calls. */
   onSessionAdmission?: () => Promise<void>;
   input: NativeExecutionInput;
@@ -2649,18 +2655,22 @@ export async function executeNativeSession(
       return snapshot;
     };
     const recoveredSnapshot = await session.snapshot();
-    const restartContinuation = () => ({
+    const restartContinuation = (steeredMessages: Array<{ id: string; text: string }>) => ({
       message: { role: "user" as const, text: JSON.stringify({
         schema: "paperclip.native-continuation.v1",
         events: [
           "This is a resumed session after a Paperclip control-plane restart. Paperclip restored this same conversation and workspace; the restart interrupted the previous turn.",
           "Continue the current user request from the recorded progress and preserved workspace.",
+          ...(steeredMessages.length > 0 ? [
+            "During the interrupted turn the user sent the messages in steeredMessages. The restart happened before you read them. Handle each one in this turn, oldest first, together with the current request.",
+          ] : []),
           "Shell commands and background processes started in earlier turns may no longer be running, and their tool sessions may be gone. Do not block only because such a process or its tool session cannot be found or inspected.",
           "Reconcile any unfinished tool or command before proceeding: check the workspace for the outputs or artifacts it should have produced. If the expected output is present, use it. If the expected output is missing, re-run the command.",
           "Do not repeat completed work. Before you repeat an action with external effects, such as a push, a message or a deployment, check whether it already took effect. If you cannot determine that, report the specific blocker.",
           "Finish in this turn. Nothing resumes this conversation after the turn ends, so do not end it to wait for a background command or a notification; wait for or check that work first.",
           "Then invoke exactly one paperclip_finish or paperclip_block with the accurate disposition.",
         ].join("\n"),
+        ...(steeredMessages.length > 0 ? { steeredMessages: steeredMessages.map(message => message.text) } : {}),
         completion: {
           revision: input.completionContract.contract.revision,
           criterionIds: input.completionContract.contract.criteria.map(criterion => criterion.id),
@@ -2705,7 +2715,9 @@ export async function executeNativeSession(
       continuationSetup = (async () => {
         // The driver reports the borrowed marker with the accepted turn; a
         // later snapshot may already show the marker returned.
-        const started = await session.startTurn(restartContinuation());
+        const steeredMessages = await options.getUnreadSteeredMessages?.() ?? [];
+        signal?.throwIfAborted();
+        const started = await session.startTurn(restartContinuation(steeredMessages));
         if (started.borrowedDispositionMarker === true) borrowedContinuationTurnId = started.turnId;
         await checkpoint(signal);
       })();

packages/paperclip-runner/src/steering-delivery.test.ts

@@ -0,0 +1,49 @@
+import { describe, expect, it } from "vitest";
+import { unreadSteeredMessageIds, type SteeringDeliveryEvent } from "./steering-delivery.js";
+
+const ack = (turnId: string, id: string): SteeringDeliveryEvent =>
+  ({ eventType: "item.completed", kind: "steering_acknowledgement", turnId, itemId: `${turnId}:steer:${id}` });
+const userMessage = (turnId: string): SteeringDeliveryEvent => ({ eventType: "item.started", kind: "userMessage", turnId });
+
+describe("unreadSteeredMessageIds", () => {
+  it("keeps a steer the model never read before the restart", () => {
+    expect(unreadSteeredMessageIds([userMessage("t1"), ack("t1", "c1"), { eventType: "turn.failed", turnId: "t1" }]))
+      .toEqual(["c1"]);
+  });
+
+  it("treats the turn's own input as read, not a later steer", () => {
+    // The initial input's userMessage precedes the acknowledgement.
+    expect(unreadSteeredMessageIds([userMessage("t1"), { eventType: "item.started", kind: "agentMessage", turnId: "t1" }, ack("t1", "c1")]))
+      .toEqual(["c1"]);
+  });
+
+  it("drops a steer once its userMessage item starts, oldest first", () => {
+    expect(unreadSteeredMessageIds([userMessage("t1"), ack("t1", "c1"), ack("t1", "c2"), userMessage("t1")])).toEqual(["c2"]);
+  });
+
+  it("counts a steered userMessage reported before its acknowledgement as read", () => {
+    expect(unreadSteeredMessageIds([userMessage("t1"), userMessage("t1"), ack("t1", "c1"), ack("t1", "c2")])).toEqual(["c2"]);
+    // An early read belongs to its own turn only.
+    expect(unreadSteeredMessageIds([userMessage("t1"), userMessage("t1"), { eventType: "turn.failed", turnId: "t1" },
+      userMessage("t2"), ack("t2", "c3")])).toEqual(["c3"]);
+  });
+
+  it("counts every earlier steer as read when a later turn's input starts", () => {
+    expect(unreadSteeredMessageIds([ack("t1", "c1"), ack("t1", "c2"), userMessage("t2"), ack("t2", "c3")]))
+      .toEqual(["c3"]);
+  });
+
+  it("counts every steer as read once a turn completes", () => {
+    expect(unreadSteeredMessageIds([ack("t1", "c1"), { eventType: "turn.completed", turnId: "t2" }])).toEqual([]);
+  });
+
+  it("delivers a message once even if its acknowledgement repeats", () => {
+    expect(unreadSteeredMessageIds([ack("t1", "c1"), ack("t1", "c1")])).toEqual(["c1"]);
+    expect(unreadSteeredMessageIds([userMessage("t1"), ack("t1", "c1"), userMessage("t1"), ack("t1", "c1")])).toEqual([]);
+  });
+
+  it("ignores acknowledgements without a message id", () => {
+    expect(unreadSteeredMessageIds([{ eventType: "item.completed", kind: "steering_acknowledgement", turnId: "t1", itemId: "t1:other" }]))
+      .toEqual([]);
+  });
+});

packages/paperclip-runner/src/steering-delivery.ts

@@ -0,0 +1,58 @@
+/** One durable run-log event, in persisted sequence order. */
+export interface SteeringDeliveryEvent {
+  eventType: string;
+  turnId?: string | null;
+  itemId?: string | null;
+  kind?: string | null;
+}
+
+const STEER_ITEM = /:steer:(.+)$/;
+
+/**
+ * Steering messages a provider acknowledged but the model never read, oldest
+ * first. A restart kills an in-flight turn before its next model step, and the
+ * provider forgets acknowledged input it never recorded, so recovery must
+ * deliver these again. The acknowledgement item id ends in `:steer:<id>`,
+ * where `<id>` is the steered comment or interaction id; each id counts once.
+ *
+ * Read evidence, in run-log order:
+ * - the first userMessage item of a turn is that turn's own input. It reads
+ *   every steer from earlier turns: the restart continuation carries them;
+ * - every later userMessage item in the turn is one steered input entering
+ *   the conversation. It reads the oldest unread steer of that turn. The
+ *   provider can report it before the acknowledgement event, so a read with
+ *   no unread steer waits for the next acknowledgement of that turn;
+ * - turn.completed reads every steer so far: a provider drains pending input
+ *   before it completes a turn.
+ */
+export function unreadSteeredMessageIds(events: Iterable<SteeringDeliveryEvent>): string[] {
+  const unread: Array<{ id: string; turnId: string | null }> = [];
+  const seen = new Set<string>();
+  const inputs = new Map<string | null, number>();
+  const earlyReads = new Map<string | null, number>();
+  for (const event of events) {
+    const turnId = event.turnId ?? null;
+    if (event.eventType === "item.completed" && event.kind === "steering_acknowledgement") {
+      const id = STEER_ITEM.exec(event.itemId ?? "")?.[1];
+      if (!id || seen.has(id)) continue;
+      seen.add(id);
+      const early = earlyReads.get(turnId) ?? 0;
+      if (early > 0) earlyReads.set(turnId, early - 1);
+      else unread.push({ id, turnId });
+    } else if (event.eventType === "item.started" && event.kind === "userMessage") {
+      const count = (inputs.get(turnId) ?? 0) + 1;
+      inputs.set(turnId, count);
+      if (count === 1) {
+        for (let index = unread.length - 1; index >= 0; index -= 1)
+          if (unread[index]!.turnId !== turnId) unread.splice(index, 1);
+        continue;
+      }
+      const index = unread.findIndex((entry) => entry.turnId === turnId);
+      if (index >= 0) unread.splice(index, 1);
+      else earlyReads.set(turnId, (earlyReads.get(turnId) ?? 0) + 1);
+    } else if (event.eventType === "turn.completed") {
+      unread.length = 0;
+    }
+  }
+  return unread.map((entry) => entry.id);
+}

server/src/services/native-runtime/native-session-executor.ts

@@ -56,6 +56,7 @@ import { preparePluginRunnerCredentials } from "./plugin-runner-credentials.js";
 import { createPluginRunnerTaskOwner, verifyPluginRunnerTaskReattachment } from "./plugin-runner-task.js";
 import { NativeExecutionContext } from "./native-execution-context.js";
 import { NativeStartupCleanupPendingError, clearNativeStartupContext } from "./native-startup-cleanup.js";
+import { readUnreadSteeredMessages } from "./unread-steered-messages.js";
 import { coldAdmissionTimeoutMs } from "../../vendor/paperclip-runner/index.js";
 import { applyObservabilityEnvironment, type ProviderObservabilityLaunch } from "@paperclipai/adapter-utils/agent-observability";
 import { museRunnerBroker } from "../muse-runner-broker.js";
@@ -10921,6 +10922,11 @@ async function executePaperclipNativeSessionWithinScope(
           executeNativeSession({
             resumeInterruptedTurn: input.restartRecovery?.kind === "resume_dead_runner",
             getFreshSessionHandoff: input.getFreshSessionHandoff,
+            getUnreadSteeredMessages: () => readUnreadSteeredMessages(input.db, {
+              companyId: input.execution.binding.companyId,
+              runId: input.execution.binding.runId,
+              issueId: input.execution.binding.issueId,
+            }),
             onSessionAdmission: async () => {
               if (existingWarmSession) runnerdBackend?.bindDiagnosticSession?.(existingWarmSession);
               // Invalidate prior stop evidence before a backend can spawn.

server/src/services/native-runtime/unread-steered-messages.test.ts

@@ -0,0 +1,98 @@
+import { randomUUID } from "node:crypto";
+
+import { afterAll, beforeAll, describe, expect, it } from "vitest";
+import {
+  agents,
+  companies,
+  createDb,
+  heartbeatRunEvents,
+  heartbeatRuns,
+  issueComments,
+  issueThreadInteractions,
+  issues,
+} from "@paperclipai/db";
+
+import { readUnreadSteeredMessages } from "./unread-steered-messages.js";
+import { queuedInteractionSteeringBody } from "../queued-interaction-response.js";
+import {
+  getEmbeddedPostgresTestSupport,
+  startEmbeddedPostgresTestDatabase,
+} from "../../__tests__/helpers/embedded-postgres.js";
+
+const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
+const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
+
+describeEmbeddedPostgres("unread steered messages", () => {
+  let db!: ReturnType<typeof createDb>;
+  let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
+
+  beforeAll(async () => {
+    tempDb = await startEmbeddedPostgresTestDatabase("paperclip-unread-steered-");
+    db = createDb(tempDb.connectionString);
+  }, 120_000);
+
+  afterAll(async () => {
+    await db?.$client?.end?.({ timeout: 0 });
+    await tempDb?.cleanup();
+  });
+
+  async function seedIssue(companyId: string, agentId: string, issueNumber: number, prefix: string) {
+    const id = randomUUID();
+    await db.insert(issues).values({
+      id, companyId, title: "Steered task", status: "in_progress", priority: "medium",
+      assigneeAgentId: agentId, issueNumber, identifier: `${prefix}-${issueNumber}`,
+    });
+    return id;
+  }
+
+  it("returns acknowledged steers the model never read, from the durable run log only", async () => {
+    const companyId = randomUUID();
+    const agentId = randomUUID();
+    const runId = randomUUID();
+    const prefix = `S${companyId.replaceAll("-", "").slice(0, 6)}`;
+    await db.insert(companies).values({ id: companyId, name: "Paperclip", issuePrefix: prefix, requireBoardApprovalForNewAgents: false });
+    await db.insert(agents).values({
+      id: agentId, companyId, name: "Codex", role: "engineer", status: "running",
+      adapterType: "paperclip_runner", adapterConfig: {}, runtimeConfig: {}, permissions: {},
+    });
+    const issueId = await seedIssue(companyId, agentId, 1, prefix);
+    const otherIssueId = await seedIssue(companyId, agentId, 2, prefix);
+    await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "running", contextSnapshot: { issueId } });
+    const [read, unread, notValidated, otherIssue] = await db.insert(issueComments).values([
+      { companyId, issueId, authorUserId: "board", body: "Already read" },
+      { companyId, issueId, authorUserId: "board", body: "HELLO unread" },
+      { companyId, issueId, authorUserId: "board", body: "Never acknowledged by the transport" },
+      { companyId, issueId: otherIssueId, authorUserId: "board", body: "Another task" },
+    ]).returning();
+    const [interaction] = await db.insert(issueThreadInteractions).values({
+      companyId, issueId, kind: "ask_user_questions", status: "answered", effectiveResolverPolicy: "human_only",
+      sourceRunId: runId, payload: { version: 1, questions: [] }, result: { version: 1, answers: [] },
+    }).returning();
+    let seq = 0;
+    const event = (eventType: string, turnId: string, kind: string | null, itemId?: string, validated = true) => ({
+      companyId, runId, agentId, seq: ++seq, eventType, sourceEventId: validated ? randomUUID() : null,
+      payload: { prpEvent: { turnId, ...(itemId ? { itemId } : {}), payload: kind ? { kind } : {} } },
+    });
+    await db.insert(heartbeatRunEvents).values([
+      event("item.started", "t1", "userMessage", "t1:input"),
+      event("item.completed", "t1", "steering_acknowledgement", `t1:steer:${read!.id}`),
+      event("item.started", "t1", "userMessage", "t1:steered"),
+      event("item.completed", "t1", "steering_acknowledgement", `t1:steer:${unread!.id}`),
+      event("item.completed", "t1", "steering_acknowledgement", `t1:steer:${notValidated!.id}`, false),
+      event("item.completed", "t1", "steering_acknowledgement", `t1:steer:${otherIssue!.id}`),
+      event("item.completed", "t1", "steering_acknowledgement", `t1:steer:${interaction!.id}`),
+      // A model output item alone is not read evidence.
+      event("item.started", "t1", "agentMessage", "t1:message"),
+      event("turn.failed", "t1", null),
+    ]);
+
+    expect(await readUnreadSteeredMessages(db, { companyId, runId, issueId })).toEqual([
+      { id: unread!.id, text: "HELLO unread" },
+      { id: interaction!.id, text: queuedInteractionSteeringBody(interaction!) },
+    ]);
+
+    // The continuation's input reaches the model: nothing is delivered twice.
+    await db.insert(heartbeatRunEvents).values(event("item.started", "t2", "userMessage", "t2:input"));
+    expect(await readUnreadSteeredMessages(db, { companyId, runId, issueId })).toEqual([]);
+  });
+});

server/src/services/native-runtime/unread-steered-messages.ts

@@ -0,0 +1,68 @@
+import { and, asc, eq, inArray, isNotNull, or, sql } from "drizzle-orm";
+import { heartbeatRunEvents, issueComments, issueThreadInteractions, type Db } from "@paperclipai/db";
+import { unreadSteeredMessageIds } from "../../vendor/paperclip-runner/index.js";
+import { queuedInteractionSteeringBody } from "../queued-interaction-response.js";
+
+const kind = sql<string | null>`${heartbeatRunEvents.payload}->'prpEvent'->'payload'->>'kind'`;
+
+/**
+ * Steered messages a restart lost before the model read them, oldest first.
+ * The run log is the durable record of both facts: the provider's steering
+ * acknowledgement and the userMessage item or completed turn that shows the
+ * model read it. Only events the native transport validated count.
+ */
+export async function readUnreadSteeredMessages(
+  db: Db,
+  input: { companyId: string; runId: string; issueId: string },
+): Promise<Array<{ id: string; text: string }>> {
+  const rows = await db
+    .select({
+      eventType: heartbeatRunEvents.eventType,
+      turnId: sql<string | null>`${heartbeatRunEvents.payload}->'prpEvent'->>'turnId'`,
+      itemId: sql<string | null>`${heartbeatRunEvents.payload}->'prpEvent'->>'itemId'`,
+      kind,
+    })
+    .from(heartbeatRunEvents)
+    .where(and(
+      eq(heartbeatRunEvents.companyId, input.companyId),
+      eq(heartbeatRunEvents.runId, input.runId),
+      isNotNull(heartbeatRunEvents.sourceEventId),
+      or(
+        eq(heartbeatRunEvents.eventType, "turn.completed"),
+        and(eq(heartbeatRunEvents.eventType, "item.completed"), sql`${kind} = 'steering_acknowledgement'`),
+        and(eq(heartbeatRunEvents.eventType, "item.started"), sql`${kind} = 'userMessage'`),
+      ),
+    ))
+    .orderBy(asc(heartbeatRunEvents.seq));
+  const ids = unreadSteeredMessageIds(rows);
+  if (ids.length === 0) return [];
+  const uuids = ids.filter((id) => /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i.test(id));
+  if (uuids.length === 0) return [];
+  const texts = new Map<string, string>();
+  const comments = await db
+    .select({ id: issueComments.id, body: issueComments.body })
+    .from(issueComments)
+    .where(and(
+      eq(issueComments.companyId, input.companyId),
+      eq(issueComments.issueId, input.issueId),
+      inArray(issueComments.id, uuids),
+    ));
+  for (const comment of comments) texts.set(comment.id, comment.body);
+  const remaining = uuids.filter((id) => !texts.has(id));
+  if (remaining.length > 0) {
+    // A steered interaction response uses the interaction id.
+    const interactions = await db
+      .select()
+      .from(issueThreadInteractions)
+      .where(and(
+        eq(issueThreadInteractions.companyId, input.companyId),
+        eq(issueThreadInteractions.issueId, input.issueId),
+        inArray(issueThreadInteractions.id, remaining),
+      ));
+    for (const interaction of interactions) texts.set(interaction.id, queuedInteractionSteeringBody(interaction));
+  }
+  return ids.flatMap((id) => {
+    const text = texts.get(id);
+    return text === undefined ? [] : [{ id, text }];
+  });
+}

server/src/services/queued-interaction-response.ts

@@ -55,6 +55,20 @@ function deliveredInteractionRunExists(companyId: string, issueId: string) {
       AND delivered_run.status IN ('running', 'succeeded'))`;
 }
 
+/**
+ * The explicit steering message for a resolved interaction. It preserves the
+ * question labels, selected options, custom answers, approval target and exact
+ * revision.
+ */
+export function queuedInteractionSteeringBody(interaction: {
+  id: string; kind: string; status: string; title?: string | null; payload?: unknown; result?: unknown;
+}): string {
+  return `${queuedInteractionSummary(interaction)}\n\n${JSON.stringify({
+    interactionId: interaction.id, kind: interaction.kind, status: interaction.status,
+    payload: interaction.payload, result: record(interaction.result),
+  }, null, 2)}`;
+}
+
 export type QueuedInteractionResponse = NonNullable<Awaited<ReturnType<typeof readQueuedInteractionResponse>>>;
 
 export async function readQueuedInteractionResponse(db: Db, companyId: string, issueId: string, payload: unknown) {
@@ -70,15 +84,8 @@ export async function readQueuedInteractionResponse(db: Db, companyId: string, i
   ));
   const interaction = row?.interaction;
   if (!interaction || interaction.status !== record(payload).interactionStatus || row.delivered) return null;
-  const p = record(interaction.payload);
-  const result = record(interaction.result);
   const heading = queuedInteractionSummary(interaction);
-  // Preserve the question labels, selected options, custom answers, approval
-  // target and exact revision. This is also the explicit steering message.
-  const body = `${heading}\n\n${JSON.stringify({
-    interactionId: interaction.id, kind: interaction.kind, status: interaction.status,
-    payload: interaction.payload, result,
-  }, null, 2)}`;
+  const body = queuedInteractionSteeringBody(interaction);
   const comment: IssueComment = {
     id: interaction.id, companyId, issueId, authorType: interaction.resolvedByUserId ? "user" : "agent",
     authorUserId: interaction.resolvedByUserId, authorAgentId: interaction.resolvedByAgentId,

server/src/vendor/paperclip-runner/index.ts

@@ -147,6 +147,7 @@ export const defaultCapabilityRunnerdBinary =
   runner.defaultCapabilityRunnerdBinary;
 export const runnerBinaryTarget = runner.runnerBinaryTarget;
 export const coldAdmissionTimeoutMs = runner.coldAdmissionTimeoutMs;
+export const unreadSteeredMessageIds = runner.unreadSteeredMessageIds;
 export const executeNativeSession = runner.executeNativeSession;
 export const applyNativeSessionGoalControl =
   runner.applyNativeSessionGoalControl;