From 64c3d131e1c3806d967012ef5bc205ebf234628a Mon Sep 17 00:00:00 2001 From: "liqiankun.1111" Date: Mon, 13 Jul 2026 03:01:43 -0700 Subject: [PATCH] feat(runtime): stall observation + optional reconcile capability MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Detect silently-hung turns (an active turn with no events past a threshold) and surface them as a self-healing observed state, without ever force- finalizing — the terminal-state contract stays owned by real terminals, reconcile, or user cancel (no global watchdog). Add an optional reconcile capability: on stall, query the provider's real turn state. An "idle" verdict self-heals a missed terminal (routes a normal idle → finalize + summary); active / waiting / unknown / probe-failure keep the stall notice for the user to cancel. Codex implements reconcile via thread/read's live status; Claude declares nothing and falls back to stall-only observation. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/adapters/codex/adapter.ts | 49 +++++++++- src/adapters/types.ts | 34 +++++++ src/events/types.ts | 15 ++++ src/session/runtime.ts | 139 ++++++++++++++++++++++++++++ src/store/reduce.ts | 13 +++ src/tui/protocol.ts | 7 ++ tests/capabilities.test.ts | 1 + tests/reconcile.test.ts | 164 ++++++++++++++++++++++++++++++++++ tests/stall.test.ts | 148 ++++++++++++++++++++++++++++++ tests/tui-protocol.test.ts | 11 +++ 10 files changed, 579 insertions(+), 2 deletions(-) create mode 100644 tests/reconcile.test.ts create mode 100644 tests/stall.test.ts diff --git a/src/adapters/codex/adapter.ts b/src/adapters/codex/adapter.ts index 884e3cd..a4c1522 100644 --- a/src/adapters/codex/adapter.ts +++ b/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,7 +331,7 @@ 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——契约测试钉住"声明支持就必须实现对应接口"。 @@ -315,7 +339,12 @@ export class CodexAdapter implements AgentAdapter { // 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(); 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 { + 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; diff --git a/src/adapters/types.ts b/src/adapters/types.ts index 9dc375a..92c5f28 100644 --- a/src/adapters/types.ts +++ b/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; } +/** + * 对账裁决：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; +} + +export function isReconcilable(adapter: AgentAdapter): adapter is AgentAdapter & Reconcilable { + return typeof (adapter as Partial).reconcile === "function"; +} + /** admission 检查：返回 capabilities 未声明支持的 block 类型（text 恒支持） */ export function unsupportedPromptBlocks( blocks: PromptBlock[], diff --git a/src/events/types.ts b/src/events/types.ts index 4129a72..71c3fb8 100644 --- a/src/events/types.ts +++ b/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; }; diff --git a/src/session/runtime.ts b/src/session/runtime.ts index 33a241f..f17599a 100644 --- a/src/session/runtime.ts +++ b/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(promise: Promise, ms: number): Promise { + return new Promise((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 void>(); private readonly pendingQuestions = new Map void>(); + /** 停滞扫描器：懒启动（有活跃 turn 才跑）、无活跃 turn 时自停、close 时清除 */ + private stallMonitor?: ReturnType; constructor(private readonly options: BatonSessionRuntimeOptions) {} @@ -356,6 +393,7 @@ export class BatonSessionRuntime { } async close(): Promise { + this.stopStallMonitor(); const closing: Promise[] = []; 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 { + 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?.(); } diff --git a/src/store/reduce.ts b/src/store/reduce.ts index 360de48..42e427c 100644 --- a/src/store/reduce.ts +++ b/src/store/reduce.ts @@ -75,6 +75,8 @@ export interface ActiveTurnState { startedAt?: number; /** per-turn 运行阶段（compacting…）：null phase 或本 turn idle 清除（阶段不跨 turn） */ phase?: { phase: string; title?: string }; + /** 停滞观测态：长时间无事件时为 true，活动恢复/收口清除（可自愈，非终态） */ + stalled?: boolean; } export interface SessionState { @@ -267,6 +269,7 @@ export function applyEvent(state: SessionState, ev: AnyEventEnvelope): SessionSt state: hasPendingBlocking(state, ev.turnId) ? "requires_action" : p.state, startedAt: existing?.startedAt ?? (ev.ts ? Date.parse(ev.ts) || undefined : undefined), phase: existing?.phase, + stalled: existing?.stalled, }); } break; @@ -348,6 +351,16 @@ export function applyEvent(state: SessionState, ev: AnyEventEnvelope): SessionSt if (turn) turn.phase = p.phase === null ? undefined : { phase: p.phase, title: p.title }; break; } + case "_baton_stall_notice": { + // 观测态：只标记/清除对应活跃 turn 的 stalled，不进 timeline、不改 runState、 + // 不留 notice 历史（自愈态无回看价值）。turn 已收口（不在 activeTurns）时忽略—— + // 迟到的 stall 事件不复活已结束的 turn。 + if (ev.turnId) { + const at = state.activeTurns.get(ev.turnId); + if (at) at.stalled = !ev.payload.cleared; + } + break; + } case "_baton_notice": state.notices.push({ ...ev.payload, seq: ev.seq }); state.timeline.push({ type: "notice", id: `n_${ev.seq}` }); diff --git a/src/tui/protocol.ts b/src/tui/protocol.ts index 22c8cc4..4a0d471 100644 --- a/src/tui/protocol.ts +++ b/src/tui/protocol.ts @@ -55,6 +55,13 @@ export function runStatusLabel( state: Pick, turnId?: string, ): string { + // 停滞观测优先于一切阶段/工具文案：卡住的 turn 若还显示 "running command…" 会误导 + // （停滞观测，见 docs/provider-output-lifecycle.md §5）。这里只呈现，不改 runState/busy。 + const stalled = + turnId !== undefined + ? state.activeTurns.get(turnId)?.stalled + : [...state.activeTurns.values()].some((turn) => turn.stalled); + if (stalled) return "no response — may be stuck"; const phase = turnId !== undefined ? state.activeTurns.get(turnId)?.phase diff --git a/tests/capabilities.test.ts b/tests/capabilities.test.ts index 52ecfc0..413783e 100644 --- a/tests/capabilities.test.ts +++ b/tests/capabilities.test.ts @@ -29,6 +29,7 @@ const CAPABILITY_CONTRACT: Record = { "interactions.permission": ["respond"], "interactions.question": ["respond"], "interactions.elicitation": ["respond"], + reconcile: ["reconcile"], }; function capabilityAt(adapter: AgentAdapter, path: string): unknown { diff --git a/tests/reconcile.test.ts b/tests/reconcile.test.ts new file mode 100644 index 0000000..84a700a --- /dev/null +++ b/tests/reconcile.test.ts @@ -0,0 +1,164 @@ +// 对账契约（docs/provider-output-lifecycle.md §5）：turn 停滞时 runtime 主动查 +// provider 真实运行态。idle 裁决 → 自愈 finalize；其余 → 保留提示、不动终态。 +// 未声明 reconcile 能力的 provider 只走停滞提示（不发起对账）。 + +import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +import { mapThreadStatus } from "../src/adapters/codex/adapter.ts"; +import type { + AdapterCapabilities, + AgentAdapter, + EventSink, + OpenOptions, + PromptInput, + PromptReceipt, + ProviderSessionRef, + ReconcileVerdict, +} from "../src/adapters/types.ts"; +import type { AnyEventEnvelope, AnyNewEvent } from "../src/events/types.ts"; +import { BatonSessionRuntime } from "../src/session/runtime.ts"; +import { SessionStore, type SessionHandle } from "../src/store/store.ts"; + +class ReconcilableAdapter implements AgentAdapter { + readonly capabilities: AdapterCapabilities; + sink?: EventSink; + submits: PromptInput[] = []; + reconcileCalls = 0; + + constructor( + private readonly verdict: ReconcileVerdict | (() => Promise), + opts: { declare?: boolean } = {}, + readonly provider: string = "scripted", + ) { + this.capabilities = { prompt: {}, ...(opts.declare === false ? {} : { reconcile: { supported: true } }) }; + } + + async open(_o: OpenOptions, sink: EventSink): Promise { + this.sink = sink; + return { provider: this.provider, providerSessionId: `${this.provider}-ref` }; + } + async submit(_r: ProviderSessionRef, input: PromptInput): Promise { + this.submits.push(input); + return { accepted: true }; + } + emit(ev: AnyNewEvent): void { + this.sink?.(ev); + } + async cancel(): Promise {} + async close(): Promise {} + + async reconcile(): Promise { + this.reconcileCalls++; + return typeof this.verdict === "function" ? this.verdict() : this.verdict; + } +} + +async function until(cond: () => boolean, timeoutMs = 2000): Promise { + const start = Date.now(); + while (!cond()) { + if (Date.now() - start > timeoutMs) throw new Error("condition not met in time"); + await Bun.sleep(2); + } +} + +let root: string; +let session: SessionHandle; + +beforeEach(() => { + root = mkdtempSync(join(tmpdir(), "baton-reconcile-")); + session = new SessionStore(root).createSession({ cwd: "/repo" }); +}); +afterEach(() => { + rmSync(root, { recursive: true, force: true }); +}); + +function makeRuntime(adapter: ReconcilableAdapter) { + return new BatonSessionRuntime({ + session, + mentionBudgetChars: 4096, + createAdapter: () => adapter, + stallThresholdMs: 30, + stallPollMs: 10, + }); +} + +const hasSummary = () => session.readEvents().some((ev: AnyEventEnvelope) => ev.kind === "_baton_turn_summary"); + +describe("mapThreadStatus", () => { + test("codex thread.status wire shapes → verdict", () => { + expect(mapThreadStatus({ type: "idle" })).toBe("idle"); + expect(mapThreadStatus({ type: "active", activeFlags: [] })).toBe("active"); + expect(mapThreadStatus({ type: "active", activeFlags: ["waitingOnApproval"] })).toBe("waiting_approval"); + expect(mapThreadStatus({ type: "active", activeFlags: ["waitingOnUserInput"] })).toBe("waiting_input"); + expect(mapThreadStatus({ type: "notLoaded" })).toBe("unknown"); + expect(mapThreadStatus({ type: "systemError" })).toBe("unknown"); + expect(mapThreadStatus(undefined)).toBe("unknown"); + }); +}); + +describe("runtime reconcile", () => { + test("idle verdict self-heals: turn finalizes without a real terminal", async () => { + const adapter = new ReconcilableAdapter({ state: "idle" }); + const runtime = makeRuntime(adapter); + const outcome = runtime.submit("scripted", [{ type: "text", text: "hi" }]); + await until(() => adapter.submits.length === 1); + + // 不注入任何真实终态；对账 idle 应自愈收口 + expect(await outcome).toBe("completed"); + expect(adapter.reconcileCalls).toBeGreaterThanOrEqual(1); + expect(hasSummary()).toBe(true); + expect(runtime.isBusy).toBe(false); + await runtime.close(); + }); + + test("active verdict keeps waiting: no finalize, stall notice stays", async () => { + const adapter = new ReconcilableAdapter({ state: "active" }); + const runtime = makeRuntime(adapter); + const outcome = runtime.submit("scripted", [{ type: "text", text: "hi" }]); + await until(() => adapter.submits.length === 1); + await until(() => adapter.reconcileCalls >= 2); // 每个停滞 tick 都探 + + expect(hasSummary()).toBe(false); + expect(runtime.isBusy).toBe(true); + expect(session.readEvents().some((ev) => ev.kind === "_baton_stall_notice")).toBe(true); + + adapter.emit({ kind: "state_update", provider: "scripted", turnId: adapter.submits[0]!.turnId, payload: { state: "idle", stopReason: "end_turn" } }); + expect(await outcome).toBe("completed"); + await runtime.close(); + }); + + test("probe failure = unknown: does not finalize", async () => { + const adapter = new ReconcilableAdapter(() => Promise.reject(new Error("boom"))); + const runtime = makeRuntime(adapter); + const outcome = runtime.submit("scripted", [{ type: "text", text: "hi" }]); + await until(() => adapter.submits.length === 1); + await until(() => adapter.reconcileCalls >= 2); + + expect(hasSummary()).toBe(false); + expect(runtime.isBusy).toBe(true); + + adapter.emit({ kind: "state_update", provider: "scripted", turnId: adapter.submits[0]!.turnId, payload: { state: "idle", stopReason: "end_turn" } }); + await outcome; + await runtime.close(); + }); + + test("no reconcile capability → falls back to stall-only, probe never called", async () => { + const adapter = new ReconcilableAdapter({ state: "idle" }, { declare: false }); + const runtime = makeRuntime(adapter); + const outcome = runtime.submit("scripted", [{ type: "text", text: "hi" }]); + await until(() => adapter.submits.length === 1); + await until(() => session.readEvents().some((ev) => ev.kind === "_baton_stall_notice")); + + // 能力未声明：runtime 不发起对账，即便 adapter 实现了 reconcile 也不调 + expect(adapter.reconcileCalls).toBe(0); + expect(hasSummary()).toBe(false); + expect(runtime.isBusy).toBe(true); + + adapter.emit({ kind: "state_update", provider: "scripted", turnId: adapter.submits[0]!.turnId, payload: { state: "idle", stopReason: "end_turn" } }); + await outcome; + await runtime.close(); + }); +}); diff --git a/tests/stall.test.ts b/tests/stall.test.ts new file mode 100644 index 0000000..dc2bc43 --- /dev/null +++ b/tests/stall.test.ts @@ -0,0 +1,148 @@ +// 停滞观测契约（docs/provider-output-lifecycle.md §5）：活跃 turn 长时间无事件时 +// 发一次 `_baton_stall_notice`（可见），活动恢复补 cleared 撤除——但**绝不 finalize**： +// stall 是可自愈观测态，收口仍只由真实终态 / reconcile / 用户 cancel 触发（§4.1）。 + +import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +import type { + AdapterCapabilities, + AgentAdapter, + EventSink, + OpenOptions, + PromptInput, + PromptReceipt, + ProviderSessionRef, +} from "../src/adapters/types.ts"; +import type { AnyEventEnvelope, AnyNewEvent } from "../src/events/types.ts"; +import { BatonSessionRuntime } from "../src/session/runtime.ts"; +import { applyEvent, emptySessionState } from "../src/store/reduce.ts"; +import { SessionStore, type SessionHandle } from "../src/store/store.ts"; + +class ScriptedAdapter implements AgentAdapter { + readonly capabilities: AdapterCapabilities = { prompt: {} }; + sink?: EventSink; + submits: PromptInput[] = []; + + constructor(readonly provider: string = "scripted") {} + + async open(_opts: OpenOptions, sink: EventSink): Promise { + this.sink = sink; + return { provider: this.provider, providerSessionId: `${this.provider}-ref` }; + } + + async submit(_ref: ProviderSessionRef, input: PromptInput): Promise { + this.submits.push(input); + return { accepted: true }; + } + + emit(ev: AnyNewEvent): void { + this.sink?.(ev); + } + + async cancel(_ref: ProviderSessionRef): Promise {} + async close(_ref: ProviderSessionRef): Promise {} +} + +async function until(cond: () => boolean, timeoutMs = 2000): Promise { + const start = Date.now(); + while (!cond()) { + if (Date.now() - start > timeoutMs) throw new Error("condition not met in time"); + await Bun.sleep(2); + } +} + +let root: string; +let session: SessionHandle; + +beforeEach(() => { + root = mkdtempSync(join(tmpdir(), "baton-stall-")); + session = new SessionStore(root).createSession({ cwd: "/repo" }); +}); + +afterEach(() => { + rmSync(root, { recursive: true, force: true }); +}); + +function makeRuntime(adapter: ScriptedAdapter) { + return new BatonSessionRuntime({ + session, + mentionBudgetChars: 4096, + createAdapter: () => adapter, + stallThresholdMs: 30, + stallPollMs: 10, + }); +} + +function stallNotices(events: AnyEventEnvelope[]) { + return events.filter((ev) => ev.kind === "_baton_stall_notice"); +} + +describe("stall observation", () => { + test("silent active turn emits one stall notice, but never finalizes", async () => { + const adapter = new ScriptedAdapter(); + const runtime = makeRuntime(adapter); + const outcome = runtime.submit("scripted", [{ type: "text", text: "hi" }]); + await until(() => adapter.submits.length === 1); + + // 不注入任何事件，等待超过 threshold + 一次 poll + await until(() => stallNotices(session.readEvents()).length >= 1); + + const notices = stallNotices(session.readEvents()); + expect(notices).toHaveLength(1); // 去重：只发一次 + expect((notices[0]!.payload as { cleared?: boolean }).cleared).toBeFalsy(); + expect((notices[0]!.payload as { stalledMs: number }).stalledMs).toBeGreaterThanOrEqual(30); + + // 绝不 finalize：turn 仍活跃，无 idle、无 turn summary + expect(runtime.isBusy).toBe(true); + expect(session.readEvents().some((ev) => ev.kind === "_baton_turn_summary")).toBe(false); + + // 收口仍走真实终态 + adapter.emit({ kind: "state_update", provider: "scripted", turnId: adapter.submits[0]!.turnId, payload: { state: "idle", stopReason: "end_turn" } }); + expect(await outcome).toBe("completed"); + await runtime.close(); + }); + + test("activity after stall emits a cleared notice", async () => { + const adapter = new ScriptedAdapter(); + const runtime = makeRuntime(adapter); + const outcome = runtime.submit("scripted", [{ type: "text", text: "hi" }]); + await until(() => adapter.submits.length === 1); + const turnId = adapter.submits[0]!.turnId; + + await until(() => stallNotices(session.readEvents()).length >= 1); + + // 活动恢复：任何命中该 turn 的事件都刷新进展时钟并补 cleared + adapter.emit({ kind: "agent_message_chunk", provider: "scripted", turnId, payload: { messageId: "m1", content: { type: "text", text: "back" } } }); + + await until(() => stallNotices(session.readEvents()).some((ev) => (ev.payload as { cleared?: boolean }).cleared)); + const cleared = stallNotices(session.readEvents()).find((ev) => (ev.payload as { cleared?: boolean }).cleared); + expect(cleared).toBeDefined(); + + adapter.emit({ kind: "state_update", provider: "scripted", turnId, payload: { state: "idle", stopReason: "end_turn" } }); + await outcome; + await runtime.close(); + }); + + test("reduce marks the active turn stalled, then clears it", () => { + let state = emptySessionState(); + const env = (kind: K, payload: unknown, turnId?: string, seq = ++mseq) => + ({ v: 1, ts: new Date(0).toISOString(), seq, batonSessionId: "bs", provider: "p", kind, payload, turnId }) as unknown as AnyEventEnvelope; + + state = applyEvent(state, env("state_update", { state: "running" }, "t1")); + state = applyEvent(state, env("_baton_stall_notice", { stalledMs: 999 }, "t1")); + expect(state.activeTurns.get("t1")?.stalled).toBe(true); + + state = applyEvent(state, env("_baton_stall_notice", { stalledMs: 0, cleared: true }, "t1")); + expect(state.activeTurns.get("t1")?.stalled).toBe(false); + + // idle 收口后，迟到的 stall 事件不复活已结束的 turn + state = applyEvent(state, env("state_update", { state: "idle", stopReason: "end_turn" }, "t1")); + state = applyEvent(state, env("_baton_stall_notice", { stalledMs: 999 }, "t1")); + expect(state.activeTurns.has("t1")).toBe(false); + }); +}); + +let mseq = 0; diff --git a/tests/tui-protocol.test.ts b/tests/tui-protocol.test.ts index a082a30..6b3d560 100644 --- a/tests/tui-protocol.test.ts +++ b/tests/tui-protocol.test.ts @@ -420,6 +420,17 @@ describe("runStatusLabel", () => { expect(runStatusLabel(withPhase("t1", { phase: "warming" }), "t1")).toBe("warming…"); }); + test("stalled overrides phase/tool: a stuck turn does not show a misleading activity", () => { + const stalled = (turnId: string) => ({ + ...base, + activeTurns: new Map([ + [turnId, { turnId, origin: "user" as const, state: "running" as const, phase: { phase: "compacting" }, stalled: true }], + ]), + }); + expect(runStatusLabel(stalled("t1"), "t1")).toBe("no response — may be stuck"); + expect(runStatusLabel(stalled("t1"))).toBe("no response — may be stuck"); // turnId 缺省时任一 stalled 即命中 + }); + test("shows the current tool activity instead of generic thinking", () => { const toolCalls = new Map([ [