diff --git a/docs/architecture/external-thread-queue.md b/docs/architecture/external-thread-queue.md
index 611dc3e73..6bb8075a6 100644
--- a/docs/architecture/external-thread-queue.md
+++ b/docs/architecture/external-thread-queue.md
@@ -19,9 +19,10 @@ 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，否则该轮用户消息不显示。
+- 带 `clientUserMessageId` 的 `turn/start` 和由队列开始的 Turn，均由 Host 投影带 `clientId` 的 userMessage，供 Desktop 去重；普通文本和已注册 Harness 命令使用同一规则。不带该 ID 的 Desktop `turn/start` 保留原生本地展示，Host-only 接口仍由 Host 投影输入。
 
 ## 边界
 
diff --git a/docs/architecture/external-thread-steering.md b/docs/architecture/external-thread-steering.md
index ec83b47f5..25ea0d0ff 100644
--- a/docs/architecture/external-thread-steering.md
+++ b/docs/architecture/external-thread-steering.md
@@ -3,9 +3,9 @@
 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 会声明 `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，仅接受仍在执行的 Prompt Turn；手动压缩等 Harness 命令、Pi 自主运行、已请求取消或正在记录完成历史的 Turn 不接受原生插入。其他 Adapter 对 `turn.steer` 返回 `unsupported`，保持停止后重发。
 
 ## 原生同轮插入
 
@@ -13,6 +13,8 @@ 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 已有的展示不一致。
   2. 输入是 Harness 命令（Session 提供命令、文本以 `/命令` 开头）时停止后重发，命令照常解析和执行。
@@ -31,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 69f63e1e5..b3fd2f015 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,6 +161,8 @@ export interface OmpTurnTransport {
     onEvent: (event: OmpTurnEvent) => void,
   ): Promise<OmpCompactResult>;
   runTurn(text: string, onEvent: (event: OmpTurnEvent) => void): Promise<OmpTurnResult>;
+  /** 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>;
   close(): Promise<void>;
@@ -218,6 +221,7 @@ interface ActiveTurn {
   interactionByNativeId: Map<string, HostInteractionId>;
   subagents: OmpSubagentLifecycle;
   cancellationRequested: boolean;
+  settling: boolean;
   beforeNativeTurnKeys: Set<string>;
   completion: Promise<void>;
   resolveCompletion(): void;
@@ -718,16 +722,26 @@ 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 },
-    };
+      // 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
+        );
+      },
+    }))(this);
     this.commands = {
       list: async () => ({ ok: true, value: this.#liveCommandCatalog() }),
       execute: (command) => this.#executeHarnessCommand(command),
@@ -898,7 +912,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);
@@ -981,6 +995,7 @@ class OmpHarnessSession implements HarnessSession {
           emit: (event) => this.#event(event),
         }),
         cancellationRequested: false,
+        settling: false,
         beforeNativeTurnKeys: new Set(
           beforeHistory.turns.map((turn) => turn.nativeTurnRef.nativeTurnKey),
         ),
@@ -994,6 +1009,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,
@@ -1030,6 +1046,7 @@ class OmpHarnessSession implements HarnessSession {
           }
         })
         .catch(async (error: unknown) => {
+          active.settling = true;
           let identity:
             { nativeTurnRef: NativeTurnRef; checkpoint: NativeCheckpointRef } | undefined;
           try {
@@ -1372,6 +1389,64 @@ 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.command.input.length === 0) {
+      return { ok: false, error: invalidState("Omp steer requires an active prompt") };
+    }
+    if (active.cancellationRequested) {
+      return {
+        ok: false,
+        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 {
+        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) {
+      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 } };
+  }
+
   async #cancel(command: TurnCancelCommand): Promise<HarnessResult<TurnCancelAccepted>> {
     const active = this.#active;
     if (!active || active.command.turnId !== command.turnId) {
@@ -1505,6 +1580,7 @@ class OmpHarnessSession implements HarnessSession {
           emit: (event) => this.#event(event),
         }),
         cancellationRequested: false,
+        settling: false,
         beforeNativeTurnKeys: new Set(),
         completion,
         resolveCompletion,
@@ -2076,16 +2152,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/src/omp-rpc-session.ts b/packages/adapters/omp/src/omp-rpc-session.ts
index b5ded396c..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;
@@ -547,6 +555,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;
@@ -952,6 +973,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");
   }
@@ -1194,6 +1232,31 @@ export class OmpRpcSession {
       void this.#confirmCancelledTurn(active);
       return;
     }
+    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) {
       if (value.type === "extension_ui_request" && this.#isBlockingInteraction(value)) {
         this.#fail(
@@ -1596,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));
       }
@@ -1807,12 +1880,30 @@ 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";
+        // 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-adapter.test.ts b/packages/adapters/omp/test/omp-adapter.test.ts
index d0a58eeb5..3c986e86e 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 {
@@ -1316,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,
         },
       },
@@ -2000,3 +2004,238 @@ describe("OMP Adapter Subagents", () => {
     await adapter.close();
   });
 });
+
+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) => {
+      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;
+    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 483d5bc01..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,
@@ -202,6 +203,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 });
@@ -441,6 +443,241 @@ 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();
+    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();
@@ -1197,6 +1434,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 };
@@ -2190,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 40749ca96..0d7fa600d 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,6 +176,8 @@ export interface PiTurnTransport {
     onEvent: (event: PiTurnEvent) => void,
   ): Promise<PiCompactResult>;
   runTurn(text: string, onEvent: (event: PiTurnEvent) => void): Promise<PiTurnResult>;
+  /** 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>;
   close(): Promise<void>;
@@ -453,17 +456,27 @@ 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 },
-    };
+      // 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
+        );
+      },
+    }))(this);
     this.commands = {
       list: async () => ({ ok: true, value: await this.#refreshLiveCommands() }),
       execute: (command) => this.#executeHarnessCommand(command),
@@ -587,7 +600,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 +968,56 @@ 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.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") };
+    }
+    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 {
+        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) {
+      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 } };
+  }
+
   async #cancel(command: TurnCancelCommand): Promise<HarnessResult<TurnCancelAccepted>> {
     const active = this.#active;
     if (!active || active.command.turnId !== command.turnId) {
@@ -1953,17 +2016,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/src/pi-rpc-session.ts b/packages/adapters/pi/src/pi-rpc-session.ts
index 776d64c5f..c5b400d79 100644
--- a/packages/adapters/pi/src/pi-rpc-session.ts
+++ b/packages/adapters/pi/src/pi-rpc-session.ts
@@ -196,6 +196,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;
@@ -1064,6 +1072,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");
   }
@@ -1702,8 +1731,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..7cad452dc 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,238 @@ 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 {
+      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);
+    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 7a6114e35..dbf5d752c 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,
@@ -80,6 +81,8 @@ class FakePiRpcProcess extends EventEmitter {
   signalCode: NodeJS.Signals | null = null;
   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;
@@ -275,10 +278,16 @@ 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") {
+      if (this.steerReply)
+        this.#output({ type: "response", id: command.id, command: "steer", ...this.steerReply });
+      return;
+    }
     if (command.type === "get_state" && this.holdStates) {
       this.#heldStates.push(command);
       return;
@@ -1547,6 +1556,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 notices without faulting and ignores ephemeral footer updates", async () => {
     const fake = new FakePiRpcProcess("manual-run");
     const onFault = vi.fn();
@@ -3446,3 +3491,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();
+  }
+});
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"]);
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 fecf3653b..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();
@@ -4772,13 +4781,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 } }));
@@ -4800,6 +4816,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[] => {
@@ -4940,20 +4962,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 {
@@ -4975,6 +5006,7 @@ export class AppServerHost {
           errorMessage(error),
         ),
       );
+      if (replacementStartFailed) void this.#drainExternalQueue(thread);
     } finally {
       this.#signalActiveWorkChanged();
     }
@@ -5052,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;
@@ -5563,13 +5598,25 @@ 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.
+      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.
       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"))
       ) {
+        this.#options.diagnosticOutput.write(
+          `codexhost Host Runtime: draining queue thread=${thread.id.slice(0, 8)}\n`,
+        );
         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.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);
+    }
+  },
+);
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..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
@@ -234,4 +234,282 @@ 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);
+    }
+  });
+
+  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.
+    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);
+    }
+  });
+});
+
+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);
+    }
+  });
 });
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();