fix(omp): wait for native background work instead of faulting a settled Turn

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 投影输入。
 
 ## 边界
 

docs/architecture/external-thread-steering.md

@@ -3,16 +3,18 @@
 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`，保持停止后重发。
 
 ## 原生同轮插入
 
 - `HarnessSessionCapabilities.steer` 表示 Session 当前是否允许原生同轮插入；缺省表示不支持。直连 Adapter 可用普通属性或实时 getter。跨进程 Session 可提供 `readCapabilities()`，Host 在预判和未指定送达方式的执行请求中优先读取它；读取失败不使用旧快照发送输入。
 - 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 进程重启后自然消失。

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