feat(runtime): stall observation + optional reconcile capability

src/adapters/codex/adapter.ts

@@ -25,6 +25,9 @@ import type {
   PromptReceipt,
   ProviderSessionRef,
   QuestionHandler,
+  Reconcilable,
+  ReconcileState,
+  ReconcileVerdict,
   SteerReceipt,
 } from "../types.ts";
 import { unsupportedPromptBlocks } from "../types.ts";
@@ -247,6 +250,27 @@ export interface CodexAdapterOptions {
  * turn/start 刻意不设——老版本 app-server 会合法地阻塞到 turn 结束。
  */
 const STARTUP_REQUEST_TIMEOUT_MS = 30_000;
+const RECONCILE_REQUEST_TIMEOUT_MS = 10_000;
+
+/**
+ * 把 codex `thread/read` 的 live `thread.status`（wire 形状 `{type, activeFlags?}`）翻译成
+ * baton 对账裁决。只信 live status——`includeTurns` 的 item 来自 rollout history，可能和
+ * 已漏掉的事件一样陈旧。notLoaded / systemError / 缺失一律 unknown：查不到就保守不收口。
+ */
+export function mapThreadStatus(status: { type?: string; activeFlags?: string[] } | undefined): ReconcileState {
+  switch (status?.type) {
+    case "idle":
+      return "idle";
+    case "active": {
+      const flags = status.activeFlags ?? [];
+      if (flags.includes("waitingOnApproval")) return "waiting_approval";
+      if (flags.includes("waitingOnUserInput")) return "waiting_input";
+      return "active";
+    }
+    default:
+      return "unknown";
+  }
+}
 
 /**
  * 计入"turn 有产出"的事件 kind（空回合判定，见 CodexTurn.sawOutput）。
@@ -307,15 +331,20 @@ export async function openCodexThread(
   return { threadId: threadIdFrom(response, "thread/start"), resumed: false };
 }
 
-export class CodexAdapter implements AgentAdapter {
+export class CodexAdapter implements AgentAdapter, Reconcilable {
   readonly provider = "codex";
   // 当前 adapter 最终只发送 text（design.md §3.1）；可选能力接口落地并验证后才声明
   // 对应 marker——契约测试钉住"声明支持就必须实现对应接口"。
   // sync：catch-up 走 turn/start.additionalContext（experimental API，initialize 已声明
   // experimentalApi）。曾用 thread/inject_items 注入独立 user message，但那会污染 codex
   // 原生历史（rollout 里出现无对应回合的悬空 user message）；additionalContext 由 codex
   // 以 contextual fragment 形态随本 turn 入史，且不过 UserPromptSubmit hook。
-  readonly capabilities: AdapterCapabilities = { prompt: {}, steer: { supported: true }, sync: { supported: true } };
+  readonly capabilities: AdapterCapabilities = {
+    prompt: {},
+    steer: { supported: true },
+    sync: { supported: true },
+    reconcile: { supported: true },
+  };
   private threads = new Map<string, ThreadRuntime>();
 
   constructor(private options: CodexAdapterOptions) {}
@@ -483,6 +512,22 @@ export class CodexAdapter implements AgentAdapter {
     await rt.peer.request("turn/interrupt", { threadId: rt.threadId, turnId: rt.codexTurnId });
   }
 
+  /**
+   * 对账：查 live `thread.status`（design：docs/provider-output-lifecycle.md §5）。
+   * turnId 不入参 wire——codex thread 同一时刻 ≤1 活跃 turn，thread 级 status 即该 turn
+   * 的状态；runtime 只对活跃 turn 发起对账，二者一致。探针自带超时，失败由调用方读作 unknown。
+   */
+  async reconcile(ref: ProviderSessionRef, _turnId: string): Promise<ReconcileVerdict> {
+    const rt = this.mustThread(ref);
+    const response = await rt.peer.request(
+      "thread/read",
+      { threadId: rt.threadId, includeTurns: false },
+      { timeoutMs: RECONCILE_REQUEST_TIMEOUT_MS },
+    );
+    const status = (response as { thread?: { status?: { type?: string; activeFlags?: string[] } } })?.thread?.status;
+    return { state: mapThreadStatus(status), detail: status?.type };
+  }
+
   /** cancel 早于 codex turn id 就位时的补发：fire-and-forget，失败由 runtime 宽限期兜底 */
   private flushPendingCancel(rt: ThreadRuntime): void {
     if (!rt.pendingCancel || !rt.codexTurnId) return;

src/adapters/types.ts

@@ -82,6 +82,12 @@ export interface AdapterCapabilities {
   sync?: CapabilityMarker;
   commands?: CapabilityMarker;
   config?: CapabilityMarker;
+  /**
+   * 支持对账：runtime 在 turn 停滞时可主动查询 provider 侧该 turn 的真实运行态
+   * （见 Reconcilable / docs/provider-output-lifecycle.md §5）。未声明的 provider
+   * 回落——只发 stall 提示、交用户 cancel，不做主动对账。
+   */
+  reconcile?: CapabilityMarker;
   interactions?: {
     permission?: CapabilityMarker;
     question?: CapabilityMarker;
@@ -115,6 +121,34 @@ export interface AgentAdapter {
   close(ref: ProviderSessionRef): Promise<void>;
 }
 
+/**
+ * 对账裁决：provider 眼中某 turn 此刻的真实运行态（docs/provider-output-lifecycle.md §5）。
+ * - `idle`：provider 已结束——baton 漏了终态，runtime 自愈式 finalize（最常见）；
+ * - `active`：确在跑，保留 stall 提示交用户决策，不动终态；
+ * - `waiting_approval` / `waiting_input`：被审批/输入阻塞，上报对应待决；
+ * - `unknown`：查不到 / 探针失败——保守不收口，交用户。
+ */
+export type ReconcileState = "idle" | "active" | "waiting_approval" | "waiting_input" | "unknown";
+
+export interface ReconcileVerdict {
+  state: ReconcileState;
+  /** provider 侧原始状态标识，仅诊断/日志用，不参与 runtime 决策 */
+  detail?: string;
+}
+
+/**
+ * 可选能力：对账（reconcile capability）。声明 `capabilities.reconcile` 的 adapter 必须实现。
+ * 各家用各自基元：Codex 拉 `thread/read` 的 live `thread.status`；Claude 无对等状态查询，
+ * 因此 v1 不声明、回落到仅停滞提示。语义只回答"这个 turn 现在如何"，收口动作由 runtime 决定。
+ */
+export interface Reconcilable {
+  reconcile(ref: ProviderSessionRef, turnId: string): Promise<ReconcileVerdict>;
+}
+
+export function isReconcilable(adapter: AgentAdapter): adapter is AgentAdapter & Reconcilable {
+  return typeof (adapter as Partial<Reconcilable>).reconcile === "function";
+}
+
 /** admission 检查：返回 capabilities 未声明支持的 block 类型（text 恒支持） */
 export function unsupportedPromptBlocks(
   blocks: PromptBlock[],

src/events/types.ts

@@ -289,6 +289,20 @@ export interface Notice {
   detail?: string;
 }
 
+/**
+ * turn 静默停滞观测事件（停滞观测，见 docs/provider-output-lifecycle.md §5）。turnId 在信封上。
+ * 纯观测：标记"活跃 turn 长时间无任何事件"，供投影提示与后续对账（reconcile）触发。
+ * **绝不驱动 finalize**——provider-interaction-design.md §4.1 明确不设强制 finalize 的
+ * watchdog（合法长任务不该被误杀）；收口只由真实终态、reconcile 的 idle 裁决或用户
+ * cancel 触发。stalled 是可自愈的观测态：活动恢复即发 cleared=true 撤除提示。
+ */
+export interface StallNotice {
+  /** 触发时已静默的毫秒数（cleared 事件为 0） */
+  stalledMs: number;
+  /** true = 停滞解除（活动恢复或已收口），投影据此清除提示 */
+  cleared?: boolean;
+}
+
 /**
  * 短寿命运行阶段快照（compacting…），见 docs/design.md §5.2 归一表"运行阶段"行。
  * phase 开放字符串（forward-compat）；null = 阶段结束，投影层回落默认 thinking。
@@ -353,6 +367,7 @@ export type EventPayloadMap = {
   context_usage_update: ContextUsageUpdate;
   _baton_error_update: ErrorUpdate;
   _baton_notice: Notice;
+  _baton_stall_notice: StallNotice;
   _baton_run_status: RunStatusUpdate;
   _baton_turn_summary: TurnSummary;
 };

src/session/runtime.ts

@@ -2,6 +2,7 @@ import {
   isContextSynchronizable,
   isModelConfigurable,
   isNativeSessionIdentifiable,
+  isReconcilable,
   isSteerable,
   type AgentAdapter,
   type ApprovalDecision,
@@ -62,6 +63,12 @@ interface TurnRecord {
   status: "active" | "finalized";
   startedAt: number;
   stopReason?: StopReason;
+  /** 停滞观测：命中该 turn 的任何事件都刷新（onAdapterEvent 单点 stamp） */
+  lastActivityTs: number;
+  /** 已发过 stall notice 且未清除——去重发射，并在活动恢复时补 cleared */
+  stalled?: boolean;
+  /** 对账探针在途——防止同一 turn 的多次 stall tick 重叠发起 reconcile */
+  reconciling?: boolean;
   /** driven 专属：入队原件（canSteer/steer 需要用户侧 provider 名）。finalize 后由 retire 释放 */
   turn?: QueuedTurn;
   /** driven 专属：finalize 时 resolve，释放 drain 循环推进队列。finalize 后由 retire 释放 */
@@ -106,13 +113,41 @@ export interface BatonSessionRuntimeOptions {
    * 合法的长任务不应被误杀）。
    */
   cancelGraceMs?: number;
+  /**
+   * 活跃 turn 静默多久算停滞。到期只发 `_baton_stall_notice`（可见 + 供对账
+   * reconcile 触发），**绝不 finalize**——见 StallNotice 与 §4.1。默认较大以避免误报。
+   */
+  stallThresholdMs?: number;
+  /** 停滞扫描间隔；仅在有活跃 turn 时运行，全部收口后自停 */
+  stallPollMs?: number;
 }
 
 const DEFAULT_CANCEL_GRACE_MS = 10_000;
+const DEFAULT_STALL_THRESHOLD_MS = 120_000;
+const DEFAULT_STALL_POLL_MS = 10_000;
+const RECONCILE_TIMEOUT_MS = 15_000;
 
 /** 打断标记文案：cancelled 终态时落一条 notice，TUI 时间线醒目提示（对齐 Codex 的体验） */
 export const INTERRUPTED_NOTICE_TITLE = "Conversation interrupted — tell the agent what to do differently";
 
+/** 探针超时护栏：对账本身若挂住不能反过来拖住 runtime，超时按 reject 处理（读作 unknown）。 */
+function withTimeout<T>(promise: Promise<T>, ms: number): Promise<T> {
+  return new Promise<T>((resolve, reject) => {
+    const timer = setTimeout(() => reject(new Error("reconcile probe timeout")), ms);
+    timer.unref?.();
+    promise.then(
+      (value) => {
+        clearTimeout(timer);
+        resolve(value);
+      },
+      (error) => {
+        clearTimeout(timer);
+        reject(error);
+      },
+    );
+  });
+}
+
 /**
  * 一个 BatonSession 的唯一 turn 编排入口：统一负责 provider 恢复、上下文追平与全局串行。
  * UI 只提交意图和消费事件（经 SessionHandle.subscribe 订阅事件流），不能分别维护
@@ -152,6 +187,8 @@ export class BatonSessionRuntime {
    */
   private readonly pendingApprovals = new Map<string, (d: ApprovalDecision) => void>();
   private readonly pendingQuestions = new Map<string, (d: QuestionDecision) => void>();
+  /** 停滞扫描器：懒启动（有活跃 turn 才跑）、无活跃 turn 时自停、close 时清除 */
+  private stallMonitor?: ReturnType<typeof setInterval>;
 
   constructor(private readonly options: BatonSessionRuntimeOptions) {}
 
@@ -356,6 +393,7 @@ export class BatonSessionRuntime {
   }
 
   async close(): Promise<void> {
+    this.stopStallMonitor();
     const closing: Promise<void>[] = [];
     for (const slot of this.slots.values()) {
       if (slot.ref) closing.push(slot.adapter.close(slot.ref).catch(() => {}));
@@ -404,11 +442,13 @@ export class BatonSessionRuntime {
       provider: providerKey,
       status: "active",
       startedAt: Date.now(),
+      lastActivityTs: Date.now(),
       turn,
       release,
     };
     this.turns.set(turn.turnId, record);
     this.activeDrivenTurnId = turn.turnId;
+    this.ensureStallMonitor();
     const knownProviderSessionId = this.options.session.meta.providerSessions[providerKey]?.providerSessionId;
     this.onAdapterEvent(slot, {
       kind: "user_message",
@@ -539,6 +579,9 @@ export class BatonSessionRuntime {
    */
   private onAdapterEvent(slot: ProviderSlot, ev: AnyNewEvent): void {
     const envelope = this.options.session.append(ev) as AnyEventEnvelope;
+    // 停滞观测：命中活跃 turn 的任何事件都刷新进展时钟；曾停滞则补 cleared 撤除提示。
+    // 放在这里（唯一事件入口）保证不漏，且在下面的生命周期路由之前先刷新。
+    this.refreshActivity(envelope.turnId);
     if (envelope.kind === "state_update") {
       const p = envelope.payload;
       if (p.state === "running" && p.origin === "provider" && envelope.turnId) {
@@ -551,7 +594,9 @@ export class BatonSessionRuntime {
             provider: slot.adapter.provider,
             status: "active",
             startedAt: Date.now(),
+            lastActivityTs: Date.now(),
           });
+          this.ensureStallMonitor();
         }
       } else if (p.state === "idle") {
         // 终态一律按 baton turn id 查表路由（不看 slot）。无 turnId 的终态：
@@ -724,6 +769,100 @@ export class BatonSessionRuntime {
     return { id: turn.id, turnId: turn.turnId, provider: turn.provider, blocks: [...turn.blocks] };
   }
 
+  /** 刷新 turn 进展时钟；曾停滞的 turn 在活动恢复时补发 cleared 撤除提示。 */
+  private refreshActivity(turnId: string | undefined): void {
+    if (!turnId) return;
+    const record = this.turns.get(turnId);
+    if (!record || record.status !== "active") return;
+    record.lastActivityTs = Date.now();
+    if (record.stalled) {
+      record.stalled = false;
+      this.options.session.append({
+        kind: "_baton_stall_notice",
+        provider: record.provider,
+        turnId,
+        payload: { stalledMs: 0, cleared: true },
+      });
+      this.changed();
+    }
+  }
+
+  /** 懒启动停滞扫描器（幂等）。只在有活跃 turn 时创建，checkStalls 无活跃 turn 时自停。 */
+  private ensureStallMonitor(): void {
+    if (this.stallMonitor) return;
+    const pollMs = this.options.stallPollMs ?? DEFAULT_STALL_POLL_MS;
+    this.stallMonitor = setInterval(() => this.checkStalls(), pollMs);
+    // 扫描器不应阻止进程退出（headless / 测试）——真正的收口靠事件与 close()
+    this.stallMonitor.unref?.();
+  }
+
+  /**
+   * 扫描活跃 turn，静默超阈值发一次 `_baton_stall_notice`（按 record.stalled 去重）。
+   * 纯观测——不 finalize、不 cancel（§4.1 不设强制 finalize watchdog）。无活跃 turn 时自停。
+   */
+  private checkStalls(): void {
+    const threshold = this.options.stallThresholdMs ?? DEFAULT_STALL_THRESHOLD_MS;
+    const now = Date.now();
+    let active = 0;
+    for (const record of this.turns.values()) {
+      if (record.status !== "active") continue;
+      active++;
+      const gap = now - record.lastActivityTs;
+      if (gap < threshold) continue;
+      if (!record.stalled) {
+        record.stalled = true;
+        this.options.session.append({
+          kind: "_baton_stall_notice",
+          provider: record.provider,
+          turnId: record.turnId,
+          payload: { stalledMs: gap },
+        });
+        this.changed();
+      }
+      // 每个停滞 tick 尝试对账（有能力时）。放在标记之后——即便对账让它自愈，
+      // 用户也已（短暂）看到过 stall；reconciling 去重防重叠，探针 idle 才收口。
+      void this.reconcileStalled(record);
+    }
+    if (active === 0) this.stopStallMonitor();
+  }
+
+  /**
+   * 对账：turn 停滞时查 provider 侧真实运行态，把"猜"升级成"问"。
+   * - `idle` → provider 已结束、baton 漏了终态：合成 idle 走正常 finalize（自愈，含 summary）；
+   * - 其余（active / waiting_* / unknown / 探针失败）→ 保留 stall 提示，交用户 cancel。
+   * **只有 idle 裁决才收口**——不确定一律不动终态，守住"悲观不静默判死"。未声明 reconcile
+   * 能力的 provider（如 Claude）直接返回，只走停滞提示。
+   */
+  private async reconcileStalled(record: TurnRecord): Promise<void> {
+    const adapter = record.slot.adapter;
+    const ref = record.slot.ref;
+    if (!ref || !adapter.capabilities.reconcile || !isReconcilable(adapter)) return;
+    if (record.reconciling) return;
+    record.reconciling = true;
+    try {
+      const verdict = await withTimeout(adapter.reconcile(ref, record.turnId), RECONCILE_TIMEOUT_MS);
+      if (record.status !== "active") return; // 期间已被真实终态收口，裁决作废
+      if (verdict.state === "idle") {
+        this.onAdapterEvent(record.slot, {
+          kind: "state_update",
+          provider: record.provider,
+          turnId: record.turnId,
+          payload: { state: "idle", stopReason: "reconciled" },
+        });
+      }
+    } catch {
+      // 探针超时/报错 = unknown：保守不收口，保留 stall 提示。
+    } finally {
+      record.reconciling = false;
+    }
+  }
+
+  private stopStallMonitor(): void {
+    if (!this.stallMonitor) return;
+    clearInterval(this.stallMonitor);
+    this.stallMonitor = undefined;
+  }
+
   private changed(): void {
     this.options.onStateChange?.();
   }