From 78f04e2f2a9cd08a5a32e916d50771ce87aef695 Mon Sep 17 00:00:00 2001
From: Brave <zxd-666@users.noreply.github.com>
Date: Mon, 2 Nov 2026 23:45:04 +0800
Subject: [PATCH 01/15] fix(omp): wait for native background work instead of
 faulting a settled Turn
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

A terminal agent_end can arrive while a native background job keeps the
session Streaming. #confirmSettledTurn treated that get_state answer as a
protocol fault: the healthy Turn completed as failed, which paused the
external follow-up queue forever (interrupted/failed Turns never drain)
and left Desktop showing a queued follow-up that only a manual resend
could deliver.

Keep the Turn pending and reconfirm on the native session_settled
notification — the same retry signal the cancellation path already uses.
A still-busy session still never settles the Turn early; only a genuinely
confirmable get_state does.
---
 packages/adapters/omp/src/omp-rpc-session.ts  | 19 +++-
 .../adapters/omp/test/omp-rpc-session.test.ts | 89 +++++++++++++++++++
 2 files changed, 104 insertions(+), 4 deletions(-)

diff --git a/packages/adapters/omp/src/omp-rpc-session.ts b/packages/adapters/omp/src/omp-rpc-session.ts
index b5ded396c..61958c01a 100644
--- a/packages/adapters/omp/src/omp-rpc-session.ts
+++ b/packages/adapters/omp/src/omp-rpc-session.ts
@@ -1194,6 +1194,14 @@ export class OmpRpcSession {
       void this.#confirmCancelledTurn(active);
       return;
     }
+    if (value.type === "session_settled" && active?.settlement === "pending") {
+      // A terminal agent_end can arrive while a native background job keeps the
+      // session Streaming. That blocked the first confirmation; the native
+      // settle notification is the retry signal. Reconfirm serially through the
+      // same path so a still-busy session never settles the Turn early.
+      void this.#confirmSettledTurn(active);
+      return;
+    }
     if (!active) {
       if (value.type === "extension_ui_request" && this.#isBlockingInteraction(value)) {
         this.#fail(
@@ -1807,10 +1815,13 @@ export class OmpRpcSession {
         return;
       }
       if (parseSessionStreaming(response)) {
-        throw new OmpRpcFaultError(
-          "protocolError",
-          "Omp RPC completed Turn state is still Streaming",
-        );
+        // A terminal agent_end can precede native background work draining:
+        // isStreaming stays true while those jobs run even though the Turn's
+        // own run ended. Waiting for the native session_settled notification
+        // (which reconfirms here) is correct; faulting would kill a healthy
+        // Turn whose queued follow-ups then pause forever.
+        active.settlement = "pending";
+        return;
       }
       this.#state = state;
       active.settlement = "confirmed";
diff --git a/packages/adapters/omp/test/omp-rpc-session.test.ts b/packages/adapters/omp/test/omp-rpc-session.test.ts
index 483d5bc01..8dabd1568 100644
--- a/packages/adapters/omp/test/omp-rpc-session.test.ts
+++ b/packages/adapters/omp/test/omp-rpc-session.test.ts
@@ -441,6 +441,95 @@ describe("OMP RPC session", () => {
     },
   );
 
+  it("waits for background work instead of faulting when agent_end lands while a background job still streams", async () => {
+    vi.useFakeTimers();
+    const process = new FakeOmpProcess();
+    const onFault = vi.fn();
+    let streaming = true;
+    let settled = false;
+    process.handleCommand = (command) => {
+      if (command.type === "prompt") {
+        process.sendFrame({ type: "response", id: command.id, command: "prompt", success: true });
+        const message = {
+          role: "assistant",
+          responseId: "assistant-1",
+          content: [{ type: "text", text: "PONG" }],
+        };
+        process.sendFrame({ type: "message_start", message });
+        process.sendFrame({
+          type: "message_update",
+          message,
+          assistantMessageEvent: { type: "text_delta", delta: "PONG" },
+        });
+        process.sendFrame({ type: "message_end", message: { ...message, stopReason: "stop" } });
+        // A native background job is still polling: agent_end is terminal, yet
+        // get_state keeps reporting isStreaming until the job settles.
+        process.sendFrame({ type: "agent_end", isTerminal: true });
+        return true;
+      }
+      if (command.type === "get_state") {
+        process.sendFrame({
+          type: "response",
+          id: command.id,
+          command: "get_state",
+          success: true,
+          data: {
+            sessionId: "omp-session",
+            model: { provider: "synthetic", id: "omp-model" },
+            isStreaming: streaming,
+            isSettled: settled,
+            hasPendingAsyncWork: streaming,
+          },
+        });
+        return true;
+      }
+      return false;
+    };
+    const session = new OmpRpcSession(
+      { cwd: "/synthetic", onFault },
+      { spawn: () => process as never },
+    );
+    try {
+      await session.start();
+      let result: unknown;
+      const turn = session.runTurn("background job keeps streaming", () => undefined);
+      void turn.then(
+        (value) => {
+          result = value;
+        },
+        () => undefined,
+      );
+      // First settle check observes the still-running background job.
+      await vi.advanceTimersByTimeAsync(0);
+      expect(result).toBeUndefined();
+      expect(onFault).not.toHaveBeenCalled();
+      // The background job settles natively; session_settled reconfirms.
+      streaming = false;
+      settled = true;
+      process.sendFrame({ type: "session_settled" });
+      await vi.advanceTimersByTimeAsync(0);
+      await vi.waitFor(
+        () =>
+          expect(result).toEqual({
+            text: "PONG",
+            cancelled: false,
+            agentInvoked: true,
+          }),
+        { timeout: 200 },
+      );
+      expect(onFault).not.toHaveBeenCalled();
+      process.handleCommand = null;
+      await expect(session.runTurn("ping", () => undefined)).resolves.toEqual({
+        text: "PONG",
+        cancelled: false,
+        agentInvoked: true,
+      });
+    } finally {
+      vi.useRealTimers();
+      await session.close();
+    }
+  });
+
   it("waits through slow native cancellation and accepts another Turn afterwards", async () => {
     vi.useFakeTimers();
     const process = new FakeOmpProcess();

From cb3bcbc55052f0d1104c76f6e9125f94bdb58760 Mon Sep 17 00:00:00 2001
From: Brave <zxd-666@users.noreply.github.com>
Date: Tue, 3 Nov 2026 04:00:19 +0800
Subject: [PATCH 02/15] fix(host): project the user message of a Desktop
 turn/start that carries a clientUserMessageId

Desktop 26.1027 routes follow-up submissions through its main-process queue
manager. The Renderer no longer renders the submitted input locally for these
starts, while the Host still assumed it did ('Desktop renders the input of its
own turn/start locally'). Both sides skipped the user message, so a quickly
delivered follow-up was answered by the model but never appeared in the
conversation.

Project the input whenever the start carries a clientUserMessageId: the
projected userMessage carries that clientId, so a Renderer that still renders
optimally deduplicates instead of showing the message twice. Starts without a
clientId keep the old behavior. All external Harnesses share this path.
---
 packages/host-runtime/src/app-server-host.ts  |  9 ++-
 .../app-server-host.external-queue.test.ts    | 70 +++++++++++++++++++
 2 files changed, 78 insertions(+), 1 deletion(-)

diff --git a/packages/host-runtime/src/app-server-host.ts b/packages/host-runtime/src/app-server-host.ts
index a9d77bd19..648c4f656 100644
--- a/packages/host-runtime/src/app-server-host.ts
+++ b/packages/host-runtime/src/app-server-host.ts
@@ -4769,13 +4769,20 @@ export class AppServerHost {
       await this.#writer.json(rpcError(request, -32602, errorMessage(error)));
       return;
     }
+    const clientUserMessageId =
+      typeof params.clientUserMessageId === "string" ? params.clientUserMessageId : undefined;
     try {
       const started = await this.#beginExternalInputTurn(
         thread,
         text,
         undefined,
-        typeof params.clientUserMessageId === "string" ? params.clientUserMessageId : undefined,
+        clientUserMessageId,
         params[THREAD_REFERENCE_SOURCE_HOST_PARAM],
+        // Desktop 26.1027 routes follow-up submissions through its main-process
+        // queue manager, which no longer renders the submitted input locally in
+        // the Renderer. Project the user message so the Turn visibly carries its
+        // input; the clientId lets a Renderer that still renders optimally dedupe.
+        clientUserMessageId !== undefined,
       );
       try {
         await this.#writer.json(rpcEnvelope(request, { result: { turn: started.turn } }));
diff --git a/packages/host-runtime/test/app-server-host.external-queue.test.ts b/packages/host-runtime/test/app-server-host.external-queue.test.ts
index 116bf0835..d71eae0b1 100644
--- a/packages/host-runtime/test/app-server-host.external-queue.test.ts
+++ b/packages/host-runtime/test/app-server-host.external-queue.test.ts
@@ -234,4 +234,74 @@ describe("External Thread follow-up queue", () => {
       await stopFixture(fixture);
     }
   });
+
+  it("projects the submitted input of a Desktop turn/start that carries a clientUserMessageId", async () => {
+    const fixture = createFixture();
+    try {
+      const threadId = await startPiThread(fixture);
+      // Desktop 26.1027's main-process queue manager submits follow-ups with a
+      // clientUserMessageId but no longer renders the input locally in the
+      // Renderer; the Host must project the user message with that clientId.
+      writeRequest(fixture.desktopInput, {
+        id: 2,
+        method: "turn/start",
+        params: {
+          threadId,
+          input: text("follow-up from the main-process queue"),
+          clientUserMessageId: "desktop-follow-up-1",
+        },
+      });
+      const started = await fixture.collector.waitFor((message) => requestId(message, 2));
+      expect(started).toMatchObject({ result: { turn: { status: "inProgress" } } });
+      await expect(
+        fixture.collector.waitFor(
+          (message) =>
+            method(message, "item/started") &&
+            (messageParams(message).item as JsonObject).type === "userMessage",
+        ),
+      ).resolves.toMatchObject({
+        params: {
+          item: {
+            clientId: "desktop-follow-up-1",
+            content: [{ type: "text", text: "follow-up from the main-process queue" }],
+          },
+        },
+      });
+      const session = fixture.adapter.sessions[0];
+      if (!session) throw new Error("Fake Session was not opened");
+      session.succeedTurn();
+    } finally {
+      await stopFixture(fixture);
+    }
+  });
+
+  it("does not project a synthetic user message for starts without a clientUserMessageId", async () => {
+    const fixture = createFixture();
+    try {
+      const threadId = await startPiThread(fixture);
+      writeRequest(fixture.desktopInput, {
+        id: 2,
+        method: "turn/start",
+        params: { threadId, input: text("host-only start") },
+      });
+      const started = await fixture.collector.waitFor((message) => requestId(message, 2));
+      const turn = (started.result as JsonObject)?.turn as JsonObject | undefined;
+      const turnId = typeof turn?.id === "string" ? turn.id : undefined;
+      if (!turnId) throw new Error("Synthetic turn response has no ID");
+      const session = fixture.adapter.sessions[0];
+      if (!session) throw new Error("Fake Session was not opened");
+      session.appendText("ack");
+      session.succeedTurn();
+      await fixture.collector.waitFor((message) => turnEvent(message, "turn/completed", turnId));
+      // No userMessage item may be projected for a start Desktop still renders itself.
+      const userMessages = fixture.collector.messages.filter(
+        (message) =>
+          method(message, "item/started") &&
+          (messageParams(message).item as JsonObject)?.type === "userMessage",
+      );
+      expect(userMessages).toEqual([]);
+    } finally {
+      await stopFixture(fixture);
+    }
+  });
 });

From 66a0469d15225b7bb8d84e0684efcc36051d1cf2 Mon Sep 17 00:00:00 2001
From: Brave <zxd-666@users.noreply.github.com>
Date: Tue, 3 Nov 2026 09:16:51 +0800
Subject: [PATCH 03/15] fix(host): keep the follow-up queue flowing when a
 steer replacement interrupts the Turn
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Desktop's steer setting routes a follow-up sent mid-reply through
stop-then-start: the running Turn is cancelled and the user's input starts
as the next Turn. The cancelled Turn projected as interrupted, and inter-
rupted Turns pause the follow-up queue by design — so a steer left the
queue paused with queued messages stuck until the user pressed Continue,
even though the interruption was the user changing direction, not a stop.

A stop-then-start replacement now marks its cancellation: when that exact
Turn reaches its interrupted terminal, the queue drains instead of
pausing (the replacement Turn carries the user's new input; anything
still queued follows it). The paused-queue semantics for genuine user
stops and failures are unchanged.
---
 packages/host-runtime/src/app-server-host.ts  | 12 +++--
 .../src/external-turn-steering.ts             |  9 ++++
 .../app-server-host.external-queue.test.ts    | 54 +++++++++++++++++++
 3 files changed, 72 insertions(+), 3 deletions(-)

diff --git a/packages/host-runtime/src/app-server-host.ts b/packages/host-runtime/src/app-server-host.ts
index 648c4f656..525fc39be 100644
--- a/packages/host-runtime/src/app-server-host.ts
+++ b/packages/host-runtime/src/app-server-host.ts
@@ -5567,12 +5567,18 @@ export class AppServerHost {
           ? { type: "active", activeFlags: [] }
           : { type: "idle" },
       );
+      // A stop-then-start steer cancels this Turn itself; the terminal it produces is
+      // the user changing direction, not a stop Desktop chose to pause behind.
+      const replacedBySteer = this.#externalSteering.isReplacing(thread.id, event.turnId);
       this.#externalSteering.terminal(thread.id, event.turnId, event.outcome);
-      // Interrupted and failed Turns pause the queue until Desktop starts it again.
+      // Interrupted and failed Turns pause the queue until Desktop starts it again —
+      // except when the interruption is the steer replacement itself, which immediately
+      // starts the user's new input as the next Turn and must keep the queue flowing.
       if (
         !ephemeralTurn &&
-        result.completedTurn?.status === "completed" &&
-        this.#externalQueue.has(thread.id)
+        this.#externalQueue.has(thread.id) &&
+        (result.completedTurn?.status === "completed" ||
+          (replacedBySteer && result.completedTurn?.status === "interrupted"))
       ) {
         void this.#drainExternalQueue(thread);
       }
diff --git a/packages/host-runtime/src/external-turn-steering.ts b/packages/host-runtime/src/external-turn-steering.ts
index 2f818ac16..637db859e 100644
--- a/packages/host-runtime/src/external-turn-steering.ts
+++ b/packages/host-runtime/src/external-turn-steering.ts
@@ -218,6 +218,15 @@ export class ExternalTurnSteering {
     return threadId === undefined ? this.#pending.size > 0 : this.#pending.has(threadId);
   }
 
+  /**
+   * Whether a stop-then-start replacement is cancelling this exact Turn right now.
+   * The interrupted terminal such a replacement produces is the user changing
+   * direction, not a failure: the queue must not stay paused for it.
+   */
+  isReplacing(threadId: string, turnId: string): boolean {
+    return this.#pending.get(threadId)?.turnId === turnId;
+  }
+
   run(
     thread: SteeringThread,
     params: JsonObject,
diff --git a/packages/host-runtime/test/app-server-host.external-queue.test.ts b/packages/host-runtime/test/app-server-host.external-queue.test.ts
index d71eae0b1..72d1ac84a 100644
--- a/packages/host-runtime/test/app-server-host.external-queue.test.ts
+++ b/packages/host-runtime/test/app-server-host.external-queue.test.ts
@@ -304,4 +304,58 @@ describe("External Thread follow-up queue", () => {
       await stopFixture(fixture);
     }
   });
+
+  it("keeps the queue flowing after a steer replacement interrupts the running Turn", async () => {
+    const fixture = createFixture();
+    // No native steer: turn/steer takes the stop-then-start path.
+    fixture.adapter.supportsSteer = false;
+    try {
+      const threadId = await startPiThread(fixture);
+      const runningTurnId = await startPiTurn(fixture, threadId);
+      const session = fixture.adapter.sessions[0];
+      if (!session) throw new Error("Fake Session was not opened");
+      const execute = vi.spyOn(session, "execute");
+
+      // A follow-up queues while the Turn is running.
+      writeRequest(fixture.desktopInput, {
+        id: 10,
+        method: "thread/queue/add",
+        params: { threadId, input: text("queued during steer"), clientUserMessageId: "q-1" },
+      });
+      await fixture.collector.waitFor((message) => requestId(message, 10));
+
+      // The user steers: the running Turn is interrupted and the new input replaces it.
+      writeRequest(fixture.desktopInput, {
+        id: 11,
+        method: "turn/steer",
+        params: {
+          threadId,
+          expectedTurnId: runningTurnId,
+          clientUserMessageId: "steer-1",
+          input: text("replacement direction"),
+        },
+      });
+      // turn.cancel arrives; the Fake session completes the cancellation on request.
+      session.completeCancellationOnRequest();
+      await fixture.collector.waitFor((message) =>
+        turnEvent(message, "turn/completed", runningTurnId),
+      );
+      // The replacement Turn starts and finishes normally.
+      await vi.waitFor(() =>
+        expect(execute).toHaveBeenCalledWith(
+          expect.objectContaining({ type: "turn.start", input: text("replacement direction") }),
+        ),
+      );
+      if (!session) throw new Error("Replacement session was not opened");
+      session.succeedTurn();
+      // The queued follow-up must still be delivered after the replacement Turn completes.
+      await vi.waitFor(() =>
+        expect(execute).toHaveBeenCalledWith(
+          expect.objectContaining({ type: "turn.start", input: text("queued during steer") }),
+        ),
+      );
+    } finally {
+      await stopFixture(fixture);
+    }
+  });
 });

From cd0a12f091a0ef4ca84de5ccab10baeed61f463c Mon Sep 17 00:00:00 2001
From: Brave <zxd-666@users.noreply.github.com>
Date: Tue, 3 Nov 2026 09:42:28 +0800
Subject: [PATCH 04/15] chore(host): log queue requests and drain decisions to
 the runtime log

Live diagnosis of the follow-up pipeline: Desktop's queue UI can disagree
with what actually runs, and the Host-side queue state was invisible. Every
thread/queue/* request and every turn-terminal drain decision now lands in
the bounded runtime log (~/.codexhost/logs/host-runtime-<pid>.log).
---
 packages/host-runtime/src/app-server-host.ts | 12 ++++++++++++
 1 file changed, 12 insertions(+)

diff --git a/packages/host-runtime/src/app-server-host.ts b/packages/host-runtime/src/app-server-host.ts
index 525fc39be..fbdee2d56 100644
--- a/packages/host-runtime/src/app-server-host.ts
+++ b/packages/host-runtime/src/app-server-host.ts
@@ -4804,6 +4804,12 @@ export class AppServerHost {
 
   async #handleExternalQueueRequest(request: JsonRpcRequest, threadId: string): Promise<void> {
     const queue = this.#externalQueue;
+    // Queue diagnostics: every Desktop queue request lands in the runtime log,
+    // which is the only observer of the follow-up pipeline when Desktop's own
+    // queue UI disagrees with what actually runs.
+    this.#options.diagnosticOutput.write(
+      `codexhost Host Runtime: queue ${request.method} thread=${threadId.slice(0, 8)} running=${this.#externalRuntime.get(threadId)?.running ?? "unknown"} queued=${queue.has(threadId)}\n`,
+    );
     try {
       const params = requestObject(request);
       const input = (): JsonValue[] => {
@@ -5571,6 +5577,9 @@ export class AppServerHost {
       // the user changing direction, not a stop Desktop chose to pause behind.
       const replacedBySteer = this.#externalSteering.isReplacing(thread.id, event.turnId);
       this.#externalSteering.terminal(thread.id, event.turnId, event.outcome);
+      this.#options.diagnosticOutput.write(
+        `codexhost Host Runtime: turn terminal thread=${thread.id.slice(0, 8)} status=${result.completedTurn?.status} steerReplacing=${replacedBySteer} queued=${this.#externalQueue.has(thread.id)}\n`,
+      );
       // Interrupted and failed Turns pause the queue until Desktop starts it again —
       // except when the interruption is the steer replacement itself, which immediately
       // starts the user's new input as the next Turn and must keep the queue flowing.
@@ -5580,6 +5589,9 @@ export class AppServerHost {
         (result.completedTurn?.status === "completed" ||
           (replacedBySteer && result.completedTurn?.status === "interrupted"))
       ) {
+        this.#options.diagnosticOutput.write(
+          `codexhost Host Runtime: draining queue thread=${thread.id.slice(0, 8)}\n`,
+        );
         void this.#drainExternalQueue(thread);
       }
     }

From 553840993963cc4fc4a822157a3eec65dca19cf9 Mon Sep 17 00:00:00 2001
From: Brave <zxd-666@users.noreply.github.com>
Date: Tue, 3 Nov 2026 10:20:24 +0800
Subject: [PATCH 05/15] fix(renderer): let external follow-ups reach the
 Host-served queue
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Desktop 26.903's local follow-up queue no longer exists: 26.1027 manages the
queue in its main process and serves Host-projected external Threads from the
codexhost Host's native thread/queue/* implementation. The renderer's
isEnabled patch still excluded external Threads from the server queue — which
now strands every follow-up in a queue Desktop can no longer execute locally
(messages sit above the composer forever; the runtime log shows Desktop only
polling thread/queue/list without ever sending thread/queue/add).

Let the gate pass external Threads through so their follow-ups reach the Host
queue. Native/unknown Threads and no-argument probes keep the original gate.
A non-empty legacy local queue (paused entries) still routes locally until it
drains, so paused follow-ups are not abandoned.
---
 .../src/renderer-external-queue.ts            | 11 ++++--
 .../test/renderer-external-queue.test.ts      | 39 ++++++++++++-------
 2 files changed, 33 insertions(+), 17 deletions(-)

diff --git a/packages/renderer-extension/src/renderer-external-queue.ts b/packages/renderer-extension/src/renderer-external-queue.ts
index d7fcd088e..e7c5f6c6b 100644
--- a/packages/renderer-extension/src/renderer-external-queue.ts
+++ b/packages/renderer-extension/src/renderer-external-queue.ts
@@ -3,9 +3,12 @@ function isRecord(value: unknown): value is Record<string, unknown> {
 }
 
 /**
- * Desktop 26.903.61454 still owns the local follow-up queue and its execution.
- * Its serverQueue.isEnabled(threadId) gate must exclude Host-projected external
- * Threads: the official app-server queue cannot execute a Harness Turn.
+ * Desktop 26.1027 manages the follow-up queue in its main process and serves
+ * Host-projected external Threads from the Host's native `thread/queue/*`
+ * implementation. The legacy `serverQueue.isEnabled(threadId)` gate must let
+ * those Threads through (the stock app-server cannot execute a Harness Turn,
+ * but the codexhost Host answers the same protocol natively); excluding them
+ * strands every follow-up in a queue Desktop can no longer execute locally.
  */
 export function installRendererExternalQueue(target: unknown): (() => void) | null {
   if (
@@ -35,7 +38,7 @@ export function installRendererExternalQueue(target: unknown): (() => void) | nu
         // Desktop in both newly created and hydrated conversation metadata.
         conversation.modelProvider === "codexhost"
       ) {
-        return false;
+        return true;
       }
     }
     // Preserve native/unknown Threads, feature flags, and no-argument probes.
diff --git a/packages/renderer-extension/test/renderer-external-queue.test.ts b/packages/renderer-extension/test/renderer-external-queue.test.ts
index d5efcbdb2..51f389d02 100644
--- a/packages/renderer-extension/test/renderer-external-queue.test.ts
+++ b/packages/renderer-extension/test/renderer-external-queue.test.ts
@@ -21,7 +21,10 @@ function fixture() {
   const local = new Map<string, QueuedMessage[]>();
   const remote = new Map<string, QueuedMessage[]>();
   const rpc = vi.fn(async (method: string, params: { threadId: string }) => {
-    if (conversations.get(params.threadId)?.modelProvider === "codexhost") {
+    // Desktop 26.1027 routes external Threads to the codexhost Host, which
+    // implements `thread/queue/*` natively — these requests succeed. Only a
+    // stock app-server connection would reject them (-32076).
+    if (conversations.get(params.threadId)?.modelProvider === "closed-source-only") {
       throw Object.assign(new Error(`External Thread does not support ${method}`), {
         code: -32076,
       });
@@ -149,8 +152,8 @@ function fixture() {
 
 afterEach(() => vi.unstubAllGlobals());
 
-describe("external follow-ups retain Desktop's local queue", () => {
-  it("queues another message during an external Turn without thread/queue/list or add", async () => {
+describe("external follow-ups use the Host-served queue", () => {
+  it("queues another message during an external Turn through the Host queue", async () => {
     const f = fixture();
     const adapter = f.install();
     try {
@@ -158,14 +161,17 @@ describe("external follow-ups retain Desktop's local queue", () => {
         f.coordinator.sendMessage({ conversationId: "external", message: f.message }),
       ).resolves.toEqual({ status: "queued", messageId: f.message.id });
       expect(f.coordinator.readMessages("external")).toEqual([f.message]);
-      expect(f.local.get("external")).toEqual([f.message]);
-      expect(f.rpc).not.toHaveBeenCalled();
+      expect(f.rpc.mock.calls).toEqual([
+        ["thread/queue/list", { threadId: "external" }],
+        ["thread/queue/add", { threadId: "external" }],
+      ]);
+      expect(f.local.has("external")).toBe(false);
     } finally {
       adapter.dispose();
     }
   });
 
-  it("preserves existing paused follow-ups and stays local after the queue drains", async () => {
+  it("keeps local paused follow-ups until they drain, then uses the Host queue", async () => {
     const f = fixture();
     const paused = {
       ...f.message,
@@ -175,12 +181,18 @@ describe("external follow-ups retain Desktop's local queue", () => {
     f.local.set("external", [paused]);
     const adapter = f.install();
     try {
+      // A non-empty legacy local queue still routes to Desktop's local queue:
+      // paused entries must not be abandoned on the Host-side queue.
       await f.coordinator.sendMessage({ conversationId: "external", message: f.message });
       expect(f.local.get("external")).toEqual([paused, f.message]);
       f.local.delete("external");
+      // Once the local queue is empty, the Host-served queue takes over.
       await f.coordinator.sendMessage({ conversationId: "external", message: f.message });
-      expect(f.local.get("external")).toEqual([f.message]);
-      expect(f.rpc).not.toHaveBeenCalled();
+      expect(f.rpc.mock.calls).toEqual([
+        ["thread/queue/list", { threadId: "external" }],
+        ["thread/queue/add", { threadId: "external" }],
+      ]);
+      expect(f.local.has("external")).toBe(false);
     } finally {
       adapter.dispose();
     }
@@ -197,7 +209,7 @@ describe("external follow-ups retain Desktop's local queue", () => {
       ]);
       expect(f.local.has("native")).toBe(false);
       expect(f.coordinator.readMessages("native")).toEqual([f.message]);
-      expect(f.queue.isEnabled("external")).toBe(false);
+      expect(f.queue.isEnabled("external")).toBe(true);
       expect(f.queue.isEnabled("native")).toBe(true);
       expect(f.originalGate.mock.contexts.every((receiver) => receiver === f.queue)).toBe(true);
     } finally {
@@ -210,10 +222,11 @@ describe("external follow-ups retain Desktop's local queue", () => {
     const other = fixture();
     const adapter = f.install();
     try {
-      expect(f.queue.isEnabled("external")).toBe(false);
+      expect(f.queue.isEnabled("external")).toBe(true);
       expect(other.queue.isEnabled("external")).toBe(true);
       f.conversations.set("native", { id: "native", modelProvider: "codexhost" });
-      expect(f.queue.isEnabled("native")).toBe(false);
+      expect(f.queue.isEnabled("native")).toBe(true);
+      expect(other.queue.isEnabled("native")).toBe(true);
       expect(f.queue.isEnabled("unloaded")).toBe(true);
       expect(f.queue.isEnabled()).toBe(true);
       f.originalGate.mockReturnValue(false);
@@ -227,7 +240,7 @@ describe("external follow-ups retain Desktop's local queue", () => {
     const f = fixture();
     for (let i = 0; i < 2; i++) {
       const adapter = f.install();
-      expect(f.queue.isEnabled("external")).toBe(false);
+      expect(f.queue.isEnabled("external")).toBe(true);
       adapter.dispose();
       expect(f.queue.isEnabled).toBe(f.originalGate);
       expect(Object.hasOwn(f.manager, "getTurnCoordinator")).toBe(false);
@@ -243,7 +256,7 @@ describe("external follow-ups retain Desktop's local queue", () => {
       f.conversations.delete("external");
       expect(f.queue.isEnabled("external")).toBe(true);
       f.conversations.set("external", { id: "external", modelProvider: "codexhost" });
-      expect(f.queue.isEnabled("external")).toBe(false);
+      expect(f.queue.isEnabled("external")).toBe(true);
       const replacement = vi.fn(() => false);
       f.queue.isEnabled = replacement;
       adapter.dispose();

From 6d364fe8f3a1d4a2f0cb7254c4a812cecbad59eb Mon Sep 17 00:00:00 2001
From: Brave <zxd-666@users.noreply.github.com>
Date: Tue, 3 Nov 2026 11:15:24 +0800
Subject: [PATCH 06/15] feat(omp): native same-Turn steering through OMP's
 steer RPC
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

OMP's own error message when a prompt arrives mid-run reads: 'Agent is
already processing. Use steer() or followUp() to queue messages' — the
native RPC has always had a steer primitive, but the adapter declared
turn.steer unsupported, so every mid-reply follow-up took the
stop-then-start path: the running Turn was cancelled (projecting as
interrupted), which paused Desktop's follow-up queue and left the paused
panel and spinner residue users see on Desktop 26.1027.

Deliver turn.steer through the native 'steer' RPC instead. The steered
text joins the same native run as a user message and the Turn keeps its
identity — the same experience native Codex threads have with Desktop's
steer follow-up setting. The steer capability is read live and excludes
Turns whose cancellation is already in flight; those keep the stop-and-
restart path. Transports without steer (older OMP) keep unsupportedSteer.

Validated against a live OMP 18.15.0 rpc-ui process: steer succeeds
mid-Turn, the text arrives as the run's second user message, one
agent_start/agent_end pair, no interruption.
---
 packages/adapters/omp/src/omp-adapter.ts      | 63 +++++++++++++++++--
 packages/adapters/omp/src/omp-rpc-session.ts  | 17 +++++
 .../adapters/omp/test/omp-rpc-session.test.ts | 49 +++++++++++++++
 3 files changed, 125 insertions(+), 4 deletions(-)

diff --git a/packages/adapters/omp/src/omp-adapter.ts b/packages/adapters/omp/src/omp-adapter.ts
index 69f63e1e5..55b51be50 100644
--- a/packages/adapters/omp/src/omp-adapter.ts
+++ b/packages/adapters/omp/src/omp-adapter.ts
@@ -160,6 +160,8 @@ export interface OmpTurnTransport {
     onEvent: (event: OmpTurnEvent) => void,
   ): Promise<OmpCompactResult>;
   runTurn(text: string, onEvent: (event: OmpTurnEvent) => void): Promise<OmpTurnResult>;
+  /** Native same-Turn input; present on transports backed by OMP's `steer` RPC. */
+  steer?(text: string): Promise<void>;
   respondToInteraction(response: OmpInteractionResponse): Promise<void>;
   abort(): Promise<void>;
   close(): Promise<void>;
@@ -718,16 +720,24 @@ class OmpHarnessSession implements HarnessSession {
     this.#toolOutputLimit = options.toolOutputLimit;
     this.#permissionMode = options.permissionMode;
     this.#permissionModeId = options.permissionModeId;
-    this.capabilities = {
+    this.capabilities = ((session) => ({
       configuration: {
         selectModel: true,
         selectThinkingOption: options.supportsThinkingSelection,
         selectPermissionMode: true,
-        permissionModeScope: "live",
+        permissionModeScope: "live" as const,
       },
       history: { fork: true, forkAcrossCwd: true, rollbackLastTurn: true },
       subagents: { observe: true, readTranscript: true },
-    };
+      // Read live: OMP's `steer` RPC delivers mid-Turn input into the same
+      // native run. A Turn whose cancellation is in flight must not take
+      // more input; every other case stops and restarts.
+      get steer(): boolean {
+        return (
+          session.#transport?.steer !== undefined && session.#active?.cancellationRequested !== true
+        );
+      },
+    }))(this);
     this.commands = {
       list: async () => ({ ok: true, value: this.#liveCommandCatalog() }),
       execute: (command) => this.#executeHarnessCommand(command),
@@ -898,7 +908,7 @@ class OmpHarnessSession implements HarnessSession {
       return { ok: false, error: invalidState("Omp Session is not open") };
     }
     if (command.type === "turn.cancel") return this.#cancel(command);
-    if (command.type === "turn.steer") return unsupportedSteer("OMP");
+    if (command.type === "turn.steer") return this.#steer(command);
     if (command.type === "interaction.respond") return this.#respond(command);
     if (command.type === "model.select") return this.#selectModel(command);
     if (command.type === "thinking.select") return this.#selectThinking(command);
@@ -1372,6 +1382,51 @@ class OmpHarnessSession implements HarnessSession {
     }
   }
 
+  /**
+   * Native same-Turn steering through OMP's `steer` RPC. OMP refuses a plain
+   * `prompt` while the agent runs, so without this a mid-Turn follow-up had to
+   * stop and restart the Turn, which surfaced as an interrupted Turn in the
+   * Desktop queue UI. The steered text arrives as a user message of the same
+   * native run; the turn keeps its identity.
+   */
+  async #steer(command: TurnSteerCommand): Promise<HarnessResult<TurnSteerAccepted>> {
+    const active = this.#active;
+    if (!active || active.command.turnId !== command.turnId) {
+      return {
+        ok: false,
+        error: invalidState("Omp steer must reference the active Turn"),
+      };
+    }
+    if (active.cancellationRequested) {
+      return {
+        ok: false,
+        error: invalidState("Omp steer target is being cancelled"),
+      };
+    }
+    const text = command.input.map((input) => input.text).join("\n");
+    if (text.length === 0) {
+      return {
+        ok: false,
+        error: {
+          code: "invalidRequest",
+          message: "Omp steer text must not be empty",
+          retryable: false,
+        },
+      };
+    }
+    const transport = this.#transport;
+    if (!transport?.steer) return unsupportedSteer("OMP");
+    try {
+      await transport.steer(text);
+    } catch (error) {
+      return {
+        ok: false,
+        error: invalidState(`Omp native steer failed: ${errorMessage(error)}`),
+      };
+    }
+    return { ok: true, value: { accepted: true } };
+  }
+
   async #cancel(command: TurnCancelCommand): Promise<HarnessResult<TurnCancelAccepted>> {
     const active = this.#active;
     if (!active || active.command.turnId !== command.turnId) {
diff --git a/packages/adapters/omp/src/omp-rpc-session.ts b/packages/adapters/omp/src/omp-rpc-session.ts
index 61958c01a..4c4cec345 100644
--- a/packages/adapters/omp/src/omp-rpc-session.ts
+++ b/packages/adapters/omp/src/omp-rpc-session.ts
@@ -952,6 +952,23 @@ export class OmpRpcSession {
     return settled;
   }
 
+  /**
+   * Native same-Turn input through OMP's `steer` RPC. OMP itself refuses a plain
+   * `prompt` while an agent runs ("Use steer() or followUp()"), so this is the
+   * only way to deliver mid-Turn input without stopping the Turn.
+   */
+  async steer(text: string): Promise<void> {
+    if (!this.#child || !this.#state || this.#closed || this.#failed) {
+      throw new Error("Omp RPC Session is unavailable");
+    }
+    if (text.length === 0) throw new Error("Omp steer text must not be empty");
+    try {
+      await this.#send("steer", { message: text });
+    } catch (error) {
+      throw error instanceof Error ? error : new Error(message(error));
+    }
+  }
+
   respondToInteraction(response: OmpInteractionResponse): Promise<void> {
     return this.#resolveInteraction(response, "cancelled" in response ? "cancelled" : "responded");
   }
diff --git a/packages/adapters/omp/test/omp-rpc-session.test.ts b/packages/adapters/omp/test/omp-rpc-session.test.ts
index 8dabd1568..20d51bf6b 100644
--- a/packages/adapters/omp/test/omp-rpc-session.test.ts
+++ b/packages/adapters/omp/test/omp-rpc-session.test.ts
@@ -202,6 +202,7 @@ class FakeOmpProcess extends EventEmitter {
     }
     if (command.type === "negotiate_protocol")
       return this.#response(command, { protocolVersion: 2 });
+    if (command.type === "steer") return this.#response(command);
     if (command.type === "get_state") return this.#response(command, this.#state());
     if (command.type === "get_messages")
       return this.#response(command, { messages: this.messages });
@@ -1286,6 +1287,54 @@ describe("OMP RPC session", () => {
     }
   });
 
+  it("delivers a native steer into the running Turn without stopping it", async () => {
+    const process = new FakeOmpProcess();
+    // Keep the first Turn running: no automatic PONG stream, we settle it manually.
+    process.promptHandler = (command) => {
+      process.sendFrame({ type: "response", id: command.id, command: "prompt", success: true });
+    };
+    const adapter: OmpRpcProcessAdapter = { spawn: () => process as never };
+    const session = new OmpRpcSession({ cwd: "/synthetic", commandTimeoutMs: 2_000 }, adapter);
+    try {
+      await session.start();
+      const turn = session.runTurn("first", () => undefined);
+      await vi.waitFor(() => expect(process.commands.some((c) => c.type === "prompt")).toBe(true));
+      // A steer while the Turn runs goes through OMP's `steer` RPC; the Turn
+      // itself keeps running (no abort, no second prompt).
+      await expect(session.steer("changed my mind")).resolves.toBeUndefined();
+      const steerCommand = process.commands.find((c) => c.type === "steer");
+      expect(steerCommand).toMatchObject({ message: "changed my mind" });
+      expect(process.commands.filter((c) => c.type === "prompt")).toHaveLength(1);
+      expect(process.commands.some((c) => c.type === "abort")).toBe(false);
+      // The Turn settles normally afterwards.
+      process.sendFrame({
+        type: "message_start",
+        message: {
+          role: "assistant",
+          responseId: "steered",
+          content: [{ type: "text", text: "steered answer" }],
+        },
+      });
+      process.sendFrame({
+        type: "message_end",
+        message: {
+          role: "assistant",
+          responseId: "steered",
+          content: [{ type: "text", text: "steered answer" }],
+          stopReason: "stop",
+        },
+      });
+      process.sendFrame({ type: "agent_end", isTerminal: true });
+      await expect(turn).resolves.toEqual({
+        text: "steered answer",
+        cancelled: false,
+        agentInvoked: true,
+      });
+    } finally {
+      await session.close();
+    }
+  });
+
   it("starts through ready/negotiation and settles a streamed text turn on agent_end", async () => {
     const process = new FakeOmpProcess();
     const adapter: OmpRpcProcessAdapter = { spawn: () => process as never };

From 02604f7b537003f86a3b1df7aa835f7f94939008 Mon Sep 17 00:00:00 2001
From: Brave <zxd-666@users.noreply.github.com>
Date: Tue, 3 Nov 2026 12:01:42 +0800
Subject: [PATCH 07/15] feat(pi): native same-Turn steering through Pi's steer
 RPC
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Same family and same UX gap as OMP: Pi's RPC answers a mid-run prompt
refusal with 'Use steer() or followUp()', and the adapter declared
turn.steer unsupported, forcing every mid-reply follow-up through
stop-then-start (interrupted Turn, paused queue panel, spinner residue
on Desktop 26.1027).

Deliver turn.steer through the native 'steer' RPC. The response carries
a queued admission disposition and the steered text joins the running
native run as its next user message; the Turn keeps its identity — the
same experience native Codex threads have. The capability is read live
and excludes Turns whose cancellation is in flight. Transports without
steer keep unsupportedSteer.

Validated against a live Pi rpc process: steer mid-Turn answers
{disposition: 'queued'}, the text arrives as the run's second user
message, one agent_start, no interruption.
---
 packages/adapters/pi/src/pi-adapter.ts        | 55 +++++++++++++++++--
 packages/adapters/pi/src/pi-rpc-session.ts    | 21 +++++++
 .../adapters/pi/test/pi-rpc-session.test.ts   | 42 ++++++++++++++
 3 files changed, 114 insertions(+), 4 deletions(-)

diff --git a/packages/adapters/pi/src/pi-adapter.ts b/packages/adapters/pi/src/pi-adapter.ts
index 40749ca96..a85b2ff81 100644
--- a/packages/adapters/pi/src/pi-adapter.ts
+++ b/packages/adapters/pi/src/pi-adapter.ts
@@ -175,6 +175,8 @@ export interface PiTurnTransport {
     onEvent: (event: PiTurnEvent) => void,
   ): Promise<PiCompactResult>;
   runTurn(text: string, onEvent: (event: PiTurnEvent) => void): Promise<PiTurnResult>;
+  /** Native same-Turn input; present on transports backed by Pi's `steer` RPC. */
+  steer?(text: string): Promise<void>;
   respondToInteraction(response: PiInteractionResponse): Promise<void>;
   abort(): Promise<void>;
   close(): Promise<void>;
@@ -453,17 +455,25 @@ class PiHarnessSession implements HarnessSession {
     this.#requestedThinkingOptionId = options.thinkingOptionId;
     this.#toolOutputLimit = options.toolOutputLimit;
     this.#isFastSupported = options.isFastSupported;
-    this.capabilities = {
+    this.capabilities = ((session) => ({
       configuration: {
         selectModel: true,
         selectThinkingOption: options.supportsThinkingSelection,
         selectPermissionMode: false,
-        permissionModeScope: "live",
+        permissionModeScope: "live" as const,
       },
       history: { fork: true, forkAcrossCwd: true, rollbackLastTurn: true },
       autonomousTurns: { observe: true },
       subagents: { observe: true, readTranscript: true },
-    };
+      // Read live: Pi's `steer` RPC delivers mid-Turn input into the same
+      // native run. A Turn whose cancellation is in flight must not take more
+      // input; every other case stops and restarts.
+      get steer(): boolean {
+        return (
+          session.#transport?.steer !== undefined && session.#active?.cancellationRequested !== true
+        );
+      },
+    }))(this);
     this.commands = {
       list: async () => ({ ok: true, value: await this.#refreshLiveCommands() }),
       execute: (command) => this.#executeHarnessCommand(command),
@@ -587,7 +597,7 @@ class PiHarnessSession implements HarnessSession {
     if (this.#phase !== "open") {
       return { ok: false, error: invalidState("Pi Session is not open") };
     }
-    if (command.type === "turn.steer") return unsupportedSteer("Pi");
+    if (command.type === "turn.steer") return this.#steer(command);
     if (command.type === "turn.cancel") return this.#cancel(command);
     if (command.type === "interaction.respond") return this.#respond(command);
     if (command.type === "model.select") return this.#selectModel(command);
@@ -955,6 +965,43 @@ class PiHarnessSession implements HarnessSession {
     }
   }
 
+  /**
+   * Native same-Turn steering through Pi's `steer` RPC. The steered text joins
+   * the running native run as its next user message; the Turn keeps its
+   * identity, matching native Codex threads under Desktop's steer setting.
+   */
+  async #steer(command: TurnSteerCommand): Promise<HarnessResult<TurnSteerAccepted>> {
+    const active = this.#active;
+    if (!active || active.command.turnId !== command.turnId) {
+      return { ok: false, error: invalidState("Pi steer must reference the active Turn") };
+    }
+    if (active.cancellationRequested) {
+      return { ok: false, error: invalidState("Pi steer target is being cancelled") };
+    }
+    const text = command.input.map((input) => input.text).join("\n");
+    if (text.length === 0) {
+      return {
+        ok: false,
+        error: {
+          code: "invalidRequest",
+          message: "Pi steer text must not be empty",
+          retryable: false,
+        },
+      };
+    }
+    const transport = this.#transport;
+    if (!transport?.steer) return unsupportedSteer("Pi");
+    try {
+      await transport.steer(text);
+    } catch (error) {
+      return {
+        ok: false,
+        error: invalidState(`Pi native steer failed: ${errorMessage(error)}`),
+      };
+    }
+    return { ok: true, value: { accepted: true } };
+  }
+
   async #cancel(command: TurnCancelCommand): Promise<HarnessResult<TurnCancelAccepted>> {
     const active = this.#active;
     if (!active || active.command.turnId !== command.turnId) {
diff --git a/packages/adapters/pi/src/pi-rpc-session.ts b/packages/adapters/pi/src/pi-rpc-session.ts
index 344d1fec5..b58cc9e58 100644
--- a/packages/adapters/pi/src/pi-rpc-session.ts
+++ b/packages/adapters/pi/src/pi-rpc-session.ts
@@ -1056,6 +1056,27 @@ export class PiRpcSession {
     return settled;
   }
 
+  /**
+   * Native same-Turn input through Pi's `steer` RPC. The response carries an
+   * admission disposition ("queued"); the steered text joins the running native
+   * run as its next user message without starting a new run.
+   */
+  async steer(text: string): Promise<void> {
+    if (!this.#child || !this.#state || this.#closed || this.#failed) {
+      throw new Error("Pi RPC Session is unavailable");
+    }
+    if (text.length === 0) throw new Error("Pi steer text must not be empty");
+    const active = this.#activeTurn;
+    if (!active || active.origin !== "requested") {
+      throw new Error("Pi steer must reference the active Turn");
+    }
+    try {
+      await this.#send("steer", { message: text });
+    } catch (error) {
+      throw error instanceof Error ? error : new Error(message(error));
+    }
+  }
+
   respondToInteraction(response: PiInteractionResponse): Promise<void> {
     return this.#resolveInteraction(response, "cancelled" in response ? "cancelled" : "responded");
   }
diff --git a/packages/adapters/pi/test/pi-rpc-session.test.ts b/packages/adapters/pi/test/pi-rpc-session.test.ts
index 760c6c361..08169c798 100644
--- a/packages/adapters/pi/test/pi-rpc-session.test.ts
+++ b/packages/adapters/pi/test/pi-rpc-session.test.ts
@@ -80,6 +80,7 @@ class FakePiRpcProcess extends EventEmitter {
   signalCode: NodeJS.Signals | null = null;
   fastEnabled = false;
   readonly commands: string[] = [];
+  readonly commandFrames: Record<string, unknown>[] = [];
   holdEntries = false;
   holdStates = false;
   holdAborts = false;
@@ -275,10 +276,15 @@ class FakePiRpcProcess extends EventEmitter {
     const command = value as Record<string, unknown>;
     if (typeof command.id !== "string" || typeof command.type !== "string") return;
     this.commands.push(command.type);
+    this.commandFrames.push(command);
     if (command.type === "abort" && this.holdAborts) {
       this.#heldAborts.push(command);
       return;
     }
+    if (command.type === "steer") {
+      this.#respond(command, { disposition: "queued" });
+      return;
+    }
     if (command.type === "get_state" && this.holdStates) {
       this.#heldStates.push(command);
       return;
@@ -1547,6 +1553,42 @@ describe("Pi RPC Turn aggregation", () => {
     },
   );
 
+  it("delivers a native steer into the running Turn without stopping it", async () => {
+    const fake = new FakePiRpcProcess("manual-run");
+    const rpc = new PiRpcSession(
+      { cwd: process.cwd(), commandTimeoutMs: 2_000, closeTimeoutMs: 500 },
+      { spawn: () => fake as unknown as ChildProcessWithoutNullStreams },
+    );
+    await rpc.start();
+    try {
+      const pending = rpc.runTurn("synthetic", () => undefined);
+      await vi.waitFor(() => expect(fake.commands.filter((c) => c === "prompt")).toHaveLength(1));
+      // The Turn is running; a native steer goes through Pi's `steer` RPC and
+      // reports a queued disposition. No abort, no second prompt.
+      await expect(rpc.steer("changed my mind")).resolves.toBeUndefined();
+      const steer = fake.commandFrames.find((c) => c.type === "steer");
+      expect(steer).toMatchObject({ message: "changed my mind" });
+      expect(fake.commands.filter((c) => c === "prompt")).toHaveLength(1);
+      expect(fake.commands.some((c) => c === "abort")).toBe(false);
+      // The Turn settles normally afterwards.
+      const message = {
+        role: "assistant",
+        responseId: "steered",
+        content: [{ type: "text", text: "steered answer" }],
+        stopReason: "stop",
+      };
+      fake.emitFrame({ type: "message_start", message });
+      fake.emitFrame({ type: "message_end", message });
+      fake.emitAgentSettled();
+      await expect(pending).resolves.toMatchObject({
+        text: "steered answer",
+        cancelled: false,
+      });
+    } finally {
+      await rpc.close();
+    }
+  });
+
   it("bounds idle progress without faulting, coalesces status and discloses omitted updates", async () => {
     const fake = new FakePiRpcProcess("manual-run");
     const onFault = vi.fn();

From 83e13a95967e33e7c9a54d8f135b83a67ffc5f0e Mon Sep 17 00:00:00 2001
From: Brave <zxd-666@users.noreply.github.com>
Date: Tue, 3 Nov 2026 12:35:57 +0800
Subject: [PATCH 08/15] test(workbuddy): pin the bundle layout in macOS
 discovery cases

The discovery test's isExecutable mock admits /Applications/WorkBuddy.app,
which exists on development machines with a real (newer, lite-bundle)
installation. Discovery then probes the real cli-dist files and appends the
lite bundle's --serve arguments, failing the legacy-shape assertion. Pin
isFile to report no CLI files so the case asserts the legacy invocation
regardless of what is installed locally.
---
 packages/adapters/workbuddy/test/discovery.test.ts | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/packages/adapters/workbuddy/test/discovery.test.ts b/packages/adapters/workbuddy/test/discovery.test.ts
index e02f51375..8dc9e400a 100644
--- a/packages/adapters/workbuddy/test/discovery.test.ts
+++ b/packages/adapters/workbuddy/test/discovery.test.ts
@@ -111,6 +111,9 @@ describe("WorkBuddy app discovery", () => {
     const invocation = workBuddyInvocation({ HOME: "/Users/test" }, false, {
       platform: "darwin",
       isExecutable: (candidate) => [executable, cli].includes(candidate),
+      // Pin the bundle layout: no lite CLI files, so discovery must not probe a
+      // real /Applications installation that happens to exist on this machine.
+      isFile: () => false,
     });
     expect(invocation.command).toBe(executable);
     expect(invocation.arguments).toEqual(["--eval", expect.any(String), cli, "--acp"]);

From a1e3b1c7c0285e2748feff89a7dda260a8502713 Mon Sep 17 00:00:00 2001
From: Brave <zxd-666@users.noreply.github.com>
Date: Tue, 3 Nov 2026 13:29:38 +0800
Subject: [PATCH 09/15] fix(omp): latch settle notices and bind them to the
 Turn's completion fact
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Two races in the session_settled confirmation path (upstream review):

1. Lost wake-up: a settle notice landing while a confirmation was in
   flight (settlement=confirming) was dropped by the pending-only guard.
   When the in-flight snapshot then reported streaming, the Turn went
   back to pending with no notification left to wake it — it could wait
   forever. The notice is now latched at session level and consumed at
   response time: a notice that outran the snapshot drives one bounded
   serial recheck (<= 2) instead of waiting for another notification.

2. Stale notice settling a fresh Turn: session_settled carries no turn
   binding, so a late notice from the previous Turn could trigger a
   state check that found the session idle (the new prompt still in
   native admission) and settle a Turn that never ran. A notice may now
   only confirm a Turn holding its own completion fact — terminal
   agent_end observed or a local result.

Regression tests cover both: notice-first-snapshot race (fails without
the latch: Turn waits forever) and stale notice against a freshly
admitted Turn (fails without completion-fact binding).
---
 packages/adapters/omp/src/omp-rpc-session.ts  |  57 ++++++-
 .../adapters/omp/test/omp-rpc-session.test.ts | 146 ++++++++++++++++++
 2 files changed, 197 insertions(+), 6 deletions(-)

diff --git a/packages/adapters/omp/src/omp-rpc-session.ts b/packages/adapters/omp/src/omp-rpc-session.ts
index 4c4cec345..f5df74c7b 100644
--- a/packages/adapters/omp/src/omp-rpc-session.ts
+++ b/packages/adapters/omp/src/omp-rpc-session.ts
@@ -547,6 +547,19 @@ export class OmpRpcSession {
   #buffer: Buffer<ArrayBufferLike> = Buffer.alloc(0);
   #child: ChildProcessWithoutNullStreams | null = null;
   #closed = false;
+  /**
+   * A native session_settled notification arrived and has not driven a
+   * confirmation pass yet. Session-level: it must survive the gap between a
+   * notification and the state response that reflects it, and must never be
+   * dropped just because a confirmation is already in flight.
+   */
+  #settledNotice = false;
+  /**
+   * Confirmation passes triggered by a consumed settled notice, so a notice
+   * that outran the state snapshot rechecks once instead of waiting forever
+   * for a second notification that already fired.
+   */
+  #noticeConfirmations = 0;
   #closePromise: Promise<void> | null = null;
   #compactionActive = false;
   #compactionTurn: ActiveTurn | null = null;
@@ -1211,12 +1224,29 @@ export class OmpRpcSession {
       void this.#confirmCancelledTurn(active);
       return;
     }
-    if (value.type === "session_settled" && active?.settlement === "pending") {
-      // A terminal agent_end can arrive while a native background job keeps the
-      // session Streaming. That blocked the first confirmation; the native
-      // settle notification is the retry signal. Reconfirm serially through the
-      // same path so a still-busy session never settles the Turn early.
-      void this.#confirmSettledTurn(active);
+    if (value.type === "session_settled") {
+      // Session-level notice: record it even while a confirmation is in
+      // flight — the state snapshot that confirmation reads can still show
+      // streaming when the notice outran it, and no further notification may
+      // arrive. The in-flight pass consumes the notice and rechecks.
+      this.#settledNotice = true;
+      const turn = active;
+      if (
+        turn &&
+        (turn.settlement === "pending" || turn.settlement === "confirming") &&
+        // Completion fact binding: only a Turn that has seen its own terminal
+        // agent_end (or settled locally) may be settled by this notice. A
+        // stale notice from the previous Turn must not confirm a new Turn
+        // still waiting for native admission (it has no terminalEnd yet).
+        (turn.terminalEnd || turn.localResult !== null)
+      ) {
+        if (turn.settlement === "pending") void this.#confirmSettledTurn(turn);
+      }
+      if (!turn) {
+        // No active Turn owns the notice; it stays recorded for the next
+        // confirmation pass (a new Turn's terminal agent_end path).
+        return;
+      }
       return;
     }
     if (!active) {
@@ -1838,9 +1868,24 @@ export class OmpRpcSession {
         // (which reconfirms here) is correct; faulting would kill a healthy
         // Turn whose queued follow-ups then pause forever.
         active.settlement = "pending";
+        // Consume the notice NOW, at response time: one can land while this
+        // pass was suspended awaiting the snapshot (the snapshot predates the
+        // settle), and no further notification may arrive. The bounded recheck
+        // replaces the retry that dropped notice would have driven.
+        const noticeDriven = this.#settledNotice;
+        this.#settledNotice = false;
+        this.#noticeConfirmations = noticeDriven ? this.#noticeConfirmations + 1 : 0;
+        if (noticeDriven && this.#noticeConfirmations <= 2) {
+          setTimeout(() => {
+            if (this.#activeTurn === active && active.settlement === "pending")
+              void this.#confirmSettledTurn(active);
+          }, 50);
+        }
         return;
       }
+      this.#settledNotice = false;
       this.#state = state;
+      this.#noticeConfirmations = 0;
       active.settlement = "confirmed";
       this.#finishSettledTurn(active);
     } catch (error) {
diff --git a/packages/adapters/omp/test/omp-rpc-session.test.ts b/packages/adapters/omp/test/omp-rpc-session.test.ts
index 20d51bf6b..a13918c65 100644
--- a/packages/adapters/omp/test/omp-rpc-session.test.ts
+++ b/packages/adapters/omp/test/omp-rpc-session.test.ts
@@ -442,6 +442,152 @@ describe("OMP RPC session", () => {
     },
   );
 
+  it("rechecks once when the settle notice shares the moment with a still-streaming state", async () => {
+    const process = new FakeOmpProcess();
+    const onFault = vi.fn();
+    // The settle notice arrives while the get_state snapshot still shows the
+    // background job streaming; the state flips idle just after the chunk.
+    let streaming = true;
+    let turnSeen = false;
+    process.handleCommand = (command) => {
+      if (command.type === "prompt") {
+        turnSeen = true;
+        process.sendFrame({ type: "response", id: command.id, command: "prompt", success: true });
+        const message = {
+          role: "assistant",
+          responseId: "assistant-1",
+          content: [{ type: "text", text: "PONG" }],
+        };
+        process.sendFrame({ type: "message_start", message });
+        process.sendFrame({ type: "message_end", message: { ...message, stopReason: "stop" } });
+        process.sendFrame({ type: "agent_end", isTerminal: true });
+        return true;
+      }
+      if (command.type === "get_state") {
+        // The notice and the busy snapshot ride the same stdout chunk, with
+        // the notice FIRST: it lands while the Turn's confirmation is still
+        // in "confirming" awaiting this very response — whose snapshot was
+        // taken BEFORE the settle and must still say streaming. Without the
+        // notice latch, the dropped notification leaves the Turn waiting
+        // forever. Only the Turn's own confirmation queries participate
+        // (start() probes state too).
+        const snapshotStreaming = streaming;
+        if (turnSeen && streaming) {
+          process.sendFrame({ type: "session_settled" });
+          streaming = false;
+        }
+        process.sendFrame({
+          type: "response",
+          id: command.id,
+          command: "get_state",
+          success: true,
+          data: {
+            sessionId: "omp-session",
+            model: { provider: "synthetic", id: "omp-model" },
+            isStreaming: snapshotStreaming,
+            hasPendingAsyncWork: snapshotStreaming,
+          },
+        });
+        return true;
+      }
+      return false;
+    };
+    const session = new OmpRpcSession(
+      { cwd: "/synthetic", onFault },
+      { spawn: () => process as never },
+    );
+    try {
+      await session.start();
+      let result: unknown;
+      void session
+        .runTurn("notice outruns state", () => undefined)
+        .then(
+          (value) => {
+            result = value;
+          },
+          () => undefined,
+        );
+      // The consumed notice drives one bounded recheck; no further
+      // notification arrives, and the Turn must still settle.
+      await vi.waitFor(
+        () => expect(result).toEqual({ text: "PONG", cancelled: false, agentInvoked: true }),
+        { timeout: 2_000 },
+      );
+      expect(onFault).not.toHaveBeenCalled();
+    } finally {
+      await session.close();
+    }
+  });
+
+  it("ignores a stale settle notice so a freshly admitted Turn is not settled early", async () => {
+    const process = new FakeOmpProcess();
+    const onFault = vi.fn();
+    // Session state is idle from the start (previous Turn fully drained); the
+    // stale notice belongs to that previous Turn.
+    process.handleCommand = (command) => {
+      if (command.type === "prompt") {
+        // Admission only: no assistant output, no agent_end for this Turn yet.
+        process.sendFrame({ type: "response", id: command.id, command: "prompt", success: true });
+        return true;
+      }
+      if (command.type === "get_state") {
+        process.sendFrame({
+          type: "response",
+          id: command.id,
+          command: "get_state",
+          success: true,
+          data: {
+            sessionId: "omp-session",
+            model: { provider: "synthetic", id: "omp-model" },
+            isStreaming: false,
+            isSettled: true,
+            hasPendingAsyncWork: false,
+          },
+        });
+        return true;
+      }
+      return false;
+    };
+    const session = new OmpRpcSession(
+      { cwd: "/synthetic", onFault },
+      { spawn: () => process as never },
+    );
+    try {
+      await session.start();
+      let result: unknown;
+      void session
+        .runTurn("fresh turn", () => undefined)
+        .then(
+          (value) => {
+            result = value;
+          },
+          () => undefined,
+        );
+      // The stale notice from the previous Turn: the fresh Turn has neither a
+      // terminal agent_end nor a local result, so it must NOT be confirmed —
+      // not now, and not after the bounded recheck window.
+      process.sendFrame({ type: "session_settled" });
+      await new Promise((resolve) => setTimeout(resolve, 300));
+      expect(result).toBeUndefined();
+      expect(onFault).not.toHaveBeenCalled();
+      // The Turn's own terminal agent_end is the fact that may settle it.
+      const message = {
+        role: "assistant",
+        responseId: "assistant-1",
+        content: [{ type: "text", text: "PONG" }],
+      };
+      process.sendFrame({ type: "message_start", message });
+      process.sendFrame({ type: "message_end", message: { ...message, stopReason: "stop" } });
+      process.sendFrame({ type: "agent_end", isTerminal: true });
+      await vi.waitFor(
+        () => expect(result).toEqual({ text: "PONG", cancelled: false, agentInvoked: true }),
+        { timeout: 2_000 },
+      );
+    } finally {
+      await session.close();
+    }
+  });
+
   it("waits for background work instead of faulting when agent_end lands while a background job still streams", async () => {
     vi.useFakeTimers();
     const process = new FakeOmpProcess();

From 38a77d0f73a347729bc88a00bf09ce1b30bc031c Mon Sep 17 00:00:00 2001
From: Brave <zxd-666@users.noreply.github.com>
Date: Tue, 3 Nov 2026 13:37:01 +0800
Subject: [PATCH 10/15] fix(host): drain the follow-up queue when a steer
 replacement cannot start
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Upstream review showed the interrupted-terminal drain branch was dead in
practice: at the interrupted Turn's terminal the steer reservation still
holds the Thread busy, and when the replacement then starts and completes,
the pre-existing completed branch drains the queue anyway — so no path
exercised the new condition, and a replacement that fails to start left
the queue paused behind a 继续 no one would press.

Drain when the steer request settles with an error instead: the
reservation is gone and the interrupted terminal is the Thread's last
word, so the queued follow-ups flow. FakeHarnessSession gains
failNextStart to stage a refused replacement start.

Regression test verified to fail without the fix (queue waits forever)
and pass with it.
---
 packages/harness-adapter/src/testing.ts       | 11 ++++
 packages/host-runtime/src/app-server-host.ts  |  5 ++
 .../app-server-host.external-queue.test.ts    | 59 +++++++++++++++++++
 3 files changed, 75 insertions(+)

diff --git a/packages/harness-adapter/src/testing.ts b/packages/harness-adapter/src/testing.ts
index 1e5bb4f82..fc409a702 100644
--- a/packages/harness-adapter/src/testing.ts
+++ b/packages/harness-adapter/src/testing.ts
@@ -162,6 +162,7 @@ export class FakeHarnessSession implements HarnessSession {
   readonly interactionResponses: InteractionRespondCommand[] = [];
   readonly steers: TurnSteerCommand[] = [];
   #nextSteerRefusal: HarnessError | null = null;
+  #nextStartRefusal: HarnessError | null = null;
   readonly outputs: AsyncIterable<HarnessOutput>;
   snapshotReads = 0;
   usageRefreshes = 0;
@@ -413,6 +414,11 @@ export class FakeHarnessSession implements HarnessSession {
       };
     }
     const text = command.input.map((input) => input.text).join("\n");
+    if (this.#nextStartRefusal) {
+      const error = this.#nextStartRefusal;
+      this.#nextStartRefusal = null;
+      return { ok: false, error };
+    }
     if (text.length === 0) {
       return {
         ok: false,
@@ -460,6 +466,11 @@ export class FakeHarnessSession implements HarnessSession {
     this.#nextSteerRefusal = { code: "unsupported", message, retryable: false };
   }
 
+  /** The next turn.start is refused, as when a replacement start is rejected. */
+  failNextStart(error: HarnessError): void {
+    this.#nextStartRefusal = error;
+  }
+
   /** Accepts steered input into the active Turn, as a native insertion primitive would. */
   #steer(command: TurnSteerCommand): HarnessResult<TurnSteerAccepted> {
     if (this.#nextSteerRefusal) {
diff --git a/packages/host-runtime/src/app-server-host.ts b/packages/host-runtime/src/app-server-host.ts
index fbdee2d56..b2c536ca3 100644
--- a/packages/host-runtime/src/app-server-host.ts
+++ b/packages/host-runtime/src/app-server-host.ts
@@ -4985,6 +4985,11 @@ export class AppServerHost {
           errorMessage(error),
         ),
       );
+      // A steer that could not start its replacement leaves the interrupted
+      // terminal as the Thread's last word: the queued follow-ups would wait
+      // for a "继续" that no one will press. The steer reservation is gone,
+      // so the queue can flow again.
+      void this.#drainExternalQueue(thread);
     } finally {
       this.#signalActiveWorkChanged();
     }
diff --git a/packages/host-runtime/test/app-server-host.external-queue.test.ts b/packages/host-runtime/test/app-server-host.external-queue.test.ts
index 72d1ac84a..a92d99a78 100644
--- a/packages/host-runtime/test/app-server-host.external-queue.test.ts
+++ b/packages/host-runtime/test/app-server-host.external-queue.test.ts
@@ -305,6 +305,65 @@ describe("External Thread follow-up queue", () => {
     }
   });
 
+  it("drains the queue when a steer replacement fails to start its replacement Turn", async () => {
+    const fixture = createFixture();
+    // No native steer: turn/steer takes the stop-then-start path.
+    fixture.adapter.supportsSteer = false;
+    try {
+      const threadId = await startPiThread(fixture);
+      const runningTurnId = await startPiTurn(fixture, threadId);
+      const session = fixture.adapter.sessions[0];
+      if (!session) throw new Error("Fake Session was not opened");
+      const execute = vi.spyOn(session, "execute");
+
+      // A follow-up queues while the Turn is running.
+      writeRequest(fixture.desktopInput, {
+        id: 10,
+        method: "thread/queue/add",
+        params: { threadId, input: text("queued during failed steer"), clientUserMessageId: "q-1" },
+      });
+      await fixture.collector.waitFor((message) => requestId(message, 10));
+
+      // The user steers, but the session refuses the replacement start.
+      session.failNextStart({
+        code: "invalidState",
+        message: "Session refused the replacement start",
+        retryable: false,
+      });
+      writeRequest(fixture.desktopInput, {
+        id: 11,
+        method: "turn/steer",
+        params: {
+          threadId,
+          expectedTurnId: runningTurnId,
+          clientUserMessageId: "steer-1",
+          input: text("replacement that will not start"),
+        },
+      });
+      // turn.cancel arrives; the Fake session completes the cancellation on request.
+      session.completeCancellationOnRequest();
+      // The steer request settles as an error (replacement refused).
+      const reply = await fixture.collector.waitFor((message) => requestId(message, 11));
+      expect(reply).toHaveProperty("error");
+      // The interrupted Turn is the Thread's last word and no replacement runs.
+      await fixture.collector.waitFor((message) =>
+        turnEvent(message, "turn/completed", runningTurnId),
+      );
+      // The queued follow-up must still be delivered: a failed steer is not a
+      // stop the queue should pause behind.
+      await vi.waitFor(() =>
+        expect(execute).toHaveBeenCalledWith(
+          expect.objectContaining({
+            type: "turn.start",
+            input: text("queued during failed steer"),
+          }),
+        ),
+      );
+    } finally {
+      await stopFixture(fixture);
+    }
+  });
+
   it("keeps the queue flowing after a steer replacement interrupts the running Turn", async () => {
     const fixture = createFixture();
     // No native steer: turn/steer takes the stop-then-start path.

From 98def0e459bb0ae40ac35552a1b57cbe5fc14fe5 Mon Sep 17 00:00:00 2001
From: BytePioneer-AI <28417416@qq.com>
Date: Tue, 3 Nov 2026 21:57:59 +0800
Subject: [PATCH 11/15] fix(host): preserve paused queues after cancelled
 steering

---
 docs/architecture/external-thread-queue.md    |  1 +
 packages/host-runtime/src/app-server-host.ts  | 35 ++++---
 .../app-server-host.external-queue.test.ts    | 95 +++++++++++++++++++
 3 files changed, 116 insertions(+), 15 deletions(-)

diff --git a/docs/architecture/external-thread-queue.md b/docs/architecture/external-thread-queue.md
index 611dc3e73..72de18e8f 100644
--- a/docs/architecture/external-thread-queue.md
+++ b/docs/architecture/external-thread-queue.md
@@ -19,6 +19,7 @@ Desktop 在 Turn 运行中提交的跟进消息，经官方 app-server 的 `thre
 
 - Turn 正常完成（`completed`）后，Host 自动开始队首提交，对应官方服务端队列的自动推进。若入队请求晚于完成事件到达，Host 在回复入队后补查：仅当前进程已投影的最近一次 Turn 正常完成、Thread 空闲且没有待处理命令时推进；恢复出的历史不作为自动执行依据。
 - Turn 被中断或失败时不自动推进；Desktop 显示“队列已暂停”，用户点击“继续”时发送 `thread/queue/start`。
+- 停止后重发已停止旧轮、替换启动失败且替换仍有效时，Host 在释放替换预留后继续推进队列。用户明确停止、Session fault 或过期/无效的 steer 请求不会解除暂停；等待命令目录期间发生的停止同样有效。
 - 运行中点击“引导”时，Desktop 先发 `turn/steer`，再 `thread/queue/delete`；投递方式由外部 steering 决定，见 `external-thread-steering.md`。
 - 开始失败时，提交放回原位置并再次发出 `thread/queue/changed`。
 - Desktop 只为自己 `turn/start` 的输入做本地乐观展示。由队列开始的 Turn 由 Host 投影带 `clientId` 的 userMessage，否则该轮用户消息不显示。
diff --git a/packages/host-runtime/src/app-server-host.ts b/packages/host-runtime/src/app-server-host.ts
index b2c536ca3..20d8988b6 100644
--- a/packages/host-runtime/src/app-server-host.ts
+++ b/packages/host-runtime/src/app-server-host.ts
@@ -4950,20 +4950,29 @@ export class AppServerHost {
   }
 
   async #steerExternalTurn(request: JsonRpcRequest, thread: ExternalThread): Promise<void> {
+    let replacementStartFailed = false;
     try {
       const started = await this.#externalSteering.run(
         thread,
         requestObject(request),
-        (text, assertActive) =>
-          this.#beginExternalInputTurn(
-            thread,
-            text,
-            assertActive,
-            typeof requestObject(request).clientUserMessageId === "string"
-              ? (requestObject(request).clientUserMessageId as string)
-              : undefined,
-            requestObject(request)[THREAD_REFERENCE_SOURCE_HOST_PARAM],
-          ),
+        async (text, assertActive) => {
+          try {
+            return await this.#beginExternalInputTurn(
+              thread,
+              text,
+              assertActive,
+              typeof requestObject(request).clientUserMessageId === "string"
+                ? (requestObject(request).clientUserMessageId as string)
+                : undefined,
+              requestObject(request)[THREAD_REFERENCE_SOURCE_HOST_PARAM],
+            );
+          } catch (error) {
+            // A stop or Session fault while command discovery was pending keeps the queue paused.
+            assertActive();
+            replacementStartFailed = true;
+            throw error;
+          }
+        },
         (input) => this.#insertExternalInput(thread, input),
       );
       try {
@@ -4985,11 +4994,7 @@ export class AppServerHost {
           errorMessage(error),
         ),
       );
-      // A steer that could not start its replacement leaves the interrupted
-      // terminal as the Thread's last word: the queued follow-ups would wait
-      // for a "继续" that no one will press. The steer reservation is gone,
-      // so the queue can flow again.
-      void this.#drainExternalQueue(thread);
+      if (replacementStartFailed) void this.#drainExternalQueue(thread);
     } finally {
       this.#signalActiveWorkChanged();
     }
diff --git a/packages/host-runtime/test/app-server-host.external-queue.test.ts b/packages/host-runtime/test/app-server-host.external-queue.test.ts
index a92d99a78..504c14d2f 100644
--- a/packages/host-runtime/test/app-server-host.external-queue.test.ts
+++ b/packages/host-runtime/test/app-server-host.external-queue.test.ts
@@ -418,3 +418,98 @@ describe("External Thread follow-up queue", () => {
     }
   });
 });
+
+describe("Steering failure queue boundaries", () => {
+  it("keeps queued input paused when a replacement is stopped during command discovery", async () => {
+    const fixture = createFixture();
+    fixture.adapter.supportsSteer = false;
+    try {
+      const threadId = await startPiThread(fixture);
+      const turnId = await startPiTurn(fixture, threadId);
+      const session = fixture.adapter.sessions[0];
+      if (!session) throw new Error("Fake Session was not opened");
+      const execute = vi.spyOn(session, "execute");
+      writeRequest(fixture.desktopInput, {
+        id: 80,
+        method: "thread/queue/add",
+        params: { threadId, input: text("remain paused"), clientUserMessageId: "paused-1" },
+      });
+      await fixture.collector.waitFor((message) => requestId(message, 80));
+      session.completeCancellationOnRequest();
+      const catalog = Promise.withResolvers<{ ok: true; value: { commands: [] } }>();
+      const list = vi.fn(() => catalog.promise);
+      session.commands = { list, execute: vi.fn() };
+      writeRequest(fixture.desktopInput, {
+        id: 81,
+        method: "turn/steer",
+        params: { threadId, expectedTurnId: turnId, input: text("/unknown") },
+      });
+      await vi.waitFor(() => expect(list).toHaveBeenCalled());
+      writeRequest(fixture.desktopInput, {
+        id: 82,
+        method: "turn/interrupt",
+        params: { threadId, turnId },
+      });
+      await fixture.collector.waitFor((message) => requestId(message, 82));
+      catalog.resolve({ ok: true, value: { commands: [] } });
+      await expect(
+        fixture.collector.waitFor((message) => requestId(message, 81)),
+      ).resolves.toHaveProperty("error");
+      writeRequest(fixture.desktopInput, {
+        id: 83,
+        method: "thread/queue/list",
+        params: { threadId },
+      });
+      await expect(
+        fixture.collector.waitFor((message) => requestId(message, 83)),
+      ).resolves.toMatchObject({ result: { data: [{ input: text("remain paused") }] } });
+      expect(execute).not.toHaveBeenCalledWith(expect.objectContaining({ type: "turn.start" }));
+    } finally {
+      await stopFixture(fixture);
+    }
+  });
+
+  it("does not resume a stopped queue for a stale steer request", async () => {
+    const fixture = createFixture();
+    try {
+      const threadId = await startPiThread(fixture);
+      const turnId = await startPiTurn(fixture, threadId);
+      const session = fixture.adapter.sessions[0];
+      if (!session) throw new Error("Fake Session was not opened");
+      const execute = vi.spyOn(session, "execute");
+      writeRequest(fixture.desktopInput, {
+        id: 90,
+        method: "thread/queue/add",
+        params: { threadId, input: text("remain paused"), clientUserMessageId: "paused-2" },
+      });
+      await fixture.collector.waitFor((message) => requestId(message, 90));
+      session.completeCancellationOnRequest();
+      writeRequest(fixture.desktopInput, {
+        id: 91,
+        method: "turn/interrupt",
+        params: { threadId, turnId },
+      });
+      await fixture.collector.waitFor((message) => requestId(message, 91));
+      await fixture.collector.waitFor((message) => turnEvent(message, "turn/completed", turnId));
+      writeRequest(fixture.desktopInput, {
+        id: 92,
+        method: "turn/steer",
+        params: { threadId, expectedTurnId: turnId, input: text("stale input") },
+      });
+      await expect(
+        fixture.collector.waitFor((message) => requestId(message, 92)),
+      ).resolves.toHaveProperty("error");
+      writeRequest(fixture.desktopInput, {
+        id: 93,
+        method: "thread/queue/list",
+        params: { threadId },
+      });
+      await expect(
+        fixture.collector.waitFor((message) => requestId(message, 93)),
+      ).resolves.toMatchObject({ result: { data: [{ input: text("remain paused") }] } });
+      expect(execute).not.toHaveBeenCalledWith(expect.objectContaining({ type: "turn.start" }));
+    } finally {
+      await stopFixture(fixture);
+    }
+  });
+});

From b450393916e0ac6364c4023aa305d61e0a235c21 Mon Sep 17 00:00:00 2001
From: BytePioneer-AI <28417416@qq.com>
Date: Tue, 3 Nov 2026 21:58:06 +0800
Subject: [PATCH 12/15] fix(adapters): preserve uncertain native steer delivery

---
 docs/architecture/external-thread-steering.md |  3 +-
 packages/adapters/omp/src/omp-adapter.ts      | 18 +++--
 packages/adapters/omp/src/omp-rpc-session.ts  | 20 ++++-
 .../adapters/omp/test/omp-adapter.test.ts     | 73 ++++++++++++++++++-
 .../adapters/omp/test/omp-rpc-session.test.ts | 37 ++++++++++
 packages/adapters/pi/src/pi-adapter.ts        | 18 +++--
 packages/adapters/pi/src/pi-rpc-session.ts    | 20 ++++-
 packages/adapters/pi/test/pi-adapter.test.ts  | 59 +++++++++++++++
 .../adapters/pi/test/pi-rpc-session.test.ts   | 34 ++++++++-
 9 files changed, 267 insertions(+), 15 deletions(-)

diff --git a/docs/architecture/external-thread-steering.md b/docs/architecture/external-thread-steering.md
index ec83b47f5..096ce07b8 100644
--- a/docs/architecture/external-thread-steering.md
+++ b/docs/architecture/external-thread-steering.md
@@ -5,7 +5,7 @@ codexhost 对外部 Harness Thread 保留一个「调整方向」操作，按当
 - **原生同轮插入**：Session 此刻声明 `capabilities.steer` 时，Host 发送公共命令 `turn.steer`，由 Adapter 调用 Harness 自己的插入原语，在原生安全边界把输入送入**当前 Turn**，不取消、不另起 Turn。
 - **停止后重发（退路）**：其余情况下，Host **取消当前 Turn，等待它终结，再自动执行本次新输入**。任何不确定能并入当前 Turn 的情况都走这条路径，保证不比没有原生插入时差。
 
-用户继续使用 Codex Desktop 原有按钮、跟进处理方式设置和单条消息反向操作，不需要手动停止后重发。官方 Codex Thread 仍透传原生 `turn/steer`，不改变同轮追加输入的语义。当前只有 Claude Code 会声明 `capabilities.steer`，并且只在确定插入会并入当前 Turn 时声明；其条件与原生依据见 [`../harnesses/claude-code/claude-code-steer.md`](../harnesses/claude-code/claude-code-steer.md)；其他 Adapter 对 `turn.steer` 返回 `unsupported`，保持停止后重发。
+用户继续使用 Codex Desktop 原有按钮、跟进处理方式设置和单条消息反向操作，不需要手动停止后重发。官方 Codex Thread 仍透传原生 `turn/steer`，不改变同轮追加输入的语义。Claude Code、OMP 和 Pi 会声明 `capabilities.steer`：Claude Code 的条件与原生依据见 [`../harnesses/claude-code/claude-code-steer.md`](../harnesses/claude-code/claude-code-steer.md)；OMP/Pi 使用原生 `steer` RPC，已请求取消的 Turn 不接受原生插入。其他 Adapter 对 `turn.steer` 返回 `unsupported`，保持停止后重发。
 
 ## 原生同轮插入
 
@@ -13,6 +13,7 @@ codexhost 对外部 Harness Thread 保留一个「调整方向」操作，按当
 - Harness Broker 通过 `session.readCapabilities` 按需读取原生 Session 的能力，不轮询、不要求 Adapter 发布能力变更事件。打开响应声明是否支持这项查询；连接旧 Broker 时保持停止后重发，不向其发送未知方法。普通 Adapter 不必实现 `readCapabilities()`，Broker 可以直接读取它的 `capabilities`。
 - 能力查询不是对执行时机的预留。Adapter 在 `turn.steer` 最终写入之前必须再次检查目标 Turn 和原生插入条件；不满足时不写入并拒绝。GUI 已选择同轮展示时，拒绝只报错，不偷偷改走新轮。
 - `TurnSteerCommand { turnId, input }` 定义在 `packages/harness-adapter/src/text-session.ts`。Adapter 在 Harness 确认接受或入队后返回 `{ accepted: true }`，不等待插入实际执行完毕。返回失败即表示 Harness **没有收下**这条输入。已经写入但丢失回执不能作为未收下处理；Broker 对无法确认结果的请求抛错，不转换成触发停止后重发的拒绝结果。没有原生插入的 Adapter 用 `unsupportedSteer()` 拒绝。
+- OMP/Pi 仅把匹配 `steer` 命令且明确返回 `success: false` 的原生响应视为拒绝。回执超时、连接错误或格式不完整的失败响应直接抛错，Host 不据此自动取消、重发或推进队列。
 - `ExternalTurnSteering`（`packages/host-runtime/src/external-turn-steering.ts`）对两种送达方式统一校验非空文本、`expectedTurnId` 和 `threadId + clientUserMessageId` 去重，并按以下顺序决定送达方式：
   1. `turn/steer` 带有 `codexhostSteerDelivery`（Renderer 已经按该方式展示）时照它执行：`newTurn` 停止后重发；`activeTurn` 原生插入，Harness 没收下时报错、不改走退路，避免和 Desktop 已有的展示不一致。
   2. 输入是 Harness 命令（Session 提供命令、文本以 `/命令` 开头）时停止后重发，命令照常解析和执行。
diff --git a/packages/adapters/omp/src/omp-adapter.ts b/packages/adapters/omp/src/omp-adapter.ts
index 55b51be50..9100954f2 100644
--- a/packages/adapters/omp/src/omp-adapter.ts
+++ b/packages/adapters/omp/src/omp-adapter.ts
@@ -98,6 +98,7 @@ import { OmpSessionImport } from "./session-import.js";
 import { rollbackOmpLastTurn } from "./omp-last-turn-rollback.js";
 import {
   OmpRpcFaultError,
+  OmpRpcSteerRejectedError,
   OmpRpcSession,
   OmpRpcUnsupportedCommandError,
   type OmpInteractionRequest,
@@ -160,7 +161,7 @@ export interface OmpTurnTransport {
     onEvent: (event: OmpTurnEvent) => void,
   ): Promise<OmpCompactResult>;
   runTurn(text: string, onEvent: (event: OmpTurnEvent) => void): Promise<OmpTurnResult>;
-  /** Native same-Turn input; present on transports backed by OMP's `steer` RPC. */
+  /** Native same-Turn input; unconfirmed delivery errors must propagate to the caller. */
   steer?(text: string): Promise<void>;
   respondToInteraction(response: OmpInteractionResponse): Promise<void>;
   abort(): Promise<void>;
@@ -1419,10 +1420,17 @@ class OmpHarnessSession implements HarnessSession {
     try {
       await transport.steer(text);
     } catch (error) {
-      return {
-        ok: false,
-        error: invalidState(`Omp native steer failed: ${errorMessage(error)}`),
-      };
+      if (
+        error instanceof OmpRpcSteerRejectedError ||
+        error instanceof OmpRpcUnsupportedCommandError
+      ) {
+        return {
+          ok: false,
+          error: invalidState(`Omp native steer failed: ${errorMessage(error)}`),
+        };
+      }
+      // An unconfirmed write may already have delivered input; it cannot authorize a resend.
+      throw error;
     }
     return { ok: true, value: { accepted: true } };
   }
diff --git a/packages/adapters/omp/src/omp-rpc-session.ts b/packages/adapters/omp/src/omp-rpc-session.ts
index f5df74c7b..dc4d45ce5 100644
--- a/packages/adapters/omp/src/omp-rpc-session.ts
+++ b/packages/adapters/omp/src/omp-rpc-session.ts
@@ -171,6 +171,14 @@ export class OmpRpcUnsupportedCommandError extends Error {
   }
 }
 
+/** The native RPC explicitly refused steering without accepting its input. */
+export class OmpRpcSteerRejectedError extends Error {
+  constructor(message: string) {
+    super(message);
+    this.name = "OmpRpcSteerRejectedError";
+  }
+}
+
 export interface OmpRpcSessionOptions {
   cwd: string;
   command?: string;
@@ -1651,8 +1659,18 @@ export class OmpRpcSession {
       pending.resolve(value);
     } else {
       const error = typeof value.error === "string" ? value.error : "Omp RPC command failed";
-      if (value.command === pending.command && error === `Unknown command: ${pending.command}`) {
+      if (
+        value.success === false &&
+        value.command === pending.command &&
+        error === `Unknown command: ${pending.command}`
+      ) {
         pending.reject(new OmpRpcUnsupportedCommandError(pending.command));
+      } else if (
+        value.success === false &&
+        pending.command === "steer" &&
+        value.command === pending.command
+      ) {
+        pending.reject(new OmpRpcSteerRejectedError(error));
       } else {
         pending.reject(new Error(error));
       }
diff --git a/packages/adapters/omp/test/omp-adapter.test.ts b/packages/adapters/omp/test/omp-adapter.test.ts
index d0a58eeb5..99b54b1a5 100644
--- a/packages/adapters/omp/test/omp-adapter.test.ts
+++ b/packages/adapters/omp/test/omp-adapter.test.ts
@@ -29,7 +29,11 @@ import type {
   OmpTurnEvent,
   OmpTurnResult,
 } from "../src/omp-rpc-session.js";
-import { OmpRpcFaultError } from "../src/omp-rpc-session.js";
+import {
+  OmpRpcFaultError,
+  OmpRpcSteerRejectedError,
+  OmpRpcUnsupportedCommandError,
+} from "../src/omp-rpc-session.js";
 import { encodeOmpModelRef, type OmpNativeModel } from "../src/omp-model-catalog.js";
 
 class FakeOmpTransport implements OmpTurnTransport {
@@ -2000,3 +2004,70 @@ describe("OMP Adapter Subagents", () => {
     await adapter.close();
   });
 });
+
+describe("OMP native steer delivery", () => {
+  it("propagates an unconfirmed delivery instead of reporting that input was refused", async () => {
+    const transport = new FakeOmpTransport();
+    transport.autoCompleteTurn = false;
+    const timeout = new Error("Omp RPC 'steer' command timed out");
+    const steer = vi.fn<NonNullable<OmpTurnTransport["steer"]>>(async () => {
+      throw timeout;
+    });
+    const native: OmpTurnTransport = Object.assign(transport, { steer });
+    const abort = vi.spyOn(transport, "abort");
+    const adapter = new OmpAdapter({}, { createTransport: () => native });
+    try {
+      const opened = await adapter.open({ kind: "create", cwd: "/synthetic" });
+      if (!opened.ok) throw new Error(opened.error.message);
+      const session = opened.value;
+      const turnId = "steer-timeout" as HostTurnId;
+      await session.execute({
+        type: "turn.start",
+        turnId,
+        input: [{ type: "text", text: "first" }],
+      });
+      await expect(
+        session.execute({
+          type: "turn.steer",
+          turnId,
+          input: [{ type: "text", text: "already written" }],
+        }),
+      ).rejects.toBe(timeout);
+      expect(steer).toHaveBeenCalledExactlyOnceWith("already written");
+      expect(abort).not.toHaveBeenCalled();
+    } finally {
+      await adapter.close();
+    }
+  });
+});
+
+it.each([
+  { label: "accepted", error: null },
+  { label: "refused", error: new OmpRpcSteerRejectedError("Native steer refused") },
+  { label: "unsupported", error: new OmpRpcUnsupportedCommandError("steer") },
+])("classifies native Omp steer $label responses", async ({ error }) => {
+  const transport = new FakeOmpTransport();
+  transport.autoCompleteTurn = false;
+  const steer = vi.fn<NonNullable<OmpTurnTransport["steer"]>>(async () => {
+    if (error) throw error;
+  });
+  const native: OmpTurnTransport = Object.assign(transport, { steer });
+  const adapter = new OmpAdapter({}, { createTransport: () => native });
+  const opened = await adapter.open({ kind: "create", cwd: "/synthetic" });
+  if (!opened.ok) throw new Error(opened.error.message);
+  const session = opened.value;
+  const turnId = "native-steer" as HostTurnId;
+  await session.execute({ type: "turn.start", turnId, input: [{ type: "text", text: "first" }] });
+  try {
+    const result = await session.execute({
+      type: "turn.steer",
+      turnId,
+      input: [{ type: "text", text: "second" }],
+    });
+    if (error) expect(result).toMatchObject({ ok: false, error: { code: "invalidState" } });
+    else expect(result).toEqual({ ok: true, value: { accepted: true } });
+    expect(steer).toHaveBeenCalledExactlyOnceWith("second");
+  } finally {
+    await adapter.close();
+  }
+});
diff --git a/packages/adapters/omp/test/omp-rpc-session.test.ts b/packages/adapters/omp/test/omp-rpc-session.test.ts
index a13918c65..478a57b96 100644
--- a/packages/adapters/omp/test/omp-rpc-session.test.ts
+++ b/packages/adapters/omp/test/omp-rpc-session.test.ts
@@ -8,6 +8,7 @@ import { describe, expect, it, vi } from "vitest";
 
 import {
   OmpRpcSession,
+  OmpRpcSteerRejectedError,
   ompRpcProcessCommand,
   type OmpRpcProcessAdapter,
   type OmpTurnEvent,
@@ -2474,3 +2475,39 @@ describe("OMP RPC session", () => {
     await session.close();
   });
 });
+
+it.each([
+  { label: "explicit native refusal", response: { success: false, error: "Busy" }, refused: true },
+  { label: "lost acknowledgement", response: null, refused: false },
+  {
+    label: "mismatched command",
+    response: { command: "prompt", success: false, error: "Busy" },
+    refused: false,
+  },
+  { label: "missing failure status", response: { error: "Busy" }, refused: false },
+])("Omp steer distinguishes $label from confirmed non-delivery", async ({ response, refused }) => {
+  const fake = new FakeOmpProcess();
+  fake.promptHandler = (command) =>
+    fake.sendFrame({ type: "response", id: command.id, command: "prompt", success: true });
+  fake.handleCommand = (command) => {
+    if (command.type !== "steer") return false;
+    if (response)
+      fake.sendFrame({ type: "response", id: command.id, command: "steer", ...response });
+    return true;
+  };
+  const session = new OmpRpcSession(
+    { cwd: "/synthetic", commandTimeoutMs: 20 },
+    { spawn: () => fake as never },
+  );
+  try {
+    await session.start();
+    void session.runTurn("first", () => undefined).catch(() => undefined);
+    const error = await session.steer("second").catch((failure: unknown) => failure);
+    expect(error).toBeInstanceOf(Error);
+    if (refused) expect(error).toBeInstanceOf(OmpRpcSteerRejectedError);
+    else expect(error).not.toBeInstanceOf(OmpRpcSteerRejectedError);
+    expect(fake.commands.filter((command) => command.type === "steer")).toHaveLength(1);
+  } finally {
+    await session.close();
+  }
+});
diff --git a/packages/adapters/pi/src/pi-adapter.ts b/packages/adapters/pi/src/pi-adapter.ts
index a85b2ff81..fba57d980 100644
--- a/packages/adapters/pi/src/pi-adapter.ts
+++ b/packages/adapters/pi/src/pi-adapter.ts
@@ -105,6 +105,7 @@ import { PiSessionImportIndex } from "./pi-session-import.js";
 import { SessionImportScope } from "@codexhost/harness-adapter/session-import";
 import {
   PiRpcFaultError,
+  PiRpcSteerRejectedError,
   PiRpcBusyError,
   PiRpcSession,
   PiRpcUnsupportedCommandError,
@@ -175,7 +176,7 @@ export interface PiTurnTransport {
     onEvent: (event: PiTurnEvent) => void,
   ): Promise<PiCompactResult>;
   runTurn(text: string, onEvent: (event: PiTurnEvent) => void): Promise<PiTurnResult>;
-  /** Native same-Turn input; present on transports backed by Pi's `steer` RPC. */
+  /** Native same-Turn input; unconfirmed delivery errors must propagate to the caller. */
   steer?(text: string): Promise<void>;
   respondToInteraction(response: PiInteractionResponse): Promise<void>;
   abort(): Promise<void>;
@@ -994,10 +995,17 @@ class PiHarnessSession implements HarnessSession {
     try {
       await transport.steer(text);
     } catch (error) {
-      return {
-        ok: false,
-        error: invalidState(`Pi native steer failed: ${errorMessage(error)}`),
-      };
+      if (
+        error instanceof PiRpcSteerRejectedError ||
+        error instanceof PiRpcUnsupportedCommandError
+      ) {
+        return {
+          ok: false,
+          error: invalidState(`Pi native steer failed: ${errorMessage(error)}`),
+        };
+      }
+      // An unconfirmed write may already have delivered input; it cannot authorize a resend.
+      throw error;
     }
     return { ok: true, value: { accepted: true } };
   }
diff --git a/packages/adapters/pi/src/pi-rpc-session.ts b/packages/adapters/pi/src/pi-rpc-session.ts
index b58cc9e58..f2eb5e95c 100644
--- a/packages/adapters/pi/src/pi-rpc-session.ts
+++ b/packages/adapters/pi/src/pi-rpc-session.ts
@@ -195,6 +195,14 @@ export class PiRpcUnsupportedCommandError extends Error {
   }
 }
 
+/** The native RPC explicitly refused steering without accepting its input. */
+export class PiRpcSteerRejectedError extends Error {
+  constructor(message: string) {
+    super(message);
+    this.name = "PiRpcSteerRejectedError";
+  }
+}
+
 export interface PiRpcSessionOptions {
   cwd: string;
   command?: string;
@@ -1713,8 +1721,18 @@ export class PiRpcSession {
     if (value.success === true) pending.resolve(value);
     else {
       const error = typeof value.error === "string" ? value.error : "Pi RPC command failed";
-      if (value.command === pending.command && error === `Unknown command: ${pending.command}`) {
+      if (
+        value.success === false &&
+        value.command === pending.command &&
+        error === `Unknown command: ${pending.command}`
+      ) {
         pending.reject(new PiRpcUnsupportedCommandError(pending.command));
+      } else if (
+        value.success === false &&
+        pending.command === "steer" &&
+        value.command === pending.command
+      ) {
+        pending.reject(new PiRpcSteerRejectedError(error));
       } else {
         pending.reject(new Error(error));
       }
diff --git a/packages/adapters/pi/test/pi-adapter.test.ts b/packages/adapters/pi/test/pi-adapter.test.ts
index 5aa2fc915..3d537dd00 100644
--- a/packages/adapters/pi/test/pi-adapter.test.ts
+++ b/packages/adapters/pi/test/pi-adapter.test.ts
@@ -44,6 +44,8 @@ import { encodePiModelRef } from "../src/pi-model-catalog.js";
 import type { PiNativeCommand } from "../src/pi-slash-commands.js";
 import {
   PiRpcFaultError,
+  PiRpcSteerRejectedError,
+  PiRpcUnsupportedCommandError,
   type PiAutonomousTurn,
   type PiCompactResult,
   type PiInteractionResponse,
@@ -4190,3 +4192,60 @@ describe("Pi HarnessAdapter Session", () => {
     ]);
   });
 });
+
+describe("Pi native steer delivery", () => {
+  it("propagates an unconfirmed delivery instead of reporting that input was refused", async () => {
+    const f = fixture();
+    const session = await openSession(f.adapter);
+    try {
+      await session.execute(textTurn("steer-timeout"));
+      const transport = f.transports[0];
+      if (!transport) throw new Error("Fake transport was not opened");
+      const timeout = new Error("Pi RPC 'steer' command timed out");
+      const steer = vi.fn<NonNullable<PiTurnTransport["steer"]>>(async () => {
+        throw timeout;
+      });
+      Object.assign(transport, { steer });
+      await expect(
+        session.execute({
+          type: "turn.steer",
+          turnId: hostTurnIdSchema.parse("steer-timeout"),
+          input: [{ type: "text", text: "already written" }],
+        }),
+      ).rejects.toBe(timeout);
+      expect(steer).toHaveBeenCalledExactlyOnceWith("already written");
+      expect(transport.abort).not.toHaveBeenCalled();
+    } finally {
+      await f.adapter.close();
+    }
+  });
+});
+
+it.each([
+  { label: "accepted", error: null },
+  { label: "refused", error: new PiRpcSteerRejectedError("Native steer refused") },
+  { label: "unsupported", error: new PiRpcUnsupportedCommandError("steer") },
+])("classifies native Pi steer $label responses", async ({ error }) => {
+  const f = fixture();
+  const session = await openSession(f.adapter);
+  const turnId = hostTurnIdSchema.parse("native-steer");
+  await session.execute(textTurn("native-steer"));
+  const transport = f.transports[0];
+  if (!transport) throw new Error("Fake transport was not opened");
+  const steer = vi.fn<NonNullable<PiTurnTransport["steer"]>>(async () => {
+    if (error) throw error;
+  });
+  Object.assign(transport, { steer });
+  try {
+    const result = await session.execute({
+      type: "turn.steer",
+      turnId,
+      input: [{ type: "text", text: "second" }],
+    });
+    if (error) expect(result).toMatchObject({ ok: false, error: { code: "invalidState" } });
+    else expect(result).toEqual({ ok: true, value: { accepted: true } });
+    expect(steer).toHaveBeenCalledExactlyOnceWith("second");
+  } finally {
+    await f.adapter.close();
+  }
+});
diff --git a/packages/adapters/pi/test/pi-rpc-session.test.ts b/packages/adapters/pi/test/pi-rpc-session.test.ts
index 08169c798..e6203d7ea 100644
--- a/packages/adapters/pi/test/pi-rpc-session.test.ts
+++ b/packages/adapters/pi/test/pi-rpc-session.test.ts
@@ -14,6 +14,7 @@ import { CodexTurnProjector } from "@codexhost/protocol-core";
 
 import {
   PiRpcSession,
+  PiRpcSteerRejectedError,
   piRpcProcessCommand,
   type PiAutonomousTurn,
   type PiRpcProcessAdapter,
@@ -81,6 +82,7 @@ class FakePiRpcProcess extends EventEmitter {
   fastEnabled = false;
   readonly commands: string[] = [];
   readonly commandFrames: Record<string, unknown>[] = [];
+  steerReply: Record<string, unknown> | null = { success: true, data: { disposition: "queued" } };
   holdEntries = false;
   holdStates = false;
   holdAborts = false;
@@ -282,7 +284,8 @@ class FakePiRpcProcess extends EventEmitter {
       return;
     }
     if (command.type === "steer") {
-      this.#respond(command, { disposition: "queued" });
+      if (this.steerReply)
+        this.#output({ type: "response", id: command.id, command: "steer", ...this.steerReply });
       return;
     }
     if (command.type === "get_state" && this.holdStates) {
@@ -3480,3 +3483,32 @@ describe("Pi RPC Turn aggregation", () => {
     await rpc.close();
   });
 });
+
+it.each([
+  { label: "explicit native refusal", response: { success: false, error: "Busy" }, refused: true },
+  { label: "lost acknowledgement", response: null, refused: false },
+  {
+    label: "mismatched command",
+    response: { command: "prompt", success: false, error: "Busy" },
+    refused: false,
+  },
+  { label: "missing failure status", response: { error: "Busy" }, refused: false },
+])("Pi steer distinguishes $label from confirmed non-delivery", async ({ response, refused }) => {
+  const fake = new FakePiRpcProcess("manual-run");
+  fake.steerReply = response;
+  const session = new PiRpcSession(
+    { cwd: process.cwd(), commandTimeoutMs: 20 },
+    { spawn: () => fake as unknown as ChildProcessWithoutNullStreams },
+  );
+  try {
+    await session.start();
+    void session.runTurn("first", () => undefined).catch(() => undefined);
+    const error = await session.steer("second").catch((failure: unknown) => failure);
+    expect(error).toBeInstanceOf(Error);
+    if (refused) expect(error).toBeInstanceOf(PiRpcSteerRejectedError);
+    else expect(error).not.toBeInstanceOf(PiRpcSteerRejectedError);
+    expect(fake.commandFrames.filter((command) => command.type === "steer")).toHaveLength(1);
+  } finally {
+    await session.close();
+  }
+});

From 789e95647300a09f78047a2c3c3307ee5b70d111 Mon Sep 17 00:00:00 2001
From: BytePioneer-AI <28417416@qq.com>
Date: Tue, 3 Nov 2026 23:15:52 +0800
Subject: [PATCH 13/15] fix(adapters): settle steered turns by their starting
 entry

---
 docs/architecture/external-thread-steering.md |   5 +-
 packages/adapters/omp/src/omp-adapter.ts      |  30 ++--
 .../adapters/omp/test/omp-adapter.test.ts     | 137 +++++++++++++++-
 packages/adapters/pi/src/pi-adapter.ts        |  24 +--
 packages/adapters/pi/test/pi-adapter.test.ts  | 147 ++++++++++++++++++
 5 files changed, 316 insertions(+), 27 deletions(-)

diff --git a/docs/architecture/external-thread-steering.md b/docs/architecture/external-thread-steering.md
index 096ce07b8..c3a17cfc0 100644
--- a/docs/architecture/external-thread-steering.md
+++ b/docs/architecture/external-thread-steering.md
@@ -3,7 +3,7 @@
 codexhost 对外部 Harness Thread 保留一个「调整方向」操作，按当前 Session 的能力选择送达方式：
 
 - **原生同轮插入**：Session 此刻声明 `capabilities.steer` 时，Host 发送公共命令 `turn.steer`，由 Adapter 调用 Harness 自己的插入原语，在原生安全边界把输入送入**当前 Turn**，不取消、不另起 Turn。
-- **停止后重发（退路）**：其余情况下，Host **取消当前 Turn，等待它终结，再自动执行本次新输入**。任何不确定能并入当前 Turn 的情况都走这条路径，保证不比没有原生插入时差。
+- **停止后重发（退路）**：其余情况下，Host **取消当前 Turn，等待它终结，再自动执行本次新输入**。不支持原生插入，或原生明确拒绝且输入未送达时走这条路径；写入后回执不确定时不自动重发。
 
 用户继续使用 Codex Desktop 原有按钮、跟进处理方式设置和单条消息反向操作，不需要手动停止后重发。官方 Codex Thread 仍透传原生 `turn/steer`，不改变同轮追加输入的语义。Claude Code、OMP 和 Pi 会声明 `capabilities.steer`：Claude Code 的条件与原生依据见 [`../harnesses/claude-code/claude-code-steer.md`](../harnesses/claude-code/claude-code-steer.md)；OMP/Pi 使用原生 `steer` RPC，已请求取消的 Turn 不接受原生插入。其他 Adapter 对 `turn.steer` 返回 `unsupported`，保持停止后重发。
 
@@ -13,6 +13,7 @@ codexhost 对外部 Harness Thread 保留一个「调整方向」操作，按当
 - Harness Broker 通过 `session.readCapabilities` 按需读取原生 Session 的能力，不轮询、不要求 Adapter 发布能力变更事件。打开响应声明是否支持这项查询；连接旧 Broker 时保持停止后重发，不向其发送未知方法。普通 Adapter 不必实现 `readCapabilities()`，Broker 可以直接读取它的 `capabilities`。
 - 能力查询不是对执行时机的预留。Adapter 在 `turn.steer` 最终写入之前必须再次检查目标 Turn 和原生插入条件；不满足时不写入并拒绝。GUI 已选择同轮展示时，拒绝只报错，不偷偷改走新轮。
 - `TurnSteerCommand { turnId, input }` 定义在 `packages/harness-adapter/src/text-session.ts`。Adapter 在 Harness 确认接受或入队后返回 `{ accepted: true }`，不等待插入实际执行完毕。返回失败即表示 Harness **没有收下**这条输入。已经写入但丢失回执不能作为未收下处理；Broker 对无法确认结果的请求抛错，不转换成触发停止后重发的拒绝结果。没有原生插入的 Adapter 用 `unsupportedSteer()` 拒绝。
+- OMP/Pi 的实时任务以启动前历史为基线，使用有效分支中基线之后的第一条用户记录作为稳定身份；原生 `prompt` 先于后续 `steer` 输入，该顺序由原生历史保留。追加产生的用户记录不改变这次运行的身份，也不要求用户记录总数为一；缺少新的用户记录仍报错。原生执行结束、读取完成历史期间不再接收追加。重读快照继续保留每条原生用户记录及其历史分组，不把多条输入合并或丢弃。
 - OMP/Pi 仅把匹配 `steer` 命令且明确返回 `success: false` 的原生响应视为拒绝。回执超时、连接错误或格式不完整的失败响应直接抛错，Host 不据此自动取消、重发或推进队列。
 - `ExternalTurnSteering`（`packages/host-runtime/src/external-turn-steering.ts`）对两种送达方式统一校验非空文本、`expectedTurnId` 和 `threadId + clientUserMessageId` 去重，并按以下顺序决定送达方式：
   1. `turn/steer` 带有 `codexhostSteerDelivery`（Renderer 已经按该方式展示）时照它执行：`newTurn` 停止后重发；`activeTurn` 原生插入，Harness 没收下时报错、不改走退路，避免和 Desktop 已有的展示不一致。
@@ -32,7 +33,7 @@ codexhost 对外部 Harness Thread 保留一个「调整方向」操作，按当
 原生历史可能已经在插入点分轮，而 Desktop 在重新读取前仍把内容显示为一个 Turn。Host Runtime 按 Host Thread ID 在内存中拥有 `steeredTurnIds`，记录插过队的 Host Turn：
 
 - Fork 先解析请求的实际 Checkpoint 边界，再检查该边界的 Host Turn；省略 `lastTurnId` 或使用 `beforeTurnId` 不会绕过保护。
-- 对 steered 边界的 Fork，以及撤销范围内包含 steered Turn 的撤销，返回 `-32080` 并要求重新打开 Thread。原生历史只会在 steered Turn 内多分出一轮，它之后的 Turn 仍与原生 Turn 一一对应，所以只撤销之后的 Turn 不受限制。
+- 对 steered 边界的 Fork，以及撤销范围内包含 steered Turn 的撤销，返回 `-32080` 并要求重新打开 Thread。原生历史可能在 steered Turn 内按多条输入分轮，它之后的 Turn 仍与原生 Turn 一一对应，所以只撤销之后的 Turn 不受限制。
 - 空闲释放只卸载 Native Session，不等于 Desktop 已读取历史；Session 恢复后继续使用同一组未读标记。
 - Desktop 重新读取历史并成功收到 Turn 边界后，只清除实际返回的已结束 Turn 的标记。只读取 Items 或未包含该 Turn 的分页不会解除保护。
 - 标记不落盘；Host 进程重启后自然消失。
diff --git a/packages/adapters/omp/src/omp-adapter.ts b/packages/adapters/omp/src/omp-adapter.ts
index 9100954f2..39ac12176 100644
--- a/packages/adapters/omp/src/omp-adapter.ts
+++ b/packages/adapters/omp/src/omp-adapter.ts
@@ -221,6 +221,7 @@ interface ActiveTurn {
   interactionByNativeId: Map<string, HostInteractionId>;
   subagents: OmpSubagentLifecycle;
   cancellationRequested: boolean;
+  settling: boolean;
   beforeNativeTurnKeys: Set<string>;
   completion: Promise<void>;
   resolveCompletion(): void;
@@ -730,12 +731,13 @@ class OmpHarnessSession implements HarnessSession {
       },
       history: { fork: true, forkAcrossCwd: true, rollbackLastTurn: true },
       subagents: { observe: true, readTranscript: true },
-      // Read live: OMP's `steer` RPC delivers mid-Turn input into the same
-      // native run. A Turn whose cancellation is in flight must not take
-      // more input; every other case stops and restarts.
+      // Native input is unavailable during cancellation or completion settlement.
       get steer(): boolean {
         return (
-          session.#transport?.steer !== undefined && session.#active?.cancellationRequested !== true
+          session.#transport?.steer !== undefined &&
+          session.#active !== null &&
+          !session.#active.cancellationRequested &&
+          !session.#active.settling
         );
       },
     }))(this);
@@ -992,6 +994,7 @@ class OmpHarnessSession implements HarnessSession {
           emit: (event) => this.#event(event),
         }),
         cancellationRequested: false,
+        settling: false,
         beforeNativeTurnKeys: new Set(
           beforeHistory.turns.map((turn) => turn.nativeTurnRef.nativeTurnKey),
         ),
@@ -1005,6 +1008,7 @@ class OmpHarnessSession implements HarnessSession {
       void transport
         .runTurn(text, (event) => this.#handleTurnEvent(active, event))
         .then(async (result) => {
+          active.settling = true;
           if (!result.agentInvoked) {
             this.#completeTurn(
               active,
@@ -1041,6 +1045,7 @@ class OmpHarnessSession implements HarnessSession {
           }
         })
         .catch(async (error: unknown) => {
+          active.settling = true;
           let identity:
             { nativeTurnRef: NativeTurnRef; checkpoint: NativeCheckpointRef } | undefined;
           try {
@@ -1404,6 +1409,9 @@ class OmpHarnessSession implements HarnessSession {
         error: invalidState("Omp steer target is being cancelled"),
       };
     }
+    if (active.settling) {
+      return { ok: false, error: invalidState("Omp steer target is recording its completion") };
+    }
     const text = command.input.map((input) => input.text).join("\n");
     if (text.length === 0) {
       return {
@@ -1568,6 +1576,7 @@ class OmpHarnessSession implements HarnessSession {
           emit: (event) => this.#event(event),
         }),
         cancellationRequested: false,
+        settling: false,
         beforeNativeTurnKeys: new Set(),
         completion,
         resolveCompletion,
@@ -2139,16 +2148,13 @@ class OmpHarnessSession implements HarnessSession {
       sessionId: transport.state.sessionId,
       model: nativeModelForHistory(transport.state),
     });
-    const created = snapshot.turns.filter(
+    // The active branch is ordered: the first User Entry after startup is the prompt
+    // that started this run. Native steers append more Users without changing that identity.
+    const turn = snapshot.turns.find(
       (turn) => !active.beforeNativeTurnKeys.has(turn.nativeTurnRef.nativeTurnKey),
     );
-    if (created.length !== 1) {
-      throw new Error(
-        `Omp Turn persisted ${created.length} new User Entries; exactly one is required`,
-      );
-    }
-    const turn = created[0];
-    if (!turn?.checkpoint) throw new Error("Omp Turn has no terminal Checkpoint identity");
+    if (!turn) throw new Error("Omp Turn has no newly persisted User Entry");
+    if (!turn.checkpoint) throw new Error("Omp Turn has no terminal Checkpoint identity");
     return { nativeTurnRef: turn.nativeTurnRef, checkpoint: turn.checkpoint };
   }
 
diff --git a/packages/adapters/omp/test/omp-adapter.test.ts b/packages/adapters/omp/test/omp-adapter.test.ts
index 99b54b1a5..7d370ed0f 100644
--- a/packages/adapters/omp/test/omp-adapter.test.ts
+++ b/packages/adapters/omp/test/omp-adapter.test.ts
@@ -1320,7 +1320,7 @@ describe("OMP Adapter local command completion", () => {
         status: "failed",
         error: {
           code: "protocolError",
-          message: "Omp Turn persisted 0 new User Entries; exactly one is required",
+          message: "Omp Turn has no newly persisted User Entry",
           retryable: false,
         },
       },
@@ -2006,6 +2006,141 @@ describe("OMP Adapter Subagents", () => {
 });
 
 describe("OMP native steer delivery", () => {
+  it.each([false, true])(
+    "keeps the starting identity after two native steers (cancelled=%s)",
+    async (cancelled) => {
+      const transport = new FakeOmpTransport();
+      transport.autoCompleteTurn = false;
+      const steer = vi.fn(async () => undefined);
+      const adapter = new OmpAdapter(
+        {},
+        { createTransport: () => Object.assign(transport, { steer }) },
+      );
+      try {
+        const opened = await adapter.open({ kind: "create", cwd: "/synthetic" });
+        if (!opened.ok) throw new Error(opened.error.message);
+        const session = opened.value;
+        const iterator = session.outputs[Symbol.asyncIterator]();
+        await session.execute({
+          type: "turn.start",
+          turnId: "previous" as HostTurnId,
+          input: [{ type: "text", text: "previous" }],
+        });
+        transport.succeed("previous answer");
+        await nextTurnEvents(iterator);
+        await session.execute({
+          type: "turn.start",
+          turnId: "steered" as HostTurnId,
+          input: [{ type: "text", text: "first" }],
+        });
+        for (const [index, text] of ["first", "second", "third"].entries()) {
+          if (index > 0)
+            expect(
+              await session.execute({
+                type: "turn.steer",
+                turnId: "steered" as HostTurnId,
+                input: [{ type: "text", text }],
+              }),
+            ).toEqual({ ok: true, value: { accepted: true } });
+          const id = `current-user-${index}`;
+          transport.history.entries.push({
+            id,
+            parentId: transport.history.leafId,
+            type: "message",
+            message: {
+              role: "user",
+              content: [{ type: "text", text }],
+              ...(index > 0 ? { steering: true } : {}),
+            },
+          });
+          transport.history.leafId = id;
+        }
+        transport.history.entries.push({
+          id: "current-answer",
+          parentId: transport.history.leafId,
+          type: "message",
+          message: {
+            role: "assistant",
+            content: [{ type: "text", text: "done" }],
+            stopReason: cancelled ? "aborted" : "stop",
+          },
+        });
+        transport.history.leafId = "current-answer";
+        transport.completeWithoutHistory({ agentInvoked: true, text: "done", cancelled });
+        expect((await nextTurnEvents(iterator)).at(-1)).toMatchObject({
+          type: "turn.completed",
+          turnId: "steered",
+          nativeTurnRef: { nativeTurnKey: "current-user-0" },
+          outcome: {
+            status: cancelled ? "cancelled" : "succeeded",
+            checkpoint: { checkpointId: "current-user-0" },
+          },
+        });
+        expect(steer).toHaveBeenCalledTimes(2);
+        const snapshot = await session.readSnapshot();
+        expect(snapshot).toMatchObject({
+          ok: true,
+          value: {
+            turns: [
+              expect.anything(),
+              ...["first", "second", "third"].map((text, index) => ({
+                nativeTurnRef: { nativeTurnKey: `current-user-${index}` },
+                input: [{ type: "text", text }],
+              })),
+            ],
+          },
+        });
+        expect(await session.readSnapshot()).toEqual(snapshot);
+      } finally {
+        await adapter.close();
+      }
+    },
+  );
+
+  it("rejects steering while native completion history is being read", async () => {
+    const transport = new FakeOmpTransport();
+    transport.autoCompleteTurn = false;
+    const steer = vi.fn(async () => undefined);
+    const adapter = new OmpAdapter(
+      {},
+      { createTransport: () => Object.assign(transport, { steer }) },
+    );
+    const gate = Promise.withResolvers<undefined>();
+    try {
+      const opened = await adapter.open({ kind: "create", cwd: "/synthetic" });
+      if (!opened.ok) throw new Error(opened.error.message);
+      const session = opened.value;
+      const iterator = session.outputs[Symbol.asyncIterator]();
+      await session.execute({
+        type: "turn.start",
+        turnId: "settling" as HostTurnId,
+        input: [{ type: "text", text: "first" }],
+      });
+      const getEntries = vi.spyOn(transport, "getEntries").mockImplementationOnce(async () => {
+        await gate.promise;
+        return structuredClone(transport.history);
+      });
+      transport.succeed("done");
+      await vi.waitFor(() => expect(getEntries).toHaveBeenCalled());
+      expect(session.capabilities.steer).toBe(false);
+      expect(
+        await session.execute({
+          type: "turn.steer",
+          turnId: "settling" as HostTurnId,
+          input: [{ type: "text", text: "too late" }],
+        }),
+      ).toMatchObject({ ok: false, error: { code: "invalidState" } });
+      expect(steer).not.toHaveBeenCalled();
+      gate.resolve(undefined);
+      expect((await nextTurnEvents(iterator)).at(-1)).toMatchObject({
+        outcome: { status: "succeeded" },
+      });
+    } finally {
+      gate.resolve(undefined);
+      await adapter.close();
+    }
+  });
+
   it("propagates an unconfirmed delivery instead of reporting that input was refused", async () => {
     const transport = new FakeOmpTransport();
     transport.autoCompleteTurn = false;
diff --git a/packages/adapters/pi/src/pi-adapter.ts b/packages/adapters/pi/src/pi-adapter.ts
index fba57d980..041c6fc66 100644
--- a/packages/adapters/pi/src/pi-adapter.ts
+++ b/packages/adapters/pi/src/pi-adapter.ts
@@ -466,12 +466,13 @@ class PiHarnessSession implements HarnessSession {
       history: { fork: true, forkAcrossCwd: true, rollbackLastTurn: true },
       autonomousTurns: { observe: true },
       subagents: { observe: true, readTranscript: true },
-      // Read live: Pi's `steer` RPC delivers mid-Turn input into the same
-      // native run. A Turn whose cancellation is in flight must not take more
-      // input; every other case stops and restarts.
+      // Native input is unavailable during cancellation or completion settlement.
       get steer(): boolean {
         return (
-          session.#transport?.steer !== undefined && session.#active?.cancellationRequested !== true
+          session.#transport?.steer !== undefined &&
+          session.#active !== null &&
+          !session.#active.cancellationRequested &&
+          !session.#active.settling
         );
       },
     }))(this);
@@ -979,6 +980,9 @@ class PiHarnessSession implements HarnessSession {
     if (active.cancellationRequested) {
       return { ok: false, error: invalidState("Pi steer target is being cancelled") };
     }
+    if (active.settling) {
+      return { ok: false, error: invalidState("Pi steer target is recording its completion") };
+    }
     const text = command.input.map((input) => input.text).join("\n");
     if (text.length === 0) {
       return {
@@ -2008,17 +2012,13 @@ class PiHarnessSession implements HarnessSession {
       sessionId: transport.state.sessionId,
       model: nativeModelForHistory(transport.state),
     });
-    const created = snapshot.turns.filter(
+    // The active branch is ordered: the first User Entry after startup is the prompt
+    // that started this run. Native steers append more Users without changing that identity.
+    const turn = snapshot.turns.find(
       (turn) =>
         turn.input.length > 0 && !active.beforeNativeTurnKeys.has(turn.nativeTurnRef.nativeTurnKey),
     );
-    if (created.length !== 1) {
-      throw new Error(
-        `Pi Turn persisted ${created.length} new User Entries; exactly one is required`,
-      );
-    }
-    const turn = created[0];
-    if (!turn) throw new Error("Pi Turn has no persisted User identity");
+    if (!turn) throw new Error("Pi Turn has no newly persisted User Entry");
     return {
       nativeTurnRef: turn.nativeTurnRef,
       ...(turn.checkpoint ? { checkpoint: turn.checkpoint } : {}),
diff --git a/packages/adapters/pi/test/pi-adapter.test.ts b/packages/adapters/pi/test/pi-adapter.test.ts
index 3d537dd00..5cf553c4f 100644
--- a/packages/adapters/pi/test/pi-adapter.test.ts
+++ b/packages/adapters/pi/test/pi-adapter.test.ts
@@ -4194,6 +4194,153 @@ describe("Pi HarnessAdapter Session", () => {
 });
 
 describe("Pi native steer delivery", () => {
+  it("rejects completion without a newly persisted User Entry", async () => {
+    const f = fixture();
+    try {
+      const session = await openSession(f.adapter);
+      const iterator = session.outputs[Symbol.asyncIterator]();
+      await session.execute(textTurn("missing-history"));
+      const transport = f.transports[0];
+      if (!transport) throw new Error("Fake transport was not opened");
+      transport.resolveTurn?.({ text: "done", cancelled: false });
+      for (;;) {
+        const event = await nextEvent(iterator);
+        if (event.type !== "turn.completed") continue;
+        expect(event).toMatchObject({
+          outcome: {
+            status: "failed",
+            error: { code: "protocolError", message: "Pi Turn has no newly persisted User Entry" },
+          },
+        });
+        expect(event).not.toHaveProperty("nativeTurnRef");
+        break;
+      }
+    } finally {
+      await f.adapter.close();
+    }
+  });
+
+  it.each([false, true])(
+    "keeps the starting identity after two native steers (cancelled=%s)",
+    async (cancelled) => {
+      const f = fixture();
+      try {
+        const session = await openSession(f.adapter);
+        const iterator = session.outputs[Symbol.asyncIterator]();
+        await session.execute(textTurn("previous"));
+        const transport = f.transports[0];
+        if (!transport) throw new Error("Fake transport was not opened");
+        const completed = async () => {
+          for (;;) {
+            const event = await nextEvent(iterator);
+            if (event.type === "turn.completed") return event;
+          }
+        };
+        transport.succeed("previous answer");
+        await completed();
+        await session.execute(textTurn("steered"));
+        const steer = vi.fn(async () => undefined);
+        Object.assign(transport, { steer });
+        for (const [index, text] of ["first", "second", "third"].entries()) {
+          if (index > 0)
+            expect(
+              await session.execute({
+                type: "turn.steer",
+                turnId: hostTurnIdSchema.parse("steered"),
+                input: [{ type: "text", text }],
+              }),
+            ).toEqual({ ok: true, value: { accepted: true } });
+          const id = `current-user-${index}`;
+          transport.history.entries.push({
+            id,
+            parentId: transport.history.leafId,
+            type: "message",
+            message: { role: "user", content: [{ type: "text", text }] },
+          });
+          transport.history.leafId = id;
+        }
+        transport.history.entries.push({
+          id: "current-answer",
+          parentId: transport.history.leafId,
+          type: "message",
+          message: {
+            role: "assistant",
+            content: [{ type: "text", text: "done" }],
+            stopReason: cancelled ? "aborted" : "stop",
+          },
+        });
+        transport.history.leafId = "current-answer";
+        transport.resolveTurn?.({ text: "done", cancelled });
+        expect(await completed()).toMatchObject({
+          type: "turn.completed",
+          turnId: "steered",
+          nativeTurnRef: { nativeTurnKey: "current-user-0" },
+          outcome: {
+            status: cancelled ? "cancelled" : "succeeded",
+            checkpoint: { checkpointId: "current-user-0" },
+          },
+        });
+        expect(steer).toHaveBeenCalledTimes(2);
+        const snapshot = await session.readSnapshot();
+        expect(snapshot).toMatchObject({
+          ok: true,
+          value: {
+            turns: [
+              expect.anything(),
+              ...["first", "second", "third"].map((text, index) => ({
+                nativeTurnRef: { nativeTurnKey: `current-user-${index}` },
+                input: [{ type: "text", text }],
+              })),
+            ],
+          },
+        });
+        expect(await session.readSnapshot()).toEqual(snapshot);
+      } finally {
+        await f.adapter.close();
+      }
+    },
+  );
+
+  it("rejects steering while native completion history is being read", async () => {
+    const f = fixture();
+    const gate = Promise.withResolvers<undefined>();
+    try {
+      const session = await openSession(f.adapter);
+      const iterator = session.outputs[Symbol.asyncIterator]();
+      await session.execute(textTurn("settling"));
+      const transport = f.transports[0];
+      if (!transport) throw new Error("Fake transport was not opened");
+      const steer = vi.fn(async () => undefined);
+      Object.assign(transport, { steer });
+      transport.getEntries.mockClear().mockImplementationOnce(async () => {
+        await gate.promise;
+        return structuredClone(transport.history);
+      });
+      transport.succeed("done");
+      await vi.waitFor(() => expect(transport.getEntries).toHaveBeenCalled());
+      expect(session.capabilities.steer).toBe(false);
+      expect(
+        await session.execute({
+          type: "turn.steer",
+          turnId: hostTurnIdSchema.parse("settling"),
+          input: [{ type: "text", text: "too late" }],
+        }),
+      ).toMatchObject({ ok: false, error: { code: "invalidState" } });
+      expect(steer).not.toHaveBeenCalled();
+      gate.resolve(undefined);
+      for (;;) {
+        const event = await nextEvent(iterator);
+        if (event.type === "turn.completed") {
+          expect(event.outcome.status).toBe("succeeded");
+          break;
+        }
+      }
+    } finally {
+      gate.resolve(undefined);
+      await f.adapter.close();
+    }
+  });
+
   it("propagates an unconfirmed delivery instead of reporting that input was refused", async () => {
     const f = fixture();
     const session = await openSession(f.adapter);

From 99c92cc8d289ec35b7dab4ec5366f8b4283826d7 Mon Sep 17 00:00:00 2001
From: BytePioneer-AI <28417416@qq.com>
Date: Sun, 15 Nov 2026 22:51:35 +0800
Subject: [PATCH 14/15] fix(adapters): reject native steering without an active
 prompt

---
 docs/architecture/external-thread-steering.md |  2 +-
 packages/adapters/omp/src/omp-adapter.ts      |  6 +++-
 .../adapters/omp/test/omp-adapter.test.ts     | 33 +++++++++++++++++++
 packages/adapters/pi/src/pi-adapter.ts        |  6 +++-
 packages/adapters/pi/test/pi-adapter.test.ts  | 31 +++++++++++++++++
 5 files changed, 75 insertions(+), 3 deletions(-)

diff --git a/docs/architecture/external-thread-steering.md b/docs/architecture/external-thread-steering.md
index c3a17cfc0..25ea0d0ff 100644
--- a/docs/architecture/external-thread-steering.md
+++ b/docs/architecture/external-thread-steering.md
@@ -5,7 +5,7 @@ codexhost 对外部 Harness Thread 保留一个「调整方向」操作，按当
 - **原生同轮插入**：Session 此刻声明 `capabilities.steer` 时，Host 发送公共命令 `turn.steer`，由 Adapter 调用 Harness 自己的插入原语，在原生安全边界把输入送入**当前 Turn**，不取消、不另起 Turn。
 - **停止后重发（退路）**：其余情况下，Host **取消当前 Turn，等待它终结，再自动执行本次新输入**。不支持原生插入，或原生明确拒绝且输入未送达时走这条路径；写入后回执不确定时不自动重发。
 
-用户继续使用 Codex Desktop 原有按钮、跟进处理方式设置和单条消息反向操作，不需要手动停止后重发。官方 Codex Thread 仍透传原生 `turn/steer`，不改变同轮追加输入的语义。Claude Code、OMP 和 Pi 会声明 `capabilities.steer`：Claude Code 的条件与原生依据见 [`../harnesses/claude-code/claude-code-steer.md`](../harnesses/claude-code/claude-code-steer.md)；OMP/Pi 使用原生 `steer` RPC，已请求取消的 Turn 不接受原生插入。其他 Adapter 对 `turn.steer` 返回 `unsupported`，保持停止后重发。
+用户继续使用 Codex Desktop 原有按钮、跟进处理方式设置和单条消息反向操作，不需要手动停止后重发。官方 Codex Thread 仍透传原生 `turn/steer`，不改变同轮追加输入的语义。Claude Code、OMP 和 Pi 会声明 `capabilities.steer`：Claude Code 的条件与原生依据见 [`../harnesses/claude-code/claude-code-steer.md`](../harnesses/claude-code/claude-code-steer.md)；OMP/Pi 使用原生 `steer` RPC，仅接受仍在执行的 Prompt Turn；手动压缩等 Harness 命令、Pi 自主运行、已请求取消或正在记录完成历史的 Turn 不接受原生插入。其他 Adapter 对 `turn.steer` 返回 `unsupported`，保持停止后重发。
 
 ## 原生同轮插入
 
diff --git a/packages/adapters/omp/src/omp-adapter.ts b/packages/adapters/omp/src/omp-adapter.ts
index 39ac12176..b3fd2f015 100644
--- a/packages/adapters/omp/src/omp-adapter.ts
+++ b/packages/adapters/omp/src/omp-adapter.ts
@@ -731,11 +731,12 @@ class OmpHarnessSession implements HarnessSession {
       },
       history: { fork: true, forkAcrossCwd: true, rollbackLastTurn: true },
       subagents: { observe: true, readTranscript: true },
-      // Native input is unavailable during cancellation or completion settlement.
+      // Only a prompted Turn owns native steering, not a manual Harness command.
       get steer(): boolean {
         return (
           session.#transport?.steer !== undefined &&
           session.#active !== null &&
+          session.#active.command.input.length > 0 &&
           !session.#active.cancellationRequested &&
           !session.#active.settling
         );
@@ -1403,6 +1404,9 @@ class OmpHarnessSession implements HarnessSession {
         error: invalidState("Omp steer must reference the active Turn"),
       };
     }
+    if (active.command.input.length === 0) {
+      return { ok: false, error: invalidState("Omp steer requires an active prompt") };
+    }
     if (active.cancellationRequested) {
       return {
         ok: false,
diff --git a/packages/adapters/omp/test/omp-adapter.test.ts b/packages/adapters/omp/test/omp-adapter.test.ts
index 7d370ed0f..3c986e86e 100644
--- a/packages/adapters/omp/test/omp-adapter.test.ts
+++ b/packages/adapters/omp/test/omp-adapter.test.ts
@@ -2006,6 +2006,39 @@ describe("OMP Adapter Subagents", () => {
 });
 
 describe("OMP native steer delivery", () => {
+  it("does not advertise or deliver native steer during manual compaction", async () => {
+    const transport = new FakeOmpTransport();
+    const compaction = Promise.withResolvers<OmpCompactResult>();
+    vi.spyOn(transport, "compact").mockImplementationOnce(() => compaction.promise);
+    const steer = vi.fn(async () => undefined);
+    Object.assign(transport, { steer });
+    const adapter = new OmpAdapter({}, { createTransport: () => transport });
+    const turnId = "manual-compact-steer" as HostTurnId;
+    try {
+      const opened = await adapter.open({ kind: "create", cwd: "/synthetic" });
+      if (!opened.ok) throw new Error(opened.error.message);
+      const session = opened.value;
+      const commands = session.commands;
+      if (!commands) throw new Error("Omp Session did not expose commands");
+      expect(await commands.execute({ turnId, commandId: "omp.compact" })).toEqual({
+        ok: true,
+        value: { turnId },
+      });
+      expect.soft(session.capabilities.steer).toBe(false);
+      expect(
+        await session.execute({
+          type: "turn.steer",
+          turnId,
+          input: [{ type: "text", text: "new direction" }],
+        }),
+      ).toMatchObject({ ok: false, error: { code: "invalidState" } });
+      expect(steer).not.toHaveBeenCalled();
+    } finally {
+      compaction.resolve({ outcome: "succeeded" });
+      await adapter.close();
+    }
+  });
+
   it.each([false, true])(
     "keeps the starting identity after two native steers (cancelled=%s)",
     async (cancelled) => {
diff --git a/packages/adapters/pi/src/pi-adapter.ts b/packages/adapters/pi/src/pi-adapter.ts
index 041c6fc66..0d7fa600d 100644
--- a/packages/adapters/pi/src/pi-adapter.ts
+++ b/packages/adapters/pi/src/pi-adapter.ts
@@ -466,11 +466,12 @@ class PiHarnessSession implements HarnessSession {
       history: { fork: true, forkAcrossCwd: true, rollbackLastTurn: true },
       autonomousTurns: { observe: true },
       subagents: { observe: true, readTranscript: true },
-      // Native input is unavailable during cancellation or completion settlement.
+      // Only a prompted Turn owns native steering, not manual commands or autonomous work.
       get steer(): boolean {
         return (
           session.#transport?.steer !== undefined &&
           session.#active !== null &&
+          session.#active.command.input.length > 0 &&
           !session.#active.cancellationRequested &&
           !session.#active.settling
         );
@@ -977,6 +978,9 @@ class PiHarnessSession implements HarnessSession {
     if (!active || active.command.turnId !== command.turnId) {
       return { ok: false, error: invalidState("Pi steer must reference the active Turn") };
     }
+    if (active.command.input.length === 0) {
+      return { ok: false, error: invalidState("Pi steer requires an active prompt") };
+    }
     if (active.cancellationRequested) {
       return { ok: false, error: invalidState("Pi steer target is being cancelled") };
     }
diff --git a/packages/adapters/pi/test/pi-adapter.test.ts b/packages/adapters/pi/test/pi-adapter.test.ts
index 5cf553c4f..7cad452dc 100644
--- a/packages/adapters/pi/test/pi-adapter.test.ts
+++ b/packages/adapters/pi/test/pi-adapter.test.ts
@@ -4194,6 +4194,37 @@ describe("Pi HarnessAdapter Session", () => {
 });
 
 describe("Pi native steer delivery", () => {
+  it("does not advertise or deliver native steer during manual compaction", async () => {
+    const transport = new FakePiTransport();
+    const compaction = Promise.withResolvers<PiCompactResult>();
+    transport.compact.mockImplementationOnce(() => compaction.promise);
+    const steer = vi.fn(async () => undefined);
+    Object.assign(transport, { steer });
+    const adapter = new PiAdapter({}, { createTransport: () => transport });
+    const session = await openSession(adapter);
+    const turnId = hostTurnIdSchema.parse("manual-compact-steer");
+    try {
+      const commands = session.commands;
+      if (!commands) throw new Error("Pi Session did not expose commands");
+      expect(await commands.execute({ turnId, commandId: "pi.compact" })).toEqual({
+        ok: true,
+        value: { turnId },
+      });
+      expect.soft(session.capabilities.steer).toBe(false);
+      expect(
+        await session.execute({
+          type: "turn.steer",
+          turnId,
+          input: [{ type: "text", text: "new direction" }],
+        }),
+      ).toMatchObject({ ok: false, error: { code: "invalidState" } });
+      expect(steer).not.toHaveBeenCalled();
+    } finally {
+      compaction.resolve({ outcome: "succeeded" });
+      await adapter.close();
+    }
+  });
+
   it("rejects completion without a newly persisted User Entry", async () => {
     const f = fixture();
     try {

From 125d8730080806efa4377bbaf84256bc73b163d0 Mon Sep 17 00:00:00 2001
From: BytePioneer-AI <28417416@qq.com>
Date: Sun, 15 Nov 2026 22:51:36 +0800
Subject: [PATCH 15/15] fix(host): project submitted Harness command input

---
 docs/architecture/external-thread-queue.md    |  2 +-
 packages/host-runtime/src/app-server-host.ts  | 12 +++
 ...rver-host.command-input-projection.test.ts | 94 +++++++++++++++++++
 3 files changed, 107 insertions(+), 1 deletion(-)
 create mode 100644 packages/host-runtime/test/app-server-host.command-input-projection.test.ts

diff --git a/docs/architecture/external-thread-queue.md b/docs/architecture/external-thread-queue.md
index 72de18e8f..6bb8075a6 100644
--- a/docs/architecture/external-thread-queue.md
+++ b/docs/architecture/external-thread-queue.md
@@ -22,7 +22,7 @@ Desktop 在 Turn 运行中提交的跟进消息，经官方 app-server 的 `thre
 - 停止后重发已停止旧轮、替换启动失败且替换仍有效时，Host 在释放替换预留后继续推进队列。用户明确停止、Session fault 或过期/无效的 steer 请求不会解除暂停；等待命令目录期间发生的停止同样有效。
 - 运行中点击“引导”时，Desktop 先发 `turn/steer`，再 `thread/queue/delete`；投递方式由外部 steering 决定，见 `external-thread-steering.md`。
 - 开始失败时，提交放回原位置并再次发出 `thread/queue/changed`。
-- Desktop 只为自己 `turn/start` 的输入做本地乐观展示。由队列开始的 Turn 由 Host 投影带 `clientId` 的 userMessage，否则该轮用户消息不显示。
+- 带 `clientUserMessageId` 的 `turn/start` 和由队列开始的 Turn，均由 Host 投影带 `clientId` 的 userMessage，供 Desktop 去重；普通文本和已注册 Harness 命令使用同一规则。不带该 ID 的 Desktop `turn/start` 保留原生本地展示，Host-only 接口仍由 Host 投影输入。
 
 ## 边界
 
diff --git a/packages/host-runtime/src/app-server-host.ts b/packages/host-runtime/src/app-server-host.ts
index 9c527dd01..7c1e0540f 100644
--- a/packages/host-runtime/src/app-server-host.ts
+++ b/packages/host-runtime/src/app-server-host.ts
@@ -3759,6 +3759,7 @@ export class AppServerHost {
     arguments_: JsonObject | undefined,
     requestedTurnId?: HostTurnId,
     sourceHostId?: unknown,
+    projectedInput?: { text: string; clientUserMessageId?: string },
   ): Promise<{ turnId: HostTurnId; turn: JsonObject; gate: TurnProjectionGate }> {
     const commands = thread.session.commands;
     if (!commands) {
@@ -3779,6 +3780,14 @@ export class AppServerHost {
         turnId,
         cwd: thread.cwd,
         startedAtMs: Date.now(),
+        ...(projectedInput
+          ? {
+              initialInput: [{ type: "text" as const, text: projectedInput.text }],
+              ...(projectedInput.clientUserMessageId
+                ? { clientUserMessageId: projectedInput.clientUserMessageId }
+                : {}),
+            }
+          : {}),
       }),
     };
     const gate = turnProjectionGate();
@@ -5075,6 +5084,9 @@ export class AppServerHost {
         command.arguments,
         undefined,
         sourceHostId,
+        this.#options.externalOnly || projectInput
+          ? { text, ...(clientUserMessageId ? { clientUserMessageId } : {}) }
+          : undefined,
       );
     } catch (error) {
       if (error instanceof ExternalCommandError || error instanceof ExternalSteerError) throw error;
diff --git a/packages/host-runtime/test/app-server-host.command-input-projection.test.ts b/packages/host-runtime/test/app-server-host.command-input-projection.test.ts
new file mode 100644
index 000000000..1257f7676
--- /dev/null
+++ b/packages/host-runtime/test/app-server-host.command-input-projection.test.ts
@@ -0,0 +1,94 @@
+import { expect, it } from "vitest";
+import type { JsonObject } from "@codexhost/protocol-core";
+import { harnessCommandDescriptorSchema, hostItemIdSchema } from "@codexhost/shared-contracts";
+
+import {
+  createFixture,
+  messageParams,
+  method,
+  requestId,
+  startPiThread,
+  stopFixture,
+  turnEvent,
+  writeRequest,
+} from "./app-server-host-fixture.js";
+
+const command = harnessCommandDescriptorSchema.parse({
+  id: "fake.review",
+  invocation: "/review",
+  label: "Review",
+  argumentMode: "text",
+});
+
+it.each([
+  { source: "turn/start", clientId: "desktop-command", projected: true },
+  { source: "turn/start", clientId: undefined, projected: false },
+  { source: "thread/queue/start", clientId: "queued-command", projected: true },
+])(
+  "projects command input for $source with clientId=$clientId: $projected",
+  async ({ source, clientId, projected }) => {
+    const fixture = createFixture();
+    try {
+      const threadId = await startPiThread(fixture);
+      const session = fixture.adapter.sessions[0];
+      if (!session) throw new Error("Fake Session was not opened");
+      session.commands = {
+        list: async () => ({ ok: true, value: { commands: [command] } }),
+        execute: async ({ turnId }) => {
+          session.publishEphemeralCommand(turnId, {
+            type: "contextCompaction",
+            itemId: hostItemIdSchema.parse("command-output"),
+          });
+          return { ok: true, value: { turnId } };
+        },
+      };
+      const text = "/review keep  argument spacing";
+      const input = [{ type: "text", text }];
+      if (source === "thread/queue/start") {
+        writeRequest(fixture.desktopInput, {
+          id: 2,
+          method: "thread/queue/add",
+          params: { threadId, input, ...(clientId ? { clientUserMessageId: clientId } : {}) },
+        });
+        await fixture.collector.waitFor((message) => requestId(message, 2));
+      }
+      writeRequest(fixture.desktopInput, {
+        id: 3,
+        method: source,
+        params: {
+          threadId,
+          ...(source === "turn/start" ? { input } : {}),
+          ...(clientId ? { clientUserMessageId: clientId } : {}),
+        },
+      });
+      const response = await fixture.collector.waitFor((message) => requestId(message, 3));
+      expect(response).not.toHaveProperty("error");
+      const turn = (response.result as JsonObject).turn as JsonObject;
+      if (typeof turn.id !== "string") throw new Error("Command response has no Turn ID");
+      const turnId = turn.id;
+      const completed = await fixture.collector.waitFor((message) =>
+        turnEvent(message, "turn/completed", turnId),
+      );
+      const userItems = (eventMethod: string) =>
+        fixture.collector.messages
+          .filter(
+            (message) =>
+              method(message, eventMethod) &&
+              (messageParams(message).item as JsonObject)?.type === "userMessage",
+          )
+          .map((message) => messageParams(message).item);
+      const startedItems = userItems("item/started");
+      expect(startedItems).toHaveLength(projected ? 1 : 0);
+      expect(userItems("item/completed")).toEqual(startedItems);
+      const completedTurn = messageParams(completed).turn as JsonObject;
+      expect(
+        (completedTurn.items as JsonObject[]).filter((item) => item.type === "userMessage"),
+      ).toEqual(startedItems);
+      if (projected) {
+        expect(startedItems[0]).toMatchObject({ clientId, content: [{ type: "text", text }] });
+      }
+    } finally {
+      await stopFixture(fixture);
+    }
+  },
+);