From 954ce1a998a672b0ad545427cc385f4946718aec Mon Sep 17 00:00:00 2001 From: William Wang Date: Mon, 31 Aug 2026 11:11:04 +0800 Subject: [PATCH 01/14] feat: per-project prompt history recall --- src/repl/App.tsx | 18 ++++++ src/repl/history.ts | 73 +++++++++++++++++++++++ src/repl/run.ts | 63 +++++++++++++++++++- tests/repl-history.test.ts | 115 +++++++++++++++++++++++++++++++++++++ 4 files changed, 266 insertions(+), 3 deletions(-) create mode 100644 src/repl/history.ts create mode 100644 tests/repl-history.test.ts diff --git a/src/repl/App.tsx b/src/repl/App.tsx index 0c7d2d7..489df74 100644 --- a/src/repl/App.tsx +++ b/src/repl/App.tsx @@ -132,6 +132,9 @@ export interface AppProps { onAnswerQuestion: (answer: QuestionAnswer | null) => void; /** null = dismiss the picker without resuming. */ onPickSession: (sessionId: string | null) => void; + /** Prompt-history recall (ADR-0008); run.ts no-ops when nothing to walk. */ + onHistoryUp: () => void; + onHistoryDown: () => void; onExit: () => void; } @@ -715,6 +718,21 @@ export function App(props: AppProps): ReactElement { handlePermissionKey(inputChar, key); return; } + // Prompt-history recall: the completion menu keeps its claim on ↑/↓ while + // open (its own handler consumes them); with it closed — and no picker + // mounted — the arrows walk submitted prompts (ADR-0008). InputLine's + // handler also sees these keys but only nudges its (reset) selection. + if ( + (key.upArrow || key.downArrow) && + !props.isMenuOpen() && + !sessionPick && + !question && + !permission + ) { + if (key.upArrow) props.onHistoryUp(); + else props.onHistoryDown(); + return; + } // Esc interrupts the running turn — with or without queued follow-ups; // the stop handler drains the queue, so a next queued message starts at // once. With the completion menu open, the menu takes this esc instead diff --git a/src/repl/history.ts b/src/repl/history.ts new file mode 100644 index 0000000..c913028 --- /dev/null +++ b/src/repl/history.ts @@ -0,0 +1,73 @@ +/** + * Per-project prompt history (ADR-0008): JSONL at + * ~/.zcode/acp/repl-history/.jsonl, one submitted line per JSON + * string. Pure fs-in/file-fs-out helpers — the entry list itself lives in + * run.ts's external store, so Ctrl-L repaints and session swaps never lose + * it. The reader is tolerant: malformed lines are skipped, never fatal. + */ + +import { createHash } from "node:crypto"; +import { existsSync, mkdirSync, readFileSync, writeFileSync } from "node:fs"; +import { homedir } from "node:os"; +import { dirname, join } from "node:path"; + +/** Entries kept per project; enforced on load (truncate-rewrite) and save. */ +export const HISTORY_MAX = 500; + +/** Absolute path of this project's history file. */ +export function historyPath(cwd: string): string { + const digest = createHash("sha1").update(cwd).digest("hex"); + return join(homedir(), ".zcode", "acp", "repl-history", `${digest}.jsonl`); +} + +/** + * Read and parse a history file, oldest first. Bad lines are dropped; + * when the file grew past HISTORY_MAX it is rewritten with the newest tail + * so the cap holds without a separate maintenance pass. + */ +export function loadHistory(filePath: string, max = HISTORY_MAX): string[] { + if (!existsSync(filePath)) return []; + let entries: string[] = []; + try { + for (const line of readFileSync(filePath, "utf8").split("\n")) { + if (!line.trim()) continue; + try { + const parsed: unknown = JSON.parse(line); + if (typeof parsed === "string" && parsed.trim()) entries.push(parsed); + } catch { + // torn write / manual edit — skip the line + } + } + } catch { + return []; // unreadable file: start fresh rather than crash the REPL + } + if (entries.length > max) { + entries = entries.slice(-max); + saveHistory(filePath, entries, max); + } + return entries; +} + +/** Persist the whole list (≤ max entries), creating the directory once. */ +export function saveHistory(filePath: string, entries: readonly string[], max = HISTORY_MAX): void { + const tail = entries.slice(-max); + try { + mkdirSync(dirname(filePath), { recursive: true }); + writeFileSync( + filePath, + tail.map((line) => JSON.stringify(line)).join("\n") + (tail.length ? "\n" : ""), + ); + } catch { + // best-effort persistence: an unwritable home dir must not break submits + } +} + +/** + * Pure consecutive-duplicate suppression: resubmitting the recalled newest + * entry (or mashing enter on the same line) must not spam the file. + */ +export function pushHistory(entries: readonly string[], text: string): string[] { + if (!text.trim()) return [...entries]; + if (entries[entries.length - 1] === text) return [...entries]; + return [...entries, text]; +} diff --git a/src/repl/run.ts b/src/repl/run.ts index 951fc7b..cbf839b 100644 --- a/src/repl/run.ts +++ b/src/repl/run.ts @@ -47,7 +47,8 @@ import { type SessionSummary, type TurnState, } from "./model.js"; -import { createLineEditor, type LineEditor } from "./input-buffer.js"; +import { historyPath, loadHistory, pushHistory, saveHistory } from "./history.js"; +import { createLineEditor, replaceText, type LineEditor } from "./input-buffer.js"; export async function runRepl(): Promise { // Crash containment: an unexpected throw is SURFACED, never fatal — it @@ -175,6 +176,50 @@ export async function runRepl(): Promise { // snapshots, so draft text must live out here to survive rerenders. let editor: LineEditor = createLineEditor(); + // --- prompt history (ADR-0008) --- + // Entries load once at startup; `historyIdx` points at the recalled entry + // (-1 = live draft). ↑ stashes the draft before walking back; ↓ past the + // newest entry restores it. State out here so recall survives repaints. + const historyFile = historyPath(process.cwd()); + let history: string[] = []; + let historyIdx = -1; + let historyDraft: string | null = null; + try { + history = loadHistory(historyFile); + } catch { + // loadHistory is already best-effort; this guard keeps even a throw from + // its own write-back path from killing startup. + } + + /** Recall one entry older; stashes the live draft on the first step. */ + function historyUp(): void { + if (history.length === 0 || exited) return; + if (historyIdx === -1) { + historyDraft = editor.text; + historyIdx = history.length - 1; + } else if (historyIdx > 0) { + historyIdx--; + } else { + return; + } + editor = replaceText(history[historyIdx]!); + rerender(); + } + + /** Walk one entry newer; past the newest, restore the stashed draft. */ + function historyDown(): void { + if (historyIdx === -1 || exited) return; + if (historyIdx < history.length - 1) { + historyIdx++; + editor = replaceText(history[historyIdx]!); + } else { + historyIdx = -1; + editor = replaceText(historyDraft ?? ""); + historyDraft = null; + } + rerender(); + } + // --- completion-menu visibility mirror --- // InputLine publishes whether its menu is showing (plain var, no React // state); the app-level key handler reads it so an open menu takes the @@ -244,6 +289,8 @@ export async function runRepl(): Promise { resolve?.(answer); }, onPickSession: (sid) => sessionPickResolver?.(sid), + onHistoryUp: historyUp, + onHistoryDown: historyDown, onExit: () => cleanup(0), }; } @@ -488,7 +535,17 @@ export async function runRepl(): Promise { } })(); - async function onSubmit(text: string): Promise { + async function onSubmit(text: string, viaQueue = false): Promise { + // Every submit is history (slash commands included, verbatim) — recall + // exists precisely for command incantations. Recorded ONCE, at first + // submission: the queue drain re-enters through here (viaQueue) and its + // entries were already recorded when they were queued. + if (!viaQueue) { + history = pushHistory(history, text); + saveHistory(historyFile, history); + historyIdx = -1; + historyDraft = null; + } const cmd = parseCommand(text); if (cmd === "exit") { cleanup(0); @@ -556,7 +613,7 @@ export async function runRepl(): Promise { // Route through onSubmit, not startTurn directly: queued entries still // go through command parsing (a queued "/help" must render locally, // "/exit" must exit — never reach the bridge as a literal prompt). - void onSubmit(next); + void onSubmit(next, true); } function onCancelTurn(): void { diff --git a/tests/repl-history.test.ts b/tests/repl-history.test.ts new file mode 100644 index 0000000..0c33f82 --- /dev/null +++ b/tests/repl-history.test.ts @@ -0,0 +1,115 @@ +/** + * history.ts tests — per-project prompt history (ADR-0008). + * + * fs is mocked with a Map-based fake filesystem (repo pattern): loadHistory / + * saveHistory round-trip through it, including the over-cap truncate-rewrite + * and malformed-line tolerance. + */ + +import { createHash } from "node:crypto"; +import { join } from "node:path"; + +import { beforeEach, describe, expect, it, vi } from "vitest"; + +vi.mock("node:os", () => ({ homedir: () => "/home/tester" })); + +const mockFiles = new Map(); +const writes: Array<{ path: string; data: string }> = []; + +vi.mock("node:fs", async () => { + const actual = await vi.importActual("node:fs"); + return { + ...actual, + existsSync: (p: string) => mockFiles.has(p), + mkdirSync: () => undefined, + readFileSync: (p: string) => { + if (mockFiles.has(p)) return mockFiles.get(p)!; + throw new Error("ENOENT"); + }, + writeFileSync: (p: string, data: string) => { + writes.push({ path: p, data }); + mockFiles.set(p, data); + }, + }; +}); + +import { + HISTORY_MAX, + historyPath, + loadHistory, + pushHistory, + saveHistory, +} from "../src/repl/history.js"; + +const FILE = join("/home/tester", ".zcode", "acp", "repl-history", `${sha1("/proj")}.jsonl`); + +function sha1(s: string): string { + return createHash("sha1").update(s).digest("hex"); +} + +beforeEach(() => { + mockFiles.clear(); + writes.length = 0; +}); + +describe("historyPath", () => { + it("hashes the cwd into a stable file under ~/.zcode/acp/repl-history", () => { + const p = historyPath("/proj"); + expect(p.startsWith(join("/home/tester", ".zcode", "acp", "repl-history") + "/")).toBe(true); + expect(p.endsWith(`${sha1("/proj")}.jsonl`)).toBe(true); + expect(historyPath("/other")).not.toBe(p); + }); +}); + +describe("pushHistory", () => { + it("appends and suppresses only consecutive duplicates", () => { + let entries = pushHistory([], "one"); + entries = pushHistory(entries, "one"); // consecutive dup — dropped + entries = pushHistory(entries, "two"); + entries = pushHistory(entries, "one"); // non-consecutive — kept + expect(entries).toEqual(["one", "two", "one"]); + }); + + it("rejects blank submissions", () => { + expect(pushHistory(["kept"], " ")).toEqual(["kept"]); + expect(pushHistory(["kept"], "")).toEqual(["kept"]); + }); +}); + +describe("loadHistory", () => { + it("returns [] for a missing or unreadable file", () => { + expect(loadHistory(FILE)).toEqual([]); + }); + + it("parses JSONL in order and skips bad lines", () => { + mockFiles.set( + FILE, + JSON.stringify("first") + "\nnot json\n" + JSON.stringify("second") + "\n\n42\n", + ); + expect(loadHistory(FILE)).toEqual(["first", "second"]); + }); + + it("truncates past the cap and rewrites the newest tail", () => { + const all = Array.from({ length: HISTORY_MAX + 10 }, (_, i) => `entry-${i}`); + mockFiles.set(FILE, all.map((e) => JSON.stringify(e)).join("\n") + "\n"); + const loaded = loadHistory(FILE); + expect(loaded).toHaveLength(HISTORY_MAX); + expect(loaded[0]).toBe("entry-10"); + expect(writes.some((w) => w.path === FILE && w.data.includes("entry-10"))).toBe(true); + expect(writes.some((w) => w.data.includes("entry-0\n"))).toBe(false); + }); +}); + +describe("saveHistory", () => { + it("writes JSONL and enforces the cap", () => { + saveHistory(FILE, ["a", "b"]); + expect(mockFiles.get(FILE)).toBe(JSON.stringify("a") + "\n" + JSON.stringify("b") + "\n"); + saveHistory( + FILE, + Array.from({ length: HISTORY_MAX + 1 }, (_, i) => `e${i}`), + ); + const stored = (mockFiles.get(FILE) ?? "").split("\n").filter(Boolean); + expect(stored).toHaveLength(HISTORY_MAX); + expect(JSON.parse(stored[0]!)).toBe("e1"); + }); +}); From 0de41923779b8c9eaf5651f834c0febd9400d4bf Mon Sep 17 00:00:00 2001 From: William Wang Date: Mon, 31 Aug 2026 11:11:09 +0800 Subject: [PATCH 02/14] fix: fold pasted newlines to keep one prompt per paste --- src/repl/App.tsx | 35 ++++++++++++---- src/repl/input-buffer.ts | 21 ++++++++++ src/repl/run.ts | 10 +++++ tests/repl-input-paste.test.ts | 74 ++++++++++++++++++++++++++++++++++ 4 files changed, 133 insertions(+), 7 deletions(-) create mode 100644 tests/repl-input-paste.test.ts diff --git a/src/repl/App.tsx b/src/repl/App.tsx index 489df74..bd2bb9e 100644 --- a/src/repl/App.tsx +++ b/src/repl/App.tsx @@ -8,7 +8,7 @@ * input area with an arrow-key picker. */ -import { Box, Static, Text, useInput, type Key } from "ink"; +import { Box, Static, Text, useInput, usePaste, type Key } from "ink"; import Spinner from "ink-spinner"; import { Chalk } from "chalk"; import { useCallback, useEffect, useRef, useState, type ReactElement } from "react"; @@ -42,6 +42,7 @@ import { createLineEditor, ctrlChord, deleteAtCaret, + foldPasteChunk, insertAtCaret, planChunkOps, replaceText, @@ -470,19 +471,39 @@ function InputLine({ } }; + // Dedicated paste channel (ADR-0009): ink tokenizes ?2004 bracketed pastes + // and delivers them here — markers stripped, newlines intact — and NEVER + // forwards them to useInput while this hook is mounted. ink also arms and + // disarms bracketed-paste mode with this hook's lifecycle. + usePaste((text) => { + applyEdit((cur) => insertAtCaret(cur, foldPasteChunk(text))); + }); + useInput((inputChar, key) => { - // Coalesced printable chunk (paste, rapid keys) — batched via - // planChunkOps: printable runs apply as ONE editor op, semantic bytes - // stay per-character. Escape-sequence chunks (arrow keys flushed - // together) keep ink's parsed flags instead. - if (inputChar && inputChar.length > 1 && !inputChar.includes("\x1b") && !key.ctrl) { - for (const op of planChunkOps(inputChar)) { + const applyOps = (chunk: string): void => { + for (const op of planChunkOps(chunk)) { if (op.kind === "insert") { applyEdit((cur) => insertAtCaret(cur, op.text)); } else { handleChar(op.ch, derivedKey(op.ch), null); } } + }; + // Legacy-terminal fallback (no ?2004): unwrapped pastes arrive as raw + // coalesced chunks. Require a real newline to fold — lone \r bytes stay + // semantic, so coalesced keystroke bursts that happen to carry an Enter + // ("x\r" from fast typing, a held-down return) keep their submit + // semantics exactly as they had before ADR-0009. + if (inputChar && inputChar.length > 1 && inputChar.includes("\n") && !key.ctrl) { + applyEdit((cur) => insertAtCaret(cur, foldPasteChunk(inputChar))); + return; + } + // Coalesced printable chunk (paste, rapid keys) — batched via + // planChunkOps: printable runs apply as ONE editor op, semantic bytes + // stay per-character. Escape-sequence chunks (arrow keys flushed + // together) keep ink's parsed flags instead. + if (inputChar && inputChar.length > 1 && !inputChar.includes("\x1b") && !key.ctrl) { + applyOps(inputChar); return; } handleChar( diff --git a/src/repl/input-buffer.ts b/src/repl/input-buffer.ts index 37cd2b9..39cb94b 100644 --- a/src/repl/input-buffer.ts +++ b/src/repl/input-buffer.ts @@ -96,6 +96,27 @@ export function replaceText(text: string): LineEditor { return { text, caret: toArray(text).length }; } +/** + * Fold a pasted chunk into single-line text (ADR-0009). Bracketed-paste + * wrappers are stripped, CRLF/CR/LF runs and tabs collapse to single + * spaces, and remaining control bytes are sanitized away — a newline inside + * a paste must never keep its Enter semantics (pre-fix pastes fired + * line-by-line as separate prompts). Printable text, emoji, CJK pass + * through; the prompt-size cap still bounds the result. + */ +export function foldPasteChunk(chunk: string): string { + // replaceAll with string needles: the wrappers contain ESC, which ESLint's + // no-control-regex (rightly) refuses inside a regex literal. + const stripped = chunk.replaceAll("\x1b[200~", "").replaceAll("\x1b[201~", ""); + const folded = stripped.replace(/\r\n?/g, "\n").replace(/\n+/g, " ").replace(/\t/g, " "); + // Cap by CODE POINTS (Array.from), matching insertAtCaret's quota — + // String.slice counts UTF-16 units and would cleave a surrogate pair at + // the boundary on emoji-dense pastes. + const clean = Array.from(sanitizeInputChunk(folded)).slice(0, MAX_PROMPT_CHARS).join(""); + // A whitespace-only paste (blank lines, stray wrappers) inserts nothing. + return clean.trim() ? clean : ""; +} + export function insertAtCaret(editor: LineEditor, str: string): LineEditor { if (!str) return editor; // Silent tail-drop past the cap — the alternative is an unusable frame. diff --git a/src/repl/run.ts b/src/repl/run.ts index cbf839b..6624600 100644 --- a/src/repl/run.ts +++ b/src/repl/run.ts @@ -71,6 +71,14 @@ export async function runRepl(): Promise { warn(`repl absorbed an error (${crashTimes.length}/${CRASH_LIMIT} recent): ${detail}`); if (crashTimes.length >= CRASH_LIMIT) { warn("repl: too many errors in a row — UI is likely broken, shutting down"); + // Unmount so ink's stdin cleanup effects run (raw mode, bracketed + // paste): process.exit alone would leave the tty wedged. The tree may + // be exactly what's broken — guard the unmount. + try { + ink.unmount(); + } catch { + // ignore + } try { child.stdin?.end(); } catch { @@ -254,6 +262,8 @@ export async function runRepl(): Promise { quotaTimer = setInterval(() => void refreshQuota(), QUOTA_TTL_MS); quotaTimer.unref(); + // Bracketed paste (?2004) is armed by ink's usePaste hook inside InputLine + // and disarmed with its unmount — no manual terminal-mode management here. const renderOpts = { exitOnCtrlC: false }; let ink = render(createElement(App, snapshot()), renderOpts); function snapshot(): AppProps { diff --git a/tests/repl-input-paste.test.ts b/tests/repl-input-paste.test.ts new file mode 100644 index 0000000..de178ad --- /dev/null +++ b/tests/repl-input-paste.test.ts @@ -0,0 +1,74 @@ +/** + * input-buffer paste tests — foldPasteChunk (ADR-0009). + * + * A newline inside a pasted chunk must never keep its Enter semantics: + * pre-fix, planChunkOps replayed \r/\n per-character and every paragraph of + * a multi-line paste fired as its own prompt. These cases pin the folding + * rules AND the boundary (planChunkOps still treats a lone \r as semantic, + * which is what keeps a real Enter keypress submitting). + */ + +import { describe, expect, it } from "vitest"; + +import { foldPasteChunk, MAX_PROMPT_CHARS, planChunkOps } from "../src/repl/input-buffer.js"; + +describe("foldPasteChunk", () => { + it("strips bracketed-paste wrappers", () => { + expect(foldPasteChunk("\x1b[200~hello world\x1b[201~")).toBe("hello world"); + }); + + it("folds CRLF, LF and CR to single spaces", () => { + expect(foldPasteChunk("para one\r\npara two")).toBe("para one para two"); + expect(foldPasteChunk("a\nb")).toBe("a b"); + expect(foldPasteChunk("a\rb")).toBe("a b"); + }); + + it("collapses newline runs (blank lines) to one space", () => { + expect(foldPasteChunk("a\n\n\nb")).toBe("a b"); + }); + + it("folds tabs, which would otherwise trigger completion", () => { + expect(foldPasteChunk("key\tvalue")).toBe("key value"); + }); + + it("keeps emoji, CJK and printable text untouched", () => { + expect(foldPasteChunk("中文 🎉 paste")).toBe("中文 🎉 paste"); + }); + + it("drops stray control and escape bytes", () => { + expect(foldPasteChunk("a\x1bb\x00c\x7fd")).toBe("abcd"); + }); + + it("returns empty for a whitespace-only paste", () => { + expect(foldPasteChunk("\x1b[200~\r\n\r\n\x1b[201~")).toBe(""); + }); + + it("caps the folded result at MAX_PROMPT_CHARS", () => { + const huge = "\x1b[200~" + "x".repeat(MAX_PROMPT_CHARS + 1000) + "\x1b[201~"; + expect(foldPasteChunk(huge).length).toBe(MAX_PROMPT_CHARS); + }); + + it("caps by code points — never splits a surrogate pair at the boundary", () => { + const chunk = "x".repeat(MAX_PROMPT_CHARS - 1) + "😀y"; + const out = foldPasteChunk(chunk); + expect(Array.from(out).length).toBe(MAX_PROMPT_CHARS); + expect(out.endsWith("😀")).toBe(true); + }); +}); + +describe("planChunkOps boundary vs paste folding", () => { + it("still replays a lone semantic Enter per-character (real keypress submits)", () => { + const ops = planChunkOps("hi\r"); + expect(ops).toEqual([ + { kind: "insert", text: "hi" }, + { kind: "char", ch: "\r" }, + ]); + }); + + it("never emits newline semantics — callers fold those chunks before routing", () => { + // Pin the contract the App relies on: wrapped paste text fed through + // foldPasteChunk first yields insert-only ops. + const folded = foldPasteChunk("\x1b[200~a\r\nb\x1b[201~"); + expect(planChunkOps(folded)).toEqual([{ kind: "insert", text: "a b" }]); + }); +}); From 150aa2e4c5584a483cfe8f638e15f6107b3b14be Mon Sep 17 00:00:00 2001 From: William Wang Date: Mon, 31 Aug 2026 11:11:16 +0800 Subject: [PATCH 03/14] feat: REPL /new swaps in a fresh session --- CHANGELOG.md | 17 ++++++ README.md | 9 ++- src/repl/model.ts | 8 ++- src/repl/run.ts | 121 +++++++++++++++++++++++++++++++++++---- tests/repl-model.test.ts | 17 +++++- 5 files changed, 155 insertions(+), 17 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 57babea..0c6c803 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,23 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- REPL prompt history: every submit is recorded per project + (`~/.zcode/acp/repl-history/.jsonl`, newest 500 kept, runs of + duplicates collapsed) and recalled across restarts with `↑`/`↓` while the + completion menu is closed — the first `↑` stashes the live draft and `↓` + past the newest entry restores it. +- Pasted text is folded to a single line before it reaches the prompt: + bracketed-paste mode (`?2004`) is armed so ink delivers pastes as one + chunk, and newlines/tabs inside them (or any multi-character chunk + carrying a newline, for terminals without `?2004`) become single spaces. + Previously every newline in a paste submitted mid-paste, firing a + multi-paragraph paste line-by-line as separate prompts. +- REPL `/new` starts a fresh session without leaving the terminal: the live + session is swapped client-side for a new `session/new` placeholder + (config selects reseeded from the response), a divider note marks the + boundary, and the prompt draft is cleared. A running turn refuses it + (`esc` interrupts first); it is registered as a one-shot command, so + picking it in the completion menu executes immediately. - Interactive REPL (bare `zcode-acp`): an Ink terminal chat over the same bridge the editor uses, including slash-command completion with an interactive menu, a caret-aware prompt line (arrows/Ctrl-B/F/A/E/U), diff --git a/README.md b/README.md index 1cac887..5184d9b 100644 --- a/README.md +++ b/README.md @@ -246,7 +246,14 @@ confirmation. Argument-free commands (`/exit`, `/help`, `/sessions`, (skills, plugins) only fills the line, since those usually expect arguments. The arg-less forms still print a static listing over the same slash-command path the editor uses. `/help` lists every command the bridge advertises, -including plugin commands. +including plugin commands. `/new` swaps in a fresh session without leaving +the terminal (the old conversation stays in `/sessions` and in scrollback). + +Submitted prompts are history: `↑`/`↓` (with the completion menu closed) +recall them per project across restarts — the first `↑` stashes the draft +and `↓` past the newest entry restores it. Pasted text folds to a single +line (newlines and tabs become spaces), so a multi-paragraph paste lands in +the box as one prompt instead of firing line-by-line. Unexpected internal errors never take the REPL down silently: they print to stderr and surface as an `-- error absorbed: …` note in the transcript while diff --git a/src/repl/model.ts b/src/repl/model.ts index 7a37a3f..48f8275 100644 --- a/src/repl/model.ts +++ b/src/repl/model.ts @@ -299,12 +299,13 @@ export function finishTurn(state: TurnState, stopReason?: string): ReplEntry[] { } /** REPL meta-commands. Everything else is a prompt. */ -export type ReplCommand = "exit" | "sessions" | null; +export type ReplCommand = "exit" | "sessions" | "new" | null; export function parseCommand(text: string): ReplCommand { const t = text.trim(); if (t === "/exit" || t === "/quit" || t === "/q") return "exit"; if (t === "/sessions") return "sessions"; + if (t === "/new") return "new"; return null; } @@ -369,6 +370,7 @@ export function createReplStatus(): ReplStatus { /** Shared by both lists below so /help and /exit keep one description each. */ const HELP_COMMAND: CommandInfo = { name: "help", description: "list commands (REPL-local)" }; +const NEW_COMMAND: CommandInfo = { name: "new", description: "start a fresh session" }; const EXIT_COMMAND: CommandInfo = { name: "exit", description: "quit the REPL" }; /** Commands shown before the bridge's first `available_commands_update`. */ @@ -380,6 +382,7 @@ export const FALLBACK_COMMANDS: CommandInfo[] = [ { name: "compact", description: "compact conversation context" }, { name: "mcp", description: "list configured MCP servers" }, { name: "quota", description: "show plan usage card" }, + NEW_COMMAND, EXIT_COMMAND, ]; @@ -387,6 +390,7 @@ export const FALLBACK_COMMANDS: CommandInfo[] = [ const LOCAL_COMMANDS: CommandInfo[] = [ HELP_COMMAND, { name: "sessions", description: "list and resume project sessions" }, + NEW_COMMAND, EXIT_COMMAND, ]; @@ -565,7 +569,7 @@ export function isConfigArgumentMenu(value: string): boolean { * fill semantics because its bare form usually expects an argument, and * sending must stay the user's explicit act. */ -const ONE_SHOT_COMMANDS = new Set(["help", "sessions", "exit", "compact", "mcp", "quota"]); +const ONE_SHOT_COMMANDS = new Set(["help", "sessions", "new", "exit", "compact", "mcp", "quota"]); /** Whether the "/"-prefixed single-token line executes immediately when picked. */ export function isOneShotCommandValue(value: string): boolean { diff --git a/src/repl/run.ts b/src/repl/run.ts index 6624600..ec6bbf8 100644 --- a/src/repl/run.ts +++ b/src/repl/run.ts @@ -339,6 +339,11 @@ export async function runRepl(): Promise { // --- /sessions bridging: list → picker → resume --- let sessionPickResolver: ((sessionId: string | null) => void) | null = null; + // Bumped by every session swap; async continuations compare their captured + // generation after each await and bail when another swap went first — + // otherwise a slow session/load could reseed status into a FRESH session. + let swapGen = 0; + // --- ACP client connection --- const stream = acp.ndJsonStream( Writable.toWeb(child.stdin! as Writable), @@ -566,6 +571,10 @@ export async function runRepl(): Promise { void openSessionPicker(); return; } + if (cmd === "new") { + void startFreshSession(); + return; + } entries = [...entries, { kind: "user", text }]; // REPL-local commands (help / arg-less listing forms) render here and // never reach the bridge; everything else is a prompt (slash interception @@ -706,6 +715,7 @@ export async function runRepl(): Promise { * routing (`buildSession().start()` only covers session/new). */ async function resumeInto(picked: SessionSummary): Promise { + const gen = ++swapGen; const loaded = ( cx as unknown as { attachSession(response: { sessionId: string }): ActiveSession; @@ -735,9 +745,11 @@ export async function runRepl(): Promise { configOptions?: acp.SessionConfigOption[] | null; replayMeta?: { replayedMessages?: number; totalMessages?: number; hasMore?: boolean }; }; + if (gen !== swapGen) return; // another swap went first — this load is stale status = seedStatusFromNewSession(status, resp); resumeMeta = resp.replayMeta ?? null; } catch (err) { + if (gen !== swapGen) return; // stale — another swap owns the state now replayMode = false; replayTurn = null; entries = [ @@ -765,6 +777,102 @@ export async function runRepl(): Promise { rerender(); } + /** Fresh welcome panel entry; reused by startup and `/new`. */ + function welcomeEntry(): ReplEntry { + return { + kind: "welcome", + info: { + version: AGENT_INFO.version, + cwd: process.cwd(), + model: selectLabel(status.model), + mode: selectLabel(status.mode), + thought: selectLabel(status.thought), + }, + }; + } + + /** + * `/new` (ADR-0010): swap the live session for a fresh backend-created one, + * strictly client-side — routing the command to the backend's slash + * interception would rotate the session id out from under the update pump. + * Reuses the session/new bootstrap; the swap refuses while a turn runs or + * startup is still in flight (same preempt discipline as /sessions). + */ + async function startFreshSession(): Promise { + if (busy || turnActive) { + entries = [ + ...entries, + { kind: "user", text: "/new" }, + { + kind: "note", + text: turnActive + ? "a turn is running — esc interrupts it first" + : "still starting up — try /new again in a moment", + }, + ]; + rerender(); + return; + } + const gen = ++swapGen; + try { + const fresh = await cx.buildSession(process.cwd()).start(); + if (gen !== swapGen) { + fresh.dispose(); // another swap went first — this placeholder is junk + return; + } + if (turnActive) { + // A turn raced into the old session during the session/new roundtrip + // (a local submit or a remote client). Swapping now would orphan it: + // disposing the session takes its update pump down and the + // completion event is filtered by session id, so turnActive would + // stick forever and wedge the REPL. Keep the current session. + fresh.dispose(); + entries = [ + ...entries, + { kind: "user", text: "/new" }, + { + kind: "note", + text: "a turn started while /new was swapping — staying in the current session", + }, + ]; + rerender(); + return; + } + activeSession?.dispose(); + activeSession = fresh; + status = seedStatusFromNewSession(status, fresh.newSessionResponse); + // Any in-flight /sessions replay is dead: its pump read fails into the + // catch/re-arm path, and its stale continuations bail on swapGen. + replayMode = false; + replayTurn = null; + loadSettled = true; + // Native-scrollback model: entries are APPEND-ONLY — ink has + // already printed everything before its print cursor, so shrinking the + // array would silently drop the divider. The old conversation stays in + // the terminal's own history (that's the model's whole point). + entries = [ + ...entries, + { kind: "user", text: "/new" }, + { + kind: "note", + text: "── new session started — the previous conversation stays in /sessions ──", + }, + ]; + editor = createLineEditor(); + rerender(); + } catch (err) { + entries = [ + ...entries, + { kind: "user", text: "/new" }, + { + kind: "note", + text: `failed to start a new session: ${err instanceof Error ? err.message : String(err)}`, + }, + ]; + rerender(); + } + } + // Terminal resize: the dynamic footer re-wraps at the new width via a // plain rerender. Already-printed scrollback keeps its old wrapping — // that's exactly how native-history CLIs (Claude Code et al.) behave. @@ -832,17 +940,6 @@ export async function runRepl(): Promise { // Welcome panel as the first transcript entry — branding, session info, // seeded config, and key hints; pushed into scrollback by the first prompt. - entries = [ - { - kind: "welcome", - info: { - version: AGENT_INFO.version, - cwd: process.cwd(), - model: selectLabel(status.model), - mode: selectLabel(status.mode), - thought: selectLabel(status.thought), - }, - }, - ]; + entries = [welcomeEntry()]; rerender(); } diff --git a/tests/repl-model.test.ts b/tests/repl-model.test.ts index 0bea230..f6680d9 100644 --- a/tests/repl-model.test.ts +++ b/tests/repl-model.test.ts @@ -122,6 +122,11 @@ describe("parseCommand", () => { expect(parseCommand("/session")).toBe(null); expect(parseCommand("/sessions now")).toBe(null); }); + + it("recognizes the fresh-session command (no arguments)", () => { + expect(parseCommand("/new")).toBe("new"); + expect(parseCommand("/new x")).toBe(null); + }); }); describe("relativeTime", () => { @@ -312,6 +317,7 @@ describe("isOneShotCommandValue", () => { expect(isOneShotCommandValue("/exit")).toBe(true); expect(isOneShotCommandValue("/HELP")).toBe(true); // case-insensitive command expect(isOneShotCommandValue("/sessions")).toBe(true); + expect(isOneShotCommandValue("/new")).toBe(true); expect(isOneShotCommandValue("/compact")).toBe(true); }); @@ -386,8 +392,15 @@ describe("completionCandidates / applyCompletion", () => { { name: "model", description: "Switch the session model" }, ]; const out = completionCandidates("/", withBridge)!; - // Local help/sessions/exit lead even though the bridge doesn't advertise them. - expect(out.map((c) => c.value)).toEqual(["/help", "/sessions", "/exit", "/compact", "/model"]); + // Local help/sessions/new/exit lead even though the bridge doesn't advertise them. + expect(out.map((c) => c.value)).toEqual([ + "/help", + "/sessions", + "/new", + "/exit", + "/compact", + "/model", + ]); // /help output uses the same merged menu. const help = handleLocalCommand("/help", withBridge)!; expect(help.some((e) => e.kind === "note" && e.text.includes("/help —"))).toBe(true); From 0a5c32381d643d93ed7dc045a549abb9e7018a07 Mon Sep 17 00:00:00 2001 From: William Wang Date: Mon, 31 Aug 2026 11:26:33 +0800 Subject: [PATCH 04/14] fix: cap in-memory prompt history at HISTORY_MAX --- src/repl/run.ts | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/src/repl/run.ts b/src/repl/run.ts index ec6bbf8..09e49d0 100644 --- a/src/repl/run.ts +++ b/src/repl/run.ts @@ -47,7 +47,7 @@ import { type SessionSummary, type TurnState, } from "./model.js"; -import { historyPath, loadHistory, pushHistory, saveHistory } from "./history.js"; +import { HISTORY_MAX, historyPath, loadHistory, pushHistory, saveHistory } from "./history.js"; import { createLineEditor, replaceText, type LineEditor } from "./input-buffer.js"; export async function runRepl(): Promise { @@ -556,7 +556,10 @@ export async function runRepl(): Promise { // submission: the queue drain re-enters through here (viaQueue) and its // entries were already recorded when they were queued. if (!viaQueue) { - history = pushHistory(history, text); + // Keep memory in step with the file: saveHistory trims to the newest + // HISTORY_MAX entries on disk; slice here so a long-lived session's + // in-memory array cannot outgrow the same bound. + history = pushHistory(history, text).slice(-HISTORY_MAX); saveHistory(historyFile, history); historyIdx = -1; historyDraft = null; From 66e28463285c33f64c1d6ac285d68dc6225ec5e3 Mon Sep 17 00:00:00 2001 From: William Wang Date: Mon, 31 Aug 2026 11:26:33 +0800 Subject: [PATCH 05/14] chore: sync docs for REPL history recall and /new --- AGENTS.md | 2 ++ README.md | 2 +- 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/AGENTS.md b/AGENTS.md index d69a004..5ac9c21 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -73,6 +73,8 @@ src/ │ │ input box). No alternate screen, no wheel capture. │ ├── input-buffer.ts Pure caret-editing line editor (code-point caret, │ │ Ctrl-B/F/A/E/U chords) — no React, testable +│ ├── history.ts Per-project prompt history (JSONL under +│ │ ~/.zcode/acp/repl-history), pure + testable │ └── run.ts Orchestration: spawn bridge, pump updates └── bin/ ├── hub.ts Hub daemon entry (`zcode-acp hub`; spawned by absolute path) diff --git a/README.md b/README.md index 5184d9b..7c1e12d 100644 --- a/README.md +++ b/README.md @@ -241,7 +241,7 @@ completion menu — `↑`/`↓` move, `enter` picks the highlighted entry (or `t `→`; typing the exact form already sends), `esc` dismisses. After picking `/model`, `/mode`, or `/thought` the same menu lists the config options (the current one marked `●`) and **enter on a row switches immediately** — no second -confirmation. Argument-free commands (`/exit`, `/help`, `/sessions`, +confirmation. Argument-free commands (`/exit`, `/help`, `/sessions`, `/new`, `/compact`, `/mcp`, `/quota`) run on pick as well; every other completion (skills, plugins) only fills the line, since those usually expect arguments. The arg-less forms still print a static listing over the same slash-command From 6573809b35bc658115fac27290098f91d2d9cc1e Mon Sep 17 00:00:00 2001 From: William Wang Date: Mon, 31 Aug 2026 15:15:47 +0800 Subject: [PATCH 06/14] feat: live turn status row with elapsed timer --- CHANGELOG.md | 4 ++++ README.md | 4 +++- src/repl/App.tsx | 31 +++++++++++++++++++++++++++++-- 3 files changed, 36 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0c6c803..40e4454 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,6 +26,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 boundary, and the prompt draft is cleared. A running turn refuses it (`esc` interrupts first); it is registered as a one-shot command, so picking it in the completion menu executes immediately. +- A live status row while a turn runs — `⠋ working… (12s · esc to interrupt)`, + phase-labeled thinking/writing/working — re-rendering every second so + stretches with no streamed output (long tool calls) are visibly alive; the + old dim "ctrl-c to cancel" line carried no liveness signal. - Interactive REPL (bare `zcode-acp`): an Ink terminal chat over the same bridge the editor uses, including slash-command completion with an interactive menu, a caret-aware prompt line (arrows/Ctrl-B/F/A/E/U), diff --git a/README.md b/README.md index 7c1e12d..fc2e254 100644 --- a/README.md +++ b/README.md @@ -253,7 +253,9 @@ Submitted prompts are history: `↑`/`↓` (with the completion menu closed) recall them per project across restarts — the first `↑` stashes the draft and `↓` past the newest entry restores it. Pasted text folds to a single line (newlines and tabs become spaces), so a multi-paragraph paste lands in -the box as one prompt instead of firing line-by-line. +the box as one prompt instead of firing line-by-line. While a reply streams, +the footer shows a live status row — `⠋ working… (12s · esc to interrupt)` — +so stretches with no streamed output (long tool calls) still visibly tick. Unexpected internal errors never take the REPL down silently: they print to stderr and surface as an `-- error absorbed: …` note in the transcript while diff --git a/src/repl/App.tsx b/src/repl/App.tsx index bd2bb9e..3a08244 100644 --- a/src/repl/App.tsx +++ b/src/repl/App.tsx @@ -684,6 +684,21 @@ export function App(props: AppProps): ReactElement { // Second consecutive idle Ctrl-C exits; a turn-running Ctrl-C only cancels. const idleIntCount = useRef(0); + // --- live-turn liveness ticker --- + // A turn with no streamed output (model thinking, silent tool calls) would + // otherwise leave the footer frozen — indistinguishable from a stall. The + // status row below re-renders once a second so time visibly progresses. + const [, tick] = useState(0); + const turnStartRef = useRef(null); + if (turn && turnStartRef.current === null) turnStartRef.current = Date.now(); + if (!turn) turnStartRef.current = null; + const turnActive = turn !== null; + useEffect(() => { + if (!turnActive) return; + const timer = setInterval(() => tick((n) => n + 1), 1000); + return () => clearInterval(timer); + }, [turnActive]); + // Native-scrollback layout: full terminal width — no centered reading // column. `rows` still matters for capping the live-turn tail so the input // box can't be pushed off-screen mid-stream; there is no viewport math. @@ -906,6 +921,18 @@ export function App(props: AppProps): ReactElement { // estimate drift INSIDE this box: an undercounted row crops invisibly off // the top instead of stretching the footer past the fold. const turnBudget = Math.min(turnHeight(turn, cols), MAX_TURN_ROWS); + // Status row for the live-turn block: phase label + elapsed seconds. Bright + // (not dim) and self-updating — this line is the "still alive" signal. + const turnElapsed = turnStartRef.current + ? Math.max(0, Math.floor((Date.now() - turnStartRef.current) / 1000)) + : 0; + const turnLabel = turn + ? turn.textBuf + ? "writing" + : turn.thinkBuf.trim() + ? "thinking" + : "working" + : ""; return ( @@ -940,8 +967,8 @@ export function App(props: AppProps): ReactElement { ⎿ thinking · {turn.thinkBuf.replace(/\s+/g, " ").slice(-120)} ) : null} {turn.textBuf ? {colorizeCodeFences(turn.textBuf)} : null} - - ctrl-c to cancel + + {turnLabel}… ({turnElapsed}s · esc to interrupt) ) : null} From 09226a098ea4ce2bd0691a9f451b76c03e3f6c84 Mon Sep 17 00:00:00 2001 From: William Wang Date: Mon, 31 Aug 2026 15:45:42 +0800 Subject: [PATCH 07/14] fix: fan out session-scoped emits per attached ACP session alias --- CHANGELOG.md | 7 +++ src/handlers/dispatch.ts | 78 ++++++++++++++++++---------------- src/handlers/io.ts | 36 ++++++++++------ src/handlers/session.ts | 22 +++++++--- src/index.ts | 12 ++++-- src/server.ts | 37 +++++++++++++--- tests/audit-fixes.test.ts | 1 + tests/dispatch.test.ts | 69 ++++++++++++++++++++++++++++++ tests/remote-broadcast.test.ts | 15 +++++-- 9 files changed, 211 insertions(+), 66 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 40e4454..72f605c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -86,6 +86,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 overwrite the resumed one; the placeholder is now discarded. - Resume no longer force-pins a session to the first config.json model (faithful model preservation, overlay demoted to one-retry fallback). +- Turns driven from another client (mobile app, second editor) now render + live in the REPL even when the two hold different ACP session ids for the + same conversation — the common "fresh REPL session, mobile follow-up" + path previously stayed completely silent (no live turn, no streaming, no + completion, while the other client saw everything). Session-scoped + notifications (updates, turnState, prompt echo) are now emitted once per + attached session alias. ## [0.13.0] - 2026-08-26 diff --git a/src/handlers/dispatch.ts b/src/handlers/dispatch.ts index 82f762f..a2964a8 100644 --- a/src/handlers/dispatch.ts +++ b/src/handlers/dispatch.ts @@ -36,42 +36,48 @@ export async function dispatchEvent( ev: InternalEvent, chunkMsgId: string, ): Promise { - switch (ev.kind) { - case "ToolCallNew": - await dispatchToolCallNew(server, cx, acpSid, ev); - break; - case "ToolCallUpdate": - await dispatchToolCallUpdate(server, cx, acpSid, ev); - break; - case "UsageDelta": - await dispatchUsageDelta(server, cx, acpSid, ev); - break; - case "TextDelta": - await sendSessionUpdate(cx, acpSid, { - sessionUpdate: "agent_message_chunk", - content: { type: "text", text: ev.text }, - messageId: chunkMsgId, - }); - break; - case "ReasoningDelta": - await sendSessionUpdate(cx, acpSid, { - sessionUpdate: "agent_thought_chunk", - content: { type: "text", text: ev.text }, - messageId: `thought_${chunkMsgId}`, - }); - break; - case "PlanUpdate": - await sendSessionUpdate(cx, acpSid, { - sessionUpdate: "plan", - entries: ev.entries, - }); - break; - case "FilesChanged": - await dispatchFilesChanged(cx, acpSid, ev); - break; - case "ConfigChanged": - await dispatchConfigChanged(server, cx, acpSid, ev); - break; + // One conversation can be attached under several ACP ids (see + // server.sessionAliases): emit once per alias with that alias as the + // payload sessionId, or every client but the prompter starves silently. + const targets = server.sessionAliases(acpSid); + for (const sid of targets) { + switch (ev.kind) { + case "ToolCallNew": + await dispatchToolCallNew(server, cx, sid, ev); + break; + case "ToolCallUpdate": + await dispatchToolCallUpdate(server, cx, sid, ev); + break; + case "UsageDelta": + await dispatchUsageDelta(server, cx, sid, ev); + break; + case "TextDelta": + await sendSessionUpdate(cx, sid, { + sessionUpdate: "agent_message_chunk", + content: { type: "text", text: ev.text }, + messageId: chunkMsgId, + }); + break; + case "ReasoningDelta": + await sendSessionUpdate(cx, sid, { + sessionUpdate: "agent_thought_chunk", + content: { type: "text", text: ev.text }, + messageId: `thought_${chunkMsgId}`, + }); + break; + case "PlanUpdate": + await sendSessionUpdate(cx, sid, { + sessionUpdate: "plan", + entries: ev.entries, + }); + break; + case "FilesChanged": + await dispatchFilesChanged(cx, sid, ev); + break; + case "ConfigChanged": + await dispatchConfigChanged(server, cx, sid, ev); + break; + } } } diff --git a/src/handlers/io.ts b/src/handlers/io.ts index 67e81ef..4d9cb2b 100644 --- a/src/handlers/io.ts +++ b/src/handlers/io.ts @@ -126,23 +126,31 @@ export function echoUserPromptToOthers( .trim(); if (!text) return; const messageId = `uprompt_${randomUUID()}`; - void enqueueSessionSend(params.sessionId, () => - server.clients - .notifyOthers(prompter, "session/update", { - sessionId: params.sessionId, - update: { - sessionUpdate: "user_message_chunk", - content: { type: "text", text }, - messageId, - }, - }) - .catch((e: unknown) => { + void enqueueSessionSend(params.sessionId, async () => { + // Emit once per attached alias (server.sessionAliases): clients route + // session/update by payload sessionId, so the prompter's id alone never + // reaches a client holding this conversation under a different id. + const results = await Promise.allSettled( + server.sessionAliases(params.sessionId).map((sid) => + server.clients.notifyOthers(prompter, "session/update", { + sessionId: sid, + update: { + sessionUpdate: "user_message_chunk", + content: { type: "text", text }, + messageId, + }, + }), + ), + ); + for (const r of results) { + if (r.status === "rejected") { warn( `user-prompt echo failed (sid=${params.sessionId}): ` + - `${e instanceof Error ? e.message : String(e)}`, + `${r.reason instanceof Error ? r.reason.message : String(r.reason)}`, ); - }), - ); + } + } + }); } /** Shape of a slash command entry (matches ACP's AvailableCommand). */ diff --git a/src/handlers/session.ts b/src/handlers/session.ts index b8558cc..6dfdb0f 100644 --- a/src/handlers/session.ts +++ b/src/handlers/session.ts @@ -646,12 +646,24 @@ export async function prompt( } // Out-of-band running indicator: clients that did not send this prompt // (re-attached mobile, second editor) learn the turn started here — the - // session/load replayMeta only snapshots attach time. Best-effort: a dead + // session/load replayMeta only snapshots attach time. Emitted per attached + // alias (see server.sessionAliases) so a client holding this conversation + // under a different ACP id opens its live turn too. Best-effort: a dead // client must not fail the turn. - const emitTurnState = (running: boolean): Promise => - cx - .notify("$/zcode/turnState", { sessionId: params.sessionId, running }) - .catch((e) => log(`turnState notify failed: ${e instanceof Error ? e.message : String(e)}`)); + const emitTurnState = async (running: boolean): Promise => { + const results = await Promise.allSettled( + server + .sessionAliases(params.sessionId) + .map((sid) => cx.notify("$/zcode/turnState", { sessionId: sid, running })), + ); + for (const r of results) { + if (r.status === "rejected") { + log( + `turnState notify failed: ${r.reason instanceof Error ? r.reason.message : String(r.reason)}`, + ); + } + } + }; await emitTurnState(true); const listener = new EventStreamListener(backend, zcodeSid); diff --git a/src/index.ts b/src/index.ts index 29c0eba..5e264ca 100644 --- a/src/index.ts +++ b/src/index.ts @@ -122,13 +122,17 @@ export async function main(): Promise { .onRequest("initialize", (ctx) => server.initialize(ctx.params)) .onRequest("session/new", async (ctx) => { const result = await newSession(server, ctx.params); - sendAvailableCommandsDeferred(server.clients.broadcast(), result.sessionId, allCommands); + for (const sid of server.sessionAliases(result.sessionId)) { + sendAvailableCommandsDeferred(server.clients.broadcast(), sid, allCommands); + } return result; }) .onRequest("session/list", (ctx) => listSessions(server, ctx.params)) .onRequest("session/resume", async (ctx) => { const result = await resumeSession(server, ctx.params, server.clients.broadcast()); - sendAvailableCommandsDeferred(server.clients.broadcast(), ctx.params.sessionId, allCommands); + for (const sid of server.sessionAliases(ctx.params.sessionId)) { + sendAvailableCommandsDeferred(server.clients.broadcast(), sid, allCommands); + } // A client that (re)connects catches up via resume/load; any interaction // request still waiting for an answer is re-sent to it so a question // fired while it was offline becomes answerable there. @@ -137,7 +141,9 @@ export async function main(): Promise { }) .onRequest("session/load", async (ctx) => { const result = await loadSession(server, ctx.params, server.clients.broadcast()); - sendAvailableCommandsDeferred(server.clients.broadcast(), ctx.params.sessionId, allCommands); + for (const sid of server.sessionAliases(ctx.params.sessionId)) { + sendAvailableCommandsDeferred(server.clients.broadcast(), sid, allCommands); + } resendPendingInteractions(server, ctx.client, ctx.params.sessionId); return result; }) diff --git a/src/server.ts b/src/server.ts index cb6e828..4e8016b 100644 --- a/src/server.ts +++ b/src/server.ts @@ -315,6 +315,27 @@ export class ZcodeAcpServer { return this.acpSidByZcodeSid.get(zcodeSid); } + /** + * Every ACP alias attached to the same backend conversation — at least + * [acpSid] itself. Two clients can hold DIFFERENT acpSids for one + * conversation (a fresh session/new placeholder in one, a session/list id + * resumed in another), and clients route session-scoped notifications by + * the payload sessionId. Anything emitted under the prompting client's id + * alone is silently dropped by every client holding another alias, so + * session-scoped emits (updates, turnState, prompt echo) must loop this + * list. A client holding two aliases of one conversation gets both copies — + * pathological, accepted. + */ + sessionAliases(acpSid: string): string[] { + const zcodeSid = this.sessionMap.get(acpSid); + if (!zcodeSid) return [acpSid]; + const aliases: string[] = []; + for (const [sid, zsid] of this.sessionMap) { + if (zsid === zcodeSid) aliases.push(sid); + } + return aliases.length > 0 ? aliases : [acpSid]; + } + /** * Push a `session/update` notification to the client from OUTSIDE a request * handler (used by the background-task listener). Resolves the acp_sid from @@ -329,11 +350,17 @@ export class ZcodeAcpServer { try { // Broadcast notify swallows per-client failures internally (warn only). // Serialized through the replay guard so a background emission queues - // behind an in-flight replay batch for the same session. - await enqueueSessionSend(acpSid, () => - this.clients.broadcast().notify("session/update", { sessionId: acpSid, update }), - ); - return true; + // behind an in-flight replay batch for the same session. Emitted per + // attached alias (sessionAliases) so a client holding this conversation + // under a different ACP id receives it too. + let sent = false; + for (const alias of this.sessionAliases(acpSid)) { + await enqueueSessionSend(alias, () => + this.clients.broadcast().notify("session/update", { sessionId: alias, update }), + ); + sent = true; + } + return sent; } catch (e) { log(`notifyByZcodeSid: session/update failed: ${e instanceof Error ? e.message : String(e)}`); return false; diff --git a/tests/audit-fixes.test.ts b/tests/audit-fixes.test.ts index 89bda15..37680be 100644 --- a/tests/audit-fixes.test.ts +++ b/tests/audit-fixes.test.ts @@ -171,6 +171,7 @@ describe("dispatchPlanIfChanged delayed re-check", () => { const server = { ensureBackend: () => backend, nextId: () => 1, + sessionAliases: (sid: string) => [sid], } as unknown as ZcodeAcpServer; const cx = { notify: vi.fn().mockResolvedValue(undefined) } as unknown as acp.AgentContext; diff --git a/tests/dispatch.test.ts b/tests/dispatch.test.ts index 1a113cc..300baff 100644 --- a/tests/dispatch.test.ts +++ b/tests/dispatch.test.ts @@ -371,3 +371,72 @@ describe("dispatchEvent", () => { expect(options[2]).toMatchObject({ id: "thought", currentValue: "high" }); }); }); + +/** + * Cross-alias fan-out (server.sessionAliases): one conversation attached + * under several ACP ids (a local session/new placeholder AND a remote + * session/list id) must emit one update PER alias — clients route + * session/update by payload sessionId, so a single emission under the + * prompting client's id silently starves every other attached client. + */ +describe("cross-alias fan-out", () => { + /** Mock AgentContext recording the sessionId of every notify call. */ + function recordingContext(): { + cx: acp.AgentContext; + calls: Array<{ sessionId: string; update: acp.SessionUpdate }>; + } { + const calls: Array<{ sessionId: string; update: acp.SessionUpdate }> = []; + const cx = { + notify(_method: string, params: { sessionId: string; update: acp.SessionUpdate }) { + calls.push(params); + return Promise.resolve(); + }, + } as unknown as acp.AgentContext; + return { cx, calls }; + } + + it("sessionAliases lists every acp id attached to the same backend session", () => { + const server = makeServer(false); + expect(server.sessionAliases("unknown")).toEqual(["unknown"]); + server.registerSession("acp_a", "zc_1"); + expect(server.sessionAliases("acp_a")).toEqual(["acp_a"]); + server.registerSession("acp_b", "zc_1"); + server.registerSession("acp_z", "zc_1"); + expect(new Set(server.sessionAliases("acp_a"))).toEqual(new Set(["acp_a", "acp_b", "acp_z"])); + // Re-registering overwrites only that pair, never drops siblings. + server.registerSession("acp_b", "zc_1"); + expect(new Set(server.sessionAliases("acp_a"))).toEqual(new Set(["acp_a", "acp_b", "acp_z"])); + }); + + it("dispatchEvent emits once per attached alias", async () => { + const { cx, calls } = recordingContext(); + const server = makeServer(false); + server.registerSession("acp_a", "zc_1"); + server.registerSession("acp_b", "zc_1"); + await dispatchEvent( + server, + cx, + "acp_a", + { kind: "TextDelta", text: "hi" } as InternalEvent, + CHUNK, + ); + expect(calls.map((c) => c.sessionId).sort()).toEqual(["acp_a", "acp_b"]); + for (const c of calls) { + expect(c.update).toMatchObject({ sessionUpdate: "agent_message_chunk", messageId: CHUNK }); + } + }); + + it("unregistered sessions keep the single-emission fast path", async () => { + const { cx, calls } = recordingContext(); + const server = makeServer(false); + await dispatchEvent( + server, + cx, + "solo", + { kind: "TextDelta", text: "hi" } as InternalEvent, + CHUNK, + ); + expect(calls).toHaveLength(1); + expect(calls[0]!.sessionId).toBe("solo"); + }); +}); diff --git a/tests/remote-broadcast.test.ts b/tests/remote-broadcast.test.ts index 2ff4099..d22d0b6 100644 --- a/tests/remote-broadcast.test.ts +++ b/tests/remote-broadcast.test.ts @@ -184,7 +184,10 @@ describe("echoUserPromptToOthers", () => { it("echoes the prompt text to other clients, never the prompter", async () => { const { registry, zed, phone, prompter } = registryWithZedAndPhone(); - const server = { clients: registry } as unknown as ZcodeAcpServer; + const server = { + clients: registry, + sessionAliases: (sid: string) => [sid], + } as unknown as ZcodeAcpServer; echoUserPromptToOthers(server, prompter, { sessionId: "s1", prompt: "hello from zed" }); await sleep(0); @@ -203,7 +206,10 @@ describe("echoUserPromptToOthers", () => { it("joins text blocks of a structured prompt and skips non-text ones", async () => { const { registry, phone, prompter } = registryWithZedAndPhone(); - const server = { clients: registry } as unknown as ZcodeAcpServer; + const server = { + clients: registry, + sessionAliases: (sid: string) => [sid], + } as unknown as ZcodeAcpServer; const prompt = [ { type: "text", text: "look at this" }, { type: "image", data: "…" }, @@ -221,7 +227,10 @@ describe("echoUserPromptToOthers", () => { it("sends nothing for an empty prompt", async () => { const { registry, phone, prompter } = registryWithZedAndPhone(); - const server = { clients: registry } as unknown as ZcodeAcpServer; + const server = { + clients: registry, + sessionAliases: (sid: string) => [sid], + } as unknown as ZcodeAcpServer; echoUserPromptToOthers(server, prompter, { sessionId: "s1", prompt: " " }); await sleep(0); From 9058c7e3e978041ed47090610c86b43dd39a2ec2 Mon Sep 17 00:00:00 2001 From: William Wang Date: Mon, 31 Aug 2026 16:58:48 +0800 Subject: [PATCH 08/14] fix: honour cancel bridge-side and drain backend before follow-up sends --- AGENTS.md | 19 ++++ CHANGELOG.md | 29 +++++- README.md | 13 ++- src/backend/listener.ts | 11 +- src/handlers/session.ts | 101 ++++++++++++++---- src/repl/App.tsx | 7 +- tests/session-cancel.test.ts | 193 +++++++++++++++++++++++++++++++++++ 7 files changed, 340 insertions(+), 33 deletions(-) create mode 100644 tests/session-cancel.test.ts diff --git a/AGENTS.md b/AGENTS.md index 5ac9c21..cc4b8d4 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -105,6 +105,12 @@ ZCode protocol types into ACP notifications directly — always translate. - **ZCode backend version drift**: the backend may change event payloads between releases. When diff display or event handling breaks, check the raw backend event with `ZCODE_ACP_DEBUG=1` before changing translator code. +- **The backend ignores `session/stop`** (verified against app-server 0.16.5 — + the model stream runs to its natural end no matter what). Cancel is therefore + bridge-side only: the turn loop returns `cancelled` at once, and the next + prompt's turn-attribution gate (armed on a recent cancel) drops the abandoned + turn's leftover stream. Never "wait for the backend terminal event" after a + cancel — that made ESC feel dead for the whole remaining generation. - **`session/prompt` ordering**: subscribe to events BEFORE calling `session/send` — short turns can complete before a late subscribe catches them. - **Preempt lock**: concurrent prompts for the same session are serialized via @@ -133,6 +139,19 @@ ZCode protocol types into ACP notifications directly — always translate. - **REPL render state lives in run.ts, not React**: App re-renders from fresh snapshots; anything that must persist across them (prompt-line editor, queue, entries) belongs to run.ts's external store passed via snapshot props. +- **Aug-28 app-server build (still "0.16.5") ignores `session/stop`**: the + RPC returns `{}` but the model stream runs to its natural end (verified by + raw-backend probe). Cancel must therefore be honoured bridge-side — the + turn loop returns `stopReason: "cancelled"` on the flag instead of waiting + for a terminal event, and a send after a recent cancel waits for the + backend to report idle (a mid-generation send is accepted as steer input + and silently dropped when the old turn ends). If a future build fixes the + stop, the early return stays correct (client cancel semantics); only the + drain gate becomes a no-op. +- **The backend rejects JSON-RPC frames carrying a `jsonrpc` field** (strict + zod: "Unrecognized key: jsonrpc", code -32600). The bridge's backend + client never sends one — keep it that way when hand-probing + `zcode app-server --stdio` (frames are bare `{id, method, params}`). ## Docs to read before sensitive changes diff --git a/CHANGELOG.md b/CHANGELOG.md index 72f605c..db520f0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -29,7 +29,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - A live status row while a turn runs — `⠋ working… (12s · esc to interrupt)`, phase-labeled thinking/writing/working — re-rendering every second so stretches with no streamed output (long tool calls) are visibly alive; the - old dim "ctrl-c to cancel" line carried no liveness signal. + old dim "ctrl-c to cancel" line carried no liveness signal. Help lines and + the input-box hint now advertise `esc` as the interrupt (ctrl-c is quit). - Interactive REPL (bare `zcode-acp`): an Ink terminal chat over the same bridge the editor uses, including slash-command completion with an interactive menu, a caret-aware prompt line (arrows/Ctrl-B/F/A/E/U), @@ -73,6 +74,22 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- `esc`/stop now takes effect immediately. The Aug-28 app-server build + (still reporting 0.16.5) accepts `session/stop` but no longer aborts the + in-flight model stream (verified with a raw-backend probe: the stream ran + ~10s past the stop to its natural end), so the bridge's turn loop waited + for a terminal event that only came after the full generation — every + client (REPL, mobile, editor) saw the stop "not work" while output kept + streaming. The turn loop now returns `stopReason: "cancelled"` the moment + the cancel flag is observed. +- A follow-up prompt sent right after a cancel/preempt is no longer silently + dropped. The same backend build accepts a mid-generation `session/send` + as a steer and discards its input when the old turn finishes (verified: + only one `turn.completed` ever arrives, for the old prompt). The bridge + now polls the session until the backend reports idle before sending — + with a visible `[上一个回复仍在生成,等待结束后发送…]` note — bounded at + 90s, still interruptible with `esc`, falling back to a direct send on + timeout or probe failure. - Pressing ↓ with no completion menu open no longer zombifies the whole UI: the setState updater dereferenced a null menu during render, unmounting React's tree under ink without any crash signal (found by review, @@ -93,6 +110,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 completion, while the other client saw everything). Session-scoped notifications (updates, turnState, prompt echo) are now emitted once per attached session alias. +- ESC (and ctrl-c) now interrupts a running turn immediately. The backend + ignores `session/stop` (verified against app-server 0.16.5 — the model + stream runs to its natural end regardless), and the turn loop used to wait + for that terminal event before reporting cancelled, so the reply kept + streaming for the whole remaining generation (10s+ observed) while the + status row kept spinning. The loop now returns `cancelled` at once; a + follow-up prompt sent during the abandoned turn's finalisation arms the + turn-attribution gate so the residue is dropped instead of bleeding into + the new reply. REPL hint copy now advertises esc as the interrupt + ("esc interrupt") instead of ctrl-c. ## [0.13.0] - 2026-08-26 diff --git a/README.md b/README.md index fc2e254..69841b2 100644 --- a/README.md +++ b/README.md @@ -217,9 +217,16 @@ menu is open. The status row carries a compact plan-quota readout (`5h 16% · wk 4%`) refreshed every 10 minutes — `/quota` prints the full card. Pasted or dragged-in content (error logs, file drops) is sanitized and size-capped before it reaches the editor, so long pastes batch cleanly. -`Ctrl-C` cancels a running turn; `esc` interrupts one too; while idle, press -Ctrl-C twice to quit. `/exit` leaves; the session itself persists in the -ZCode backend and is available to your editor. +`esc` interrupts a running turn (immediately — the bridge resolves the prompt +as cancelled without waiting for the backend); `Ctrl-C` also interrupts, and +while idle press it twice to quit. `/exit` leaves; the session itself persists +in the ZCode backend and is available to your editor. + +Sending a follow-up right after an interrupt waits for the backend to finish +the cancelled generation first — a `[上一个回复仍在生成,等待结束后发送…]` +note explains the pause (the Aug-28 app-server accepts mid-generation sends +as steer input but drops them when the old turn ends; the bridge polls until +the session is idle, up to 90s, so the message actually runs). Messages typed while a turn is running (or the session is still starting) are queued, not lost: each shows up in the transcript immediately and a `⏸ queued` diff --git a/src/backend/listener.ts b/src/backend/listener.ts index 78cc4f3..1837188 100644 --- a/src/backend/listener.ts +++ b/src/backend/listener.ts @@ -57,11 +57,12 @@ export class EventStreamListener { */ async subscribe(nextId: NextId): Promise { // Retry transient timeouts as a lightweight safety net for cold-start / - // network blips. The cancel-preempt path no longer needs subscribe retries - // to absorb a backend stop-finalization window: the turn loop now blocks - // until the backend emits turn.completed/turn.failed before its prompt() - // exits, so by the time the next prompt reaches subscribe the backend is - // already idle. These retries are just a last-resort cushion. + // network blips. The cancel path returns from prompt() at once (the + // backend ignores session/stop), so a prompt sent right after a cancel + // CAN reach subscribe while the abandoned turn is still generating — + // prompt()'s drain gate polls the backend to idle before sending, so + // residue is gone by the time this subscribes. These retries are just a + // last-resort cushion. // // Only `timeout` is retried — non-transient errors (reader dead, pipe // broken, method-not-found, session-level business error) fail fast. diff --git a/src/handlers/session.ts b/src/handlers/session.ts index 6dfdb0f..9d33090 100644 --- a/src/handlers/session.ts +++ b/src/handlers/session.ts @@ -756,6 +756,45 @@ export async function prompt( const chunkMsgId = randomUUID(); + // Drain gate: a recent cancel/preempt means the backend is STILL + // generating — session/stop does not abort the model stream (verified + // against app-server 0.16.5) — and a send that lands mid-generation is + // accepted as a steer whose input the backend silently DROPS when the + // old turn finishes (verified: only one turn.completed ever arrives). + // Poll the projection until the backend reports idle so the user's + // follow-up actually runs; a visible chunk tells them why it waits. + // Bounded: on timeout (or a failed probe) send anyway — the steer-drop + // risk returns, but blocking the prompt forever is worse. + const DRAIN_TIMEOUT_MS = 90_000; + const DRAIN_POLL_MS = 1000; + const cancelledRecently = + server.lastCancelledAt.get(zcodeSid) !== undefined && + Date.now() - server.lastCancelledAt.get(zcodeSid)! < DRAIN_TIMEOUT_MS; + if (cancelledRecently) { + const drainMonitor = new TurnMonitor(backend, zcodeSid, () => server.nextId()); + const drainT0 = Date.now(); + let noticed = false; + while (Date.now() - drainT0 < DRAIN_TIMEOUT_MS) { + if (turn.cancelled) { + stopBackendTurn(server, zcodeSid); + return { stopReason: "cancelled" }; + } + const proj = await drainMonitor.pollOnce(); + if (!proj) break; // probe failed — don't block the send on it + if (proj.status === "idle") break; + if (!noticed) { + noticed = true; + await sendTextChunk( + cx, + params.sessionId, + "[上一个回复仍在生成,等待结束后发送…]", + randomUUID(), + ); + } + await sleep(DRAIN_POLL_MS); + } + } + // Send the prompt, retrying while the backend reports it's still busy. // The backend's prompt lock is the single authoritative readiness signal: // a rejected send (code 1308 "prompt is running") means a previous turn @@ -821,6 +860,13 @@ export async function prompt( try { // Event-driven turn loop: translate events via EventTranslator + dispatch. + // Arm the attribution gate also on a recent cancel: the abandoned turn + // is still finalising in the backend (session/stop is not honored — + // verified 0.16.5), and its leftover deltas stream past the subscribe + // of this new prompt (see the gate comment in runEventTurn). + const gateArmed = + preempted || + Date.now() - (server.lastCancelledAt.get(zcodeSid) ?? 0) < CANCEL_RESIDUE_WINDOW_MS; const result = await runEventTurn( server, listener, @@ -830,7 +876,7 @@ export async function prompt( params.sessionId, chunkMsgId, turn, - preempted, + gateArmed, ); // (Session title: already set once at the FIRST prompt, before the @@ -895,6 +941,10 @@ export async function prompt( } } +/** How long after a cancel a new prompt's attribution gate stays armed (the + * abandoned turn may still be streaming its finalisation into the backend). */ +const CANCEL_RESIDUE_WINDOW_MS = 120_000; + /** * `session/set_config_option` → dispatch model/mode/thought and emit the * resulting config_option_update (+ current_mode_update for mode). @@ -934,9 +984,11 @@ export async function setConfigOptionHandler( * on a session with no active turn, and on a turn already aborted), so firing * it eagerly is safe; the loop's `stopSent` guard prevents a second send. * - * `turn.cancelled` is still set so the turn loop switches to its silent-drain - * path (translate to detect turnDone, but discard every internal event — no - * text/tool/usage is pushed after the user stopped). + * `turn.cancelled` is still set so the turn loop returns at once (the backend + * ignores session/stop — verified 0.16.5, the model stream runs to its natural + * end — so waiting for a terminal event would hang the stop for the whole + * remaining generation). The loop's return resolves session/prompt with + * stopReason "cancelled" immediately. */ export async function cancel( server: ZcodeAcpServer, @@ -1377,7 +1429,7 @@ function getOrCreateDiffer(server: ZcodeAcpServer, zcodeSid: string): Projection * handling (requestPermission / ExitPlanMode / AskUserQuestion) lands in * Commit 6 — for now they're polled to keep the inbox clear. */ -async function runEventTurn( +export async function runEventTurn( server: ZcodeAcpServer, listener: EventStreamListener, monitor: TurnMonitor, @@ -1386,7 +1438,7 @@ async function runEventTurn( acpSid: string, chunkMsgId: string, turn: PendingTurn, - preempted: boolean, + gateArmed: boolean, ): Promise { const backend = server.ensureBackend(); const translator = new EventTranslator(); @@ -1424,18 +1476,22 @@ async function runEventTurn( } if (turn.cancelled) { - // Cancel requested: ensure stop was fired (cancel()/preempt normally do - // this, but guard anyway). We do NOT silence subsequent events here — if - // the backend ignored the stop and kept producing, that content is still - // valuable to the user and should be displayed (the backend is the single - // source of truth within a session). Cross-turn contamination is handled - // separately by the turn-attribution gate below, which discards this - // turn's leftover events from the *next* turn's queue. The loop exits - // normally on the terminal event (translator.turnDone below). + // Cancel requested: fire the stop (cancel()/preempt normally already + // did — this is a guard) and END THE TURN AT ONCE. The backend's + // session/stop is fire-and-forget and, as verified against app-server + // 0.16.5, does NOT abort the in-flight model stream — waiting for the + // backend's terminal event used to keep the turn streaming for the + // full remaining generation (10s+ past the stop) while the user stared + // at a live spinner. Returning here resolves session/prompt with + // stopReason "cancelled" immediately; the finally below unregisters + // the turn listener, so events the backend still pushes are delivered + // to no turn listener, and the next turn's turn-attribution gate + // discards any residue that slipped into the queue meanwhile. if (!turn.stopSent) { stopBackendTurn(server, turn.zcodeSid); turn.stopSent = true; } + return { stopReason: "cancelled" }; } const ev = await listener.pollEvent(500); @@ -1538,13 +1594,16 @@ async function runEventTurn( // terminal event here would flip them and make THIS turn exit prematurely // at the first check after its own turn.started passes the gate. // - // The gate is armed ONLY when this send preempted another prompt. Without - // preemption no prior-turn residue can exist: the queue can only contain - // events of a backend-owned turn that was already active at send time - // (e.g. the main-branch turn auto-resumed after a compaction) — this send - // was steered into it and produces NO new turn.started, so dropping those - // events would silently swallow the entire turn's output in the UI. - if (shouldDropEventForTurnAttribution(ev, translator.turnStarted, preempted)) { + // Armed when this send preempted another prompt, OR when a cancel is + // recent: the backend ignores session/stop, so an abandoned turn keeps + // streaming until its natural end while the turn loop has already + // returned — a prompt sent in that window reaches subscribe while the + // residue is still arriving. Backend serialisation bounds the exposure: + // a send is accepted only after the prior turn released the lock, so the + // residue can only arrive BEFORE this turn's turn.started. (In the rare + // steered-into backend-owned turn, where no turn.started ever arrives, + // the completion diff replays the dropped text at turn completion.) + if (shouldDropEventForTurnAttribution(ev, translator.turnStarted, gateArmed)) { continue; } const internalEvents = translator.translate(ev); diff --git a/src/repl/App.tsx b/src/repl/App.tsx index 3a08244..2958850 100644 --- a/src/repl/App.tsx +++ b/src/repl/App.tsx @@ -203,7 +203,8 @@ function WelcomeView({ info }: { info: WelcomeInfo }): ReactElement { {" / command menu — ↑/↓ move · enter picks · esc closes"} {" /model switch model — /mode and /thought likewise"} {" /sessions list and resume past conversations of this project"} - {" ctrl-c cancel a running turn · idle, press twice to quit"} + {" esc interrupt the running turn"} + {" ctrl-c quit — press twice when idle"} ); @@ -637,7 +638,7 @@ function InputLine({ {leftLine || "type / for commands · tab completes"} - {busy ? "" : "enter send · ctrl-c cancels/quits"} + {busy ? "" : "enter send · esc interrupt · ctrl-c quit"} @@ -653,7 +654,7 @@ function entryHeight(entry: ReplEntry, width: number): number { case "user": return 1 + estimateLines(`> ${entry.text}`, width); case "welcome": - return 1 + 11; + return 1 + 12; case "tool": return estimateLines(`• ${entry.title} (${entry.status})`, width); default: diff --git a/tests/session-cancel.test.ts b/tests/session-cancel.test.ts new file mode 100644 index 0000000..fc6cca2 --- /dev/null +++ b/tests/session-cancel.test.ts @@ -0,0 +1,193 @@ +/** + * Regression tests for the cancel path. + * + * The backend (app-server 0.16.5, verified live) ignores `session/stop` — the + * model stream runs to its natural end regardless. Cancel therefore only works + * if the bridge acts on the flag itself: the turn loop returns immediately and + * `cancel()` eagerly fires the stop + stamps the cancel time (fast-fail for a + * prompt sent during the backend's recovery window). + */ + +import type * as acp from "@agentclientprotocol/sdk"; +import { describe, expect, it, vi } from "vitest"; + +import type { EventStreamListener, TurnMonitor } from "../src/backend/listener.js"; +import type { ProjectionDiffer } from "../src/translators/projection-differ.js"; +import { cancel, runEventTurn } from "../src/handlers/session.js"; +import type { PendingTurn, ZcodeAcpServer } from "../src/server.js"; + +interface FakeTurn { + zcodeSid: string; + cancelled: boolean; + stopSent: boolean; +} + +function makeServer(turns: FakeTurn[], resolve: (sid: string) => string | undefined) { + const sent: Array<{ method: string; params: unknown }> = []; + const server = { + resolveSid: resolve, + pendingTurns: new Map(turns.map((t, i) => [`req${i}`, t])), + lastCancelledAt: new Map(), + ensureBackend: () => ({ + send: (method: string, params: unknown) => { + sent.push({ method, params }); + }, + }), + } as unknown as ZcodeAcpServer; + return { server, sent }; +} + +describe("session/cancel handler", () => { + it("marks the matching turn cancelled and fires session/stop once", async () => { + const turn: FakeTurn = { zcodeSid: "sess_z", cancelled: false, stopSent: false }; + const { server, sent } = makeServer([turn], (sid) => (sid === "acp_a" ? "sess_z" : undefined)); + + await cancel(server, { sessionId: "acp_a" } as acp.CancelNotification); + + expect(turn.cancelled).toBe(true); + expect(turn.stopSent).toBe(true); + expect(sent).toEqual([{ method: "session/stop", params: { sessionId: "sess_z" } }]); + expect(server.lastCancelledAt.get("sess_z")).toBeGreaterThan(0); + }); + + it("stops every turn of the session, not just the first match", async () => { + const old: FakeTurn = { zcodeSid: "sess_z", cancelled: false, stopSent: true }; + const live: FakeTurn = { zcodeSid: "sess_z", cancelled: false, stopSent: false }; + const other: FakeTurn = { zcodeSid: "sess_o", cancelled: false, stopSent: false }; + const { server, sent } = makeServer([old, live, other], (sid) => + sid === "acp_a" ? "sess_z" : undefined, + ); + + await cancel(server, { sessionId: "acp_a" } as acp.CancelNotification); + + expect(old.cancelled).toBe(true); + expect(live.cancelled).toBe(true); + expect(other.cancelled).toBe(false); + // The already-stopped turn does not fire a second stop. + expect(sent).toEqual([{ method: "session/stop", params: { sessionId: "sess_z" } }]); + }); + + it("no-ops for an unknown session id", async () => { + const send = vi.fn(); + const server = { + resolveSid: () => undefined, + pendingTurns: new Map(), + lastCancelledAt: new Map(), + ensureBackend: () => ({ send }), + } as unknown as ZcodeAcpServer; + + await cancel(server, { sessionId: "acp_ghost" } as acp.CancelNotification); + + expect(send).not.toHaveBeenCalled(); + }); +}); + +/** Minimal fixtures for driving runEventTurn's cancel path in isolation. */ +function makeTurnFixtures() { + const sent: Array<{ method: string; params: unknown }> = []; + const server = { + ensureBackend: () => ({ + send: (method: string, params: unknown) => { + sent.push({ method, params }); + }, + pollServerRequests: () => [], + }), + sessionAliases: (sid: string) => [sid], + } as unknown as ZcodeAcpServer; + const pollEvent = vi.fn(); + const listener = { + pollEvent, + hasQueuedEvents: () => false, + resubscribe: vi.fn(), + } as unknown as EventStreamListener; + const monitor = { pollOnce: vi.fn() } as unknown as TurnMonitor; + const differ = { + resetTurn: vi.fn(), + setLastUsage: vi.fn(), + markToolSeen: vi.fn(), + } as unknown as ProjectionDiffer; + const cx = { notify: vi.fn().mockResolvedValue(undefined) } as unknown as acp.AgentContext; + return { server, sent, pollEvent, listener, monitor, differ, cx }; +} + +describe("runEventTurn: cancel exits immediately (backend ignores session/stop)", () => { + it("returns cancelled at once when the turn is already flagged", async () => { + const f = makeTurnFixtures(); + const turn: PendingTurn = { zcodeSid: "sess_z", cancelled: true, stopSent: false }; + f.pollEvent.mockResolvedValue({ + sessionId: "sess_z", + seq: 1, + type: "model.streaming", + payload: {}, + }); + + const resp = await runEventTurn( + f.server, + f.listener, + f.monitor, + f.differ, + f.cx, + "acp_a", + "m1", + turn, + false, + ); + + expect(resp).toEqual({ stopReason: "cancelled" }); + // Guard-fired stop (stopSent was false), and NOT a single event consumed. + expect(f.sent).toEqual([{ method: "session/stop", params: { sessionId: "sess_z" } }]); + expect(f.pollEvent).not.toHaveBeenCalled(); + }); + + it("returns cancelled on the next loop pass after a mid-stream cancel", async () => { + const f = makeTurnFixtures(); + const turn: PendingTurn = { zcodeSid: "sess_z", cancelled: false }; + // Pass 1: normal turn.started → loop keeps going. Pass 2 would block on a + // null poll, but the flag flips first (as cancel() would) → early return. + f.pollEvent.mockResolvedValueOnce({ + sessionId: "sess_z", + seq: 1, + type: "turn.started", + payload: {}, + }); + f.pollEvent.mockImplementation(async () => { + turn.cancelled = true; // cancel() lands while the loop polls + return null; + }); + + const resp = await runEventTurn( + f.server, + f.listener, + f.monitor, + f.differ, + f.cx, + "acp_a", + "m1", + turn, + false, + ); + + expect(resp).toEqual({ stopReason: "cancelled" }); + expect(f.sent).toEqual([{ method: "session/stop", params: { sessionId: "sess_z" } }]); + }); + + it("does not re-fire session/stop when the guard already sent one", async () => { + const f = makeTurnFixtures(); + const turn: PendingTurn = { zcodeSid: "sess_z", cancelled: true, stopSent: true }; + + const resp = await runEventTurn( + f.server, + f.listener, + f.monitor, + f.differ, + f.cx, + "acp_a", + "m1", + turn, + false, + ); + + expect(resp).toEqual({ stopReason: "cancelled" }); + expect(f.sent).toEqual([]); + }); +}); From c006f6dd17af1d36523b831174cee77db5e0d5da Mon Sep 17 00:00:00 2001 From: William Wang Date: Mon, 31 Aug 2026 17:17:56 +0800 Subject: [PATCH 09/14] fix: escalate cancel and preempt to session/close to kill the generation --- AGENTS.md | 19 +++++---- CHANGELOG.md | 22 ++++++---- src/handlers/session.ts | 76 +++++++++++++++++++++++++++++----- tests/session-cancel.test.ts | 17 ++++++-- tests/turn-attribution.test.ts | 10 ++++- 5 files changed, 113 insertions(+), 31 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index cc4b8d4..114014d 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -141,13 +141,18 @@ ZCode protocol types into ACP notifications directly — always translate. queue, entries) belongs to run.ts's external store passed via snapshot props. - **Aug-28 app-server build (still "0.16.5") ignores `session/stop`**: the RPC returns `{}` but the model stream runs to its natural end (verified by - raw-backend probe). Cancel must therefore be honoured bridge-side — the - turn loop returns `stopReason: "cancelled"` on the flag instead of waiting - for a terminal event, and a send after a recent cancel waits for the - backend to report idle (a mid-generation send is accepted as steer input - and silently dropped when the old turn ends). If a future build fixes the - stop, the early return stays correct (client cancel semantics); only the - drain gate becomes a no-op. + raw-backend probe; the backend's own log records `hadActivePrompt: false` — + the in-flight generation's abort controller is never registered). Cancel is + therefore honoured bridge-side, in three parts: the turn loop returns + `stopReason: "cancelled"` on the flag instead of waiting for a terminal + event; cancel/preempt escalate to `session/close`, which tears down the + resident runtime and kills the generation (the conversation persists — + `session/resume` restores it); and a send after a recent cancel settles the + backend first (drain gate: reload-after-close, or poll-until-idle — a + mid-generation send is accepted as steer input and silently dropped when + the old turn ends). If a future build fixes the stop, the early return + stays correct (client cancel semantics); close just costs a reload, and + the drain gate becomes a no-op. - **The backend rejects JSON-RPC frames carrying a `jsonrpc` field** (strict zod: "Unrecognized key: jsonrpc", code -32600). The bridge's backend client never sends one — keep it that way when hand-probing diff --git a/CHANGELOG.md b/CHANGELOG.md index db520f0..d82bf4f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -76,18 +76,24 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `esc`/stop now takes effect immediately. The Aug-28 app-server build (still reporting 0.16.5) accepts `session/stop` but no longer aborts the - in-flight model stream (verified with a raw-backend probe: the stream ran - ~10s past the stop to its natural end), so the bridge's turn loop waited - for a terminal event that only came after the full generation — every - client (REPL, mobile, editor) saw the stop "not work" while output kept - streaming. The turn loop now returns `stopReason: "cancelled"` the moment - the cancel flag is observed. + in-flight model stream: the backend's own log records every stop with + `hadActivePrompt: false` — the generation's abort controller is never + registered — so the stream ran ~10s past the stop to its natural end while + the turn loop waited for a terminal event that only came after the full + generation. Two changes: the turn loop now returns `stopReason: "cancelled"` + the moment the cancel flag is observed, and cancel/preempt escalate to + `session/close`, which tears down the resident runtime and kills the + generation outright (verified: stream halts within 0.5s, and the + conversation persists — `session/resume` restores it with the partial + reply in the history). - A follow-up prompt sent right after a cancel/preempt is no longer silently dropped. The same backend build accepts a mid-generation `session/send` as a steer and discards its input when the old turn finishes (verified: only one `turn.completed` ever arrives, for the old prompt). The bridge - now polls the session until the backend reports idle before sending — - with a visible `[上一个回复仍在生成,等待结束后发送…]` note — bounded at + now settles the backend before sending: with the close escalation the + probe fails fast into a session reload; on a backend that honours stop it + polls the projection until idle — a visible + `[上一个回复仍在生成,等待结束后发送…]` note explains the wait — bounded at 90s, still interruptible with `esc`, falling back to a direct send on timeout or probe failure. - Pressing ↓ with no completion menu open no longer zombifies the whole UI: diff --git a/src/handlers/session.ts b/src/handlers/session.ts index 9d33090..d0bb585 100644 --- a/src/handlers/session.ts +++ b/src/handlers/session.ts @@ -756,15 +756,18 @@ export async function prompt( const chunkMsgId = randomUUID(); - // Drain gate: a recent cancel/preempt means the backend is STILL - // generating — session/stop does not abort the model stream (verified - // against app-server 0.16.5) — and a send that lands mid-generation is - // accepted as a steer whose input the backend silently DROPS when the - // old turn finishes (verified: only one turn.completed ever arrives). - // Poll the projection until the backend reports idle so the user's - // follow-up actually runs; a visible chunk tells them why it waits. - // Bounded: on timeout (or a failed probe) send anyway — the steer-drop - // risk returns, but blocking the prompt forever is worse. + // Drain gate: a recent cancel/preempt means the backend side needs + // settling before the send. Two regimes: (a) a backend that honours + // session/stop — the abandoned turn is still generating, and a send + // that lands mid-generation is accepted as a steer whose input the + // backend silently DROPS when the old turn finishes (verified: only + // one turn.completed ever arrives) — so poll the projection until it + // reports idle; (b) the 0.16.5 reality — cancel() escalated to + // session/close, so the session is briefly NOT ACTIVE and must be + // reloaded before the send (pollOnce fails fast into that branch). + // A visible chunk tells the user why the send waits. Bounded: on + // timeout send anyway — the steer-drop risk returns, but blocking the + // prompt forever is worse. const DRAIN_TIMEOUT_MS = 90_000; const DRAIN_POLL_MS = 1000; const cancelledRecently = @@ -780,7 +783,19 @@ export async function prompt( return { stopReason: "cancelled" }; } const proj = await drainMonitor.pollOnce(); - if (!proj) break; // probe failed — don't block the send on it + if (!proj) { + // Probe failed — most likely the session was just closed by the + // cancel escalation (close tears down the runtime). Reload it so + // the send below doesn't die on "session is not active". + try { + await reloadBackendSession(server, params.sessionId, zcodeSid); + } catch (e) { + warn( + `drain gate: reload after close failed: ${e instanceof Error ? e.message : String(e)}`, + ); + } + break; + } if (proj.status === "idle") break; if (!noticed) { noticed = true; @@ -1002,6 +1017,7 @@ export async function cancel( // could leave the live one running. Each turn guards its own stopSent, so // multiple matching turns may each fire session/stop once — the backend // treats stop as idempotent, so the duplicate is harmless. + let closed = false; for (const [, turn] of server.pendingTurns) { if (turn.zcodeSid === zcodeSid) { turn.cancelled = true; @@ -1009,6 +1025,12 @@ export async function cancel( stopBackendTurn(server, zcodeSid); turn.stopSent = true; } + // session/stop is a protocol-level formality on 0.16.5 — the generation + // only dies when the resident runtime is closed. Once per cancel. + if (!closed) { + closeBackendSession(server, zcodeSid); + closed = true; + } // Record cancel time so a prompt arriving in the backend's ~20s // model-connection recovery window can fast-fail instead of hanging. server.lastCancelledAt.set(zcodeSid, Date.now()); @@ -1052,6 +1074,31 @@ function stopBackendTurn(server: ZcodeAcpServer, zcodeSid: string): void { } } +/** + * Hard-stop a backend turn by tearing down its resident runtime. + * + * `session/stop` returns `{}` but does NOT abort the model stream on + * app-server 0.16.5 (verified live: the stream ran 10s+ past the stop to its + * natural end, and the backend's own log shows `hadActivePrompt: false` — the + * in-flight generation's abort controller is never registered, so stop finds + * nothing to abort). `session/close` closes the runtime itself, which kills + * the generation immediately; the conversation is persisted in the backend's + * session store, so `session/resume` restores it (verified live: resume + * succeeds and the partial reply is in the history). Callers reload the + * session on next use — prompt()'s subscribe recovery and the drain gate's + * reload both handle the closed window. + */ +function closeBackendSession(server: ZcodeAcpServer, zcodeSid: string): void { + try { + server.ensureBackend().send("session/close", { sessionId: zcodeSid }); + log(` [stop] session/close fired for ${zcodeSid} (backend ignores session/stop)`); + } catch (e) { + log( + ` [stop] session/close send failed (ignored): ${e instanceof Error ? e.message : String(e)}`, + ); + } +} + /** * Serialize a per-session critical section. Each section awaits the previous * one's promise before running, so concurrent prompts for the same session @@ -1124,6 +1171,7 @@ export function preemptInFlightTurn( // would retry against a busy backend for 30s and fail. Each turn guards its // own stopSent; duplicate stops are idempotent on the backend. let found = false; + let closed = false; for (const [reqId, turn] of server.pendingTurns) { if (turn.zcodeSid !== zcodeSid || reqId === selfRequestId) continue; turn.cancelled = true; // signal the old turn to stop its retry loops @@ -1131,6 +1179,14 @@ export function preemptInFlightTurn( stopBackendTurn(server, zcodeSid); turn.stopSent = true; } + // The old generation only dies when the resident runtime is closed (the + // backend ignores session/stop) — otherwise this new prompt would wait + // out the ENTIRE remaining generation in the drain gate, and a send that + // lands mid-generation is silently dropped as a steer. Once per preempt. + if (!closed) { + closeBackendSession(server, zcodeSid); + closed = true; + } // Record cancel time so the prompt()'s send-retry can use the recovery // window as a hint (see session/send retry loop). server.lastCancelledAt.set(zcodeSid, Date.now()); diff --git a/tests/session-cancel.test.ts b/tests/session-cancel.test.ts index fc6cca2..9136c8e 100644 --- a/tests/session-cancel.test.ts +++ b/tests/session-cancel.test.ts @@ -38,7 +38,7 @@ function makeServer(turns: FakeTurn[], resolve: (sid: string) => string | undefi } describe("session/cancel handler", () => { - it("marks the matching turn cancelled and fires session/stop once", async () => { + it("marks the matching turn cancelled and fires session/stop + session/close", async () => { const turn: FakeTurn = { zcodeSid: "sess_z", cancelled: false, stopSent: false }; const { server, sent } = makeServer([turn], (sid) => (sid === "acp_a" ? "sess_z" : undefined)); @@ -46,7 +46,12 @@ describe("session/cancel handler", () => { expect(turn.cancelled).toBe(true); expect(turn.stopSent).toBe(true); - expect(sent).toEqual([{ method: "session/stop", params: { sessionId: "sess_z" } }]); + // stop is the protocol formality; close is what actually kills the + // generation (the backend ignores session/stop — verified 0.16.5). + expect(sent).toEqual([ + { method: "session/stop", params: { sessionId: "sess_z" } }, + { method: "session/close", params: { sessionId: "sess_z" } }, + ]); expect(server.lastCancelledAt.get("sess_z")).toBeGreaterThan(0); }); @@ -63,8 +68,12 @@ describe("session/cancel handler", () => { expect(old.cancelled).toBe(true); expect(live.cancelled).toBe(true); expect(other.cancelled).toBe(false); - // The already-stopped turn does not fire a second stop. - expect(sent).toEqual([{ method: "session/stop", params: { sessionId: "sess_z" } }]); + // The stale turn's stopSent guard skips the stop; close fires first (per + // cancel), then the live turn's stop. Both target the same session. + expect(sent).toEqual([ + { method: "session/close", params: { sessionId: "sess_z" } }, + { method: "session/stop", params: { sessionId: "sess_z" } }, + ]); }); it("no-ops for an unknown session id", async () => { diff --git a/tests/turn-attribution.test.ts b/tests/turn-attribution.test.ts index c3c4438..849cc65 100644 --- a/tests/turn-attribution.test.ts +++ b/tests/turn-attribution.test.ts @@ -223,8 +223,14 @@ describe("preemptInFlightTurn cancels ALL matching turns", () => { expect(pendingTurns.get(101)?.cancelled).toBe(true); expect(pendingTurns.get(102)?.cancelled).toBe(true); expect(pendingTurns.get(102)?.stopSent).toBe(true); - // stopBackendTurn fires once (stopSent guard dedupes across both turns). - expect(sends).toEqual([{ method: "session/stop", sid: "zs_1" }]); + // stopBackendTurn fires once (stopSent guard dedupes across both turns), + // then the close escalation kills the runtime so the generation actually + // ends (the backend ignores session/stop — verified 0.16.5). The stale + // entry's stopSent guard skips its stop, so close lands first. + expect(sends).toEqual([ + { method: "session/close", sid: "zs_1" }, + { method: "session/stop", sid: "zs_1" }, + ]); }); it("returns false when no other turn exists for the session", () => { From f85c482a37dd30c058a9fea719ef4c971a7b3615 Mon Sep 17 00:00:00 2001 From: William Wang Date: Mon, 31 Aug 2026 17:46:19 +0800 Subject: [PATCH 10/14] fix: stop turns via the official v4/command stop RPC --- AGENTS.md | 23 +++--- CHANGELOG.md | 31 ++++---- src/handlers/session.ts | 135 ++++++++++++++++++++------------- src/server.ts | 6 ++ tests/session-cancel.test.ts | 28 +++---- tests/turn-attribution.test.ts | 12 +-- 6 files changed, 133 insertions(+), 102 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 114014d..1d0016b 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -142,17 +142,18 @@ ZCode protocol types into ACP notifications directly — always translate. - **Aug-28 app-server build (still "0.16.5") ignores `session/stop`**: the RPC returns `{}` but the model stream runs to its natural end (verified by raw-backend probe; the backend's own log records `hadActivePrompt: false` — - the in-flight generation's abort controller is never registered). Cancel is - therefore honoured bridge-side, in three parts: the turn loop returns - `stopReason: "cancelled"` on the flag instead of waiting for a terminal - event; cancel/preempt escalate to `session/close`, which tears down the - resident runtime and kills the generation (the conversation persists — - `session/resume` restores it); and a send after a recent cancel settles the - backend first (drain gate: reload-after-close, or poll-until-idle — a - mid-generation send is accepted as steer input and silently dropped when - the old turn ends). If a future build fixes the stop, the early return - stays correct (client cancel semantics); close just costs a reload, and - the drain gate becomes a no-op. + the in-flight generation's abort controller is never registered). The + official desktop app never hits that path: its stop button sends a + `v4/command` RPC of type `stop` (`payload.expectedForegroundExecutionId` + optional), which asks the runtime to stop the active foreground execution + — found by grepping the app bundle. stopBackendTurn sends both: the + session/stop formality plus the v4 stop, which kills the generation + instantly (verified: `turn.completed` in 0.0s). Cancel is otherwise + bridge-side: the turn loop returns `stopReason: "cancelled"` on the flag, + and a send after a recent cancel settles the backend first (drain gate: + poll-until-idle, with a `session/close` escalation after a 5s grace if a + generation somehow survives both stops — a mid-generation send is accepted + as steer input and silently dropped when the old turn ends). - **The backend rejects JSON-RPC frames carrying a `jsonrpc` field** (strict zod: "Unrecognized key: jsonrpc", code -32600). The bridge's backend client never sends one — keep it that way when hand-probing diff --git a/CHANGELOG.md b/CHANGELOG.md index d82bf4f..262081c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -75,24 +75,27 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed - `esc`/stop now takes effect immediately. The Aug-28 app-server build - (still reporting 0.16.5) accepts `session/stop` but no longer aborts the - in-flight model stream: the backend's own log records every stop with - `hadActivePrompt: false` — the generation's abort controller is never - registered — so the stream ran ~10s past the stop to its natural end while - the turn loop waited for a terminal event that only came after the full - generation. Two changes: the turn loop now returns `stopReason: "cancelled"` - the moment the cancel flag is observed, and cancel/preempt escalate to - `session/close`, which tears down the resident runtime and kills the - generation outright (verified: stream halts within 0.5s, and the - conversation persists — `session/resume` restores it with the partial - reply in the history). + (still reporting 0.16.5) accepts `session/stop` but never aborts the + in-flight model stream — its own log records every stop with + `hadActivePrompt: false`, i.e. the generation's abort controller is never + registered, so the stream ran ~10s past the stop to its natural end while + the turn loop waited for a terminal event. Digging through the desktop + app's bundle revealed the stop path the official client actually uses: a + `v4/command` RPC of type `stop` that asks the runtime to stop the active + foreground execution (not the broken `session/stop`). The bridge now sends + that v4 stop alongside `session/stop` — verified live: the generation dies + the instant the command lands (`turn.completed` in 0.0s, vs +39.7s natural + drift before). The turn loop also returns `stopReason: "cancelled"` at + once instead of waiting for a terminal event. - A follow-up prompt sent right after a cancel/preempt is no longer silently dropped. The same backend build accepts a mid-generation `session/send` as a steer and discards its input when the old turn finishes (verified: only one `turn.completed` ever arrives, for the old prompt). The bridge - now settles the backend before sending: with the close escalation the - probe fails fast into a session reload; on a backend that honours stop it - polls the projection until idle — a visible + now settles the backend before sending: with the v4 stop the probe sees + idle immediately; on a backend that honours `session/stop` it polls the + projection until idle; if a generation somehow survives both stops, a + `session/close` escalation after a 5s grace tears down the runtime (the + probe then fails into a session reload). A visible `[上一个回复仍在生成,等待结束后发送…]` note explains the wait — bounded at 90s, still interruptible with `esc`, falling back to a direct send on timeout or probe failure. diff --git a/src/handlers/session.ts b/src/handlers/session.ts index d0bb585..f8ca0e6 100644 --- a/src/handlers/session.ts +++ b/src/handlers/session.ts @@ -738,7 +738,7 @@ export async function prompt( // reconcile the differ baseline so the retried turn's new messages // aren't treated as already-seen, surface a retry hint, then back off. if (turn.cancelled) { - stopBackendTurn(server, zcodeSid); + stopBackendTurn(server, zcodeSid, turn.foregroundExecutionId); return { stopReason: "cancelled" }; } differ.markSeen(await fetchMessages(server, zcodeSid)); @@ -757,19 +757,20 @@ export async function prompt( const chunkMsgId = randomUUID(); // Drain gate: a recent cancel/preempt means the backend side needs - // settling before the send. Two regimes: (a) a backend that honours - // session/stop — the abandoned turn is still generating, and a send - // that lands mid-generation is accepted as a steer whose input the - // backend silently DROPS when the old turn finishes (verified: only - // one turn.completed ever arrives) — so poll the projection until it - // reports idle; (b) the 0.16.5 reality — cancel() escalated to - // session/close, so the session is briefly NOT ACTIVE and must be - // reloaded before the send (pollOnce fails fast into that branch). - // A visible chunk tells the user why the send waits. Bounded: on - // timeout send anyway — the steer-drop risk returns, but blocking the - // prompt forever is worse. + // settling before the send. Primary path: stopBackendTurn's v4/command + // stop kills the generation at once, so the first probe here already + // sees idle. Fallbacks: on a backend that honours session/stop we poll + // the projection until idle (a send that lands mid-generation is + // accepted as a steer whose input the backend silently DROPS when the + // old turn finishes); if the generation is STILL running after a grace + // period — both stops ignored — escalate to session/close, which tears + // down the runtime and kills it outright (probe then fails into the + // reload branch). A visible chunk tells the user why the send waits. + // Bounded: on timeout send anyway — the steer-drop risk returns, but + // blocking the prompt forever is worse. const DRAIN_TIMEOUT_MS = 90_000; const DRAIN_POLL_MS = 1000; + const V4_STOP_ESCALATE_MS = 5_000; const cancelledRecently = server.lastCancelledAt.get(zcodeSid) !== undefined && Date.now() - server.lastCancelledAt.get(zcodeSid)! < DRAIN_TIMEOUT_MS; @@ -777,16 +778,17 @@ export async function prompt( const drainMonitor = new TurnMonitor(backend, zcodeSid, () => server.nextId()); const drainT0 = Date.now(); let noticed = false; + let escalated = false; while (Date.now() - drainT0 < DRAIN_TIMEOUT_MS) { if (turn.cancelled) { - stopBackendTurn(server, zcodeSid); + stopBackendTurn(server, zcodeSid, turn.foregroundExecutionId); return { stopReason: "cancelled" }; } const proj = await drainMonitor.pollOnce(); if (!proj) { // Probe failed — most likely the session was just closed by the - // cancel escalation (close tears down the runtime). Reload it so - // the send below doesn't die on "session is not active". + // escalation (close tears down the runtime). Reload it so the + // send below doesn't die on "session is not active". try { await reloadBackendSession(server, params.sessionId, zcodeSid); } catch (e) { @@ -797,6 +799,14 @@ export async function prompt( break; } if (proj.status === "idle") break; + if ( + proj.status === "running" && + !escalated && + Date.now() - drainT0 > V4_STOP_ESCALATE_MS + ) { + escalated = true; + closeBackendSession(server, zcodeSid); + } if (!noticed) { noticed = true; await sendTextChunk( @@ -828,7 +838,7 @@ export async function prompt( let sendAttempt = 0; while (true) { if (turn.cancelled) { - stopBackendTurn(server, zcodeSid); + stopBackendTurn(server, zcodeSid, turn.foregroundExecutionId); return { stopReason: "cancelled" }; } sendAttempt++; @@ -843,7 +853,7 @@ export async function prompt( if (expectBusy) { await sleep(SEND_RETRY_INTERVAL_MS); if (turn.cancelled) { - stopBackendTurn(server, zcodeSid); + stopBackendTurn(server, zcodeSid, turn.foregroundExecutionId); return { stopReason: "cancelled" }; } } @@ -1015,22 +1025,15 @@ export async function cancel( // prior turn is still finalising, pendingTurns holds both it and any newer // prompt waiting on the backend's prompt lock; breaking on the first match // could leave the live one running. Each turn guards its own stopSent, so - // multiple matching turns may each fire session/stop once — the backend - // treats stop as idempotent, so the duplicate is harmless. - let closed = false; + // multiple matching turns may each fire the stop pair once — the backend + // treats both as idempotent, so the duplicate is harmless. for (const [, turn] of server.pendingTurns) { if (turn.zcodeSid === zcodeSid) { turn.cancelled = true; if (!turn.stopSent) { - stopBackendTurn(server, zcodeSid); + stopBackendTurn(server, zcodeSid, turn.foregroundExecutionId); turn.stopSent = true; } - // session/stop is a protocol-level formality on 0.16.5 — the generation - // only dies when the resident runtime is closed. Once per cancel. - if (!closed) { - closeBackendSession(server, zcodeSid); - closed = true; - } // Record cancel time so a prompt arriving in the backend's ~20s // model-connection recovery window can fast-fail instead of hanging. server.lastCancelledAt.set(zcodeSid, Date.now()); @@ -1064,7 +1067,11 @@ class TurnFailedError extends Error { * backend's prompt lock releases when ITS finalisation completes — that, * not any bridge-side signal, is what the next prompt's send-retry waits on. */ -function stopBackendTurn(server: ZcodeAcpServer, zcodeSid: string): void { +function stopBackendTurn( + server: ZcodeAcpServer, + zcodeSid: string, + foregroundExecutionId?: string, +): void { try { server.ensureBackend().send("session/stop", { sessionId: zcodeSid }); } catch (e) { @@ -1072,21 +1079,45 @@ function stopBackendTurn(server: ZcodeAcpServer, zcodeSid: string): void { ` [stop] session/stop send failed (ignored): ${e instanceof Error ? e.message : String(e)}`, ); } + // The official stop path (this is what the desktop app's stop button uses — + // found in the app bundle): a v4 command that asks the runtime to stop the + // active foreground execution. session/stop alone is a no-op on the Aug-28 + // app-server (its abort controller is never registered; backend log shows + // `hadActivePrompt: false`), while this kills the generation instantly — + // verified: turn.completed arrives the same instant the command lands. + // expectedForegroundExecutionId is optional and omitted: cancelling targets + // whatever is currently foreground for the session. + try { + server.ensureBackend().send("v4/command", { + commandId: randomUUID(), + clientId: "zcode-acp-server", + sessionId: zcodeSid, + type: "stop", + payload: foregroundExecutionId + ? { expectedForegroundExecutionId: foregroundExecutionId } + : {}, + issuedAt: Date.now(), + }); + log(` [stop] v4/command stop sent for ${zcodeSid}`); + } catch (e) { + log( + ` [stop] v4/command stop send failed (ignored): ${e instanceof Error ? e.message : String(e)}`, + ); + } } /** - * Hard-stop a backend turn by tearing down its resident runtime. + * Last-resort stop: tear down the session's resident runtime, killing any + * generation that survived the stop pair (session/stop + v4/command stop). * - * `session/stop` returns `{}` but does NOT abort the model stream on - * app-server 0.16.5 (verified live: the stream ran 10s+ past the stop to its - * natural end, and the backend's own log shows `hadActivePrompt: false` — the - * in-flight generation's abort controller is never registered, so stop finds - * nothing to abort). `session/close` closes the runtime itself, which kills - * the generation immediately; the conversation is persisted in the backend's - * session store, so `session/resume` restores it (verified live: resume - * succeeds and the partial reply is in the history). Callers reload the - * session on next use — prompt()'s subscribe recovery and the drain gate's - * reload both handle the closed window. + * The primary path is stopBackendTurn's v4/command stop — the official one — + * which kills the generation instantly. This close is the escalation when + * both stops are ignored (drain gate, 5s grace): `session/close` closes the + * runtime itself, which kills the generation immediately; the conversation + * is persisted in the backend's session store, so `session/resume` restores + * it (verified live: resume succeeds and the partial reply is in the + * history). Callers reload the session on next use — prompt()'s subscribe + * recovery and the drain gate's reload both handle the closed window. */ function closeBackendSession(server: ZcodeAcpServer, zcodeSid: string): void { try { @@ -1171,22 +1202,13 @@ export function preemptInFlightTurn( // would retry against a busy backend for 30s and fail. Each turn guards its // own stopSent; duplicate stops are idempotent on the backend. let found = false; - let closed = false; for (const [reqId, turn] of server.pendingTurns) { if (turn.zcodeSid !== zcodeSid || reqId === selfRequestId) continue; turn.cancelled = true; // signal the old turn to stop its retry loops if (!turn.stopSent) { - stopBackendTurn(server, zcodeSid); + stopBackendTurn(server, zcodeSid, turn.foregroundExecutionId); turn.stopSent = true; } - // The old generation only dies when the resident runtime is closed (the - // backend ignores session/stop) — otherwise this new prompt would wait - // out the ENTIRE remaining generation in the drain gate, and a send that - // lands mid-generation is silently dropped as a steer. Once per preempt. - if (!closed) { - closeBackendSession(server, zcodeSid); - closed = true; - } // Record cancel time so the prompt()'s send-retry can use the recovery // window as a hint (see session/send retry loop). server.lastCancelledAt.set(zcodeSid, Date.now()); @@ -1544,7 +1566,7 @@ export async function runEventTurn( // to no turn listener, and the next turn's turn-attribution gate // discards any residue that slipped into the queue meanwhile. if (!turn.stopSent) { - stopBackendTurn(server, turn.zcodeSid); + stopBackendTurn(server, turn.zcodeSid, turn.foregroundExecutionId); turn.stopSent = true; } return { stopReason: "cancelled" }; @@ -1607,7 +1629,7 @@ export async function runEventTurn( await sendTextChunk(cx, acpSid, reply.text, chunkMsgId); } else if (!emittedOutput) { // No text and no output → suspected failure. - stopBackendTurn(server, turn.zcodeSid); + stopBackendTurn(server, turn.zcodeSid, turn.foregroundExecutionId); throw new RequestError(-32603, "turn produced no output"); } } @@ -1662,6 +1684,13 @@ export async function runEventTurn( if (shouldDropEventForTurnAttribution(ev, translator.turnStarted, gateArmed)) { continue; } + if (ev.type === "turn.started") { + // Remember the runtime's foreground execution id: the v4/command stop + // (see stopBackendTurn) targets it if the user cancels mid-turn. + const fge = (ev.payload as { foregroundExecutionId?: string } | undefined) + ?.foregroundExecutionId; + if (fge) turn.foregroundExecutionId = fge; + } const internalEvents = translator.translate(ev); // Capture the turn-start timestamp for the thinking-phase hint above. // Done after translate so the flag flip on the turn.started event is @@ -1735,7 +1764,7 @@ export async function runEventTurn( } if (translator.turnFailed) { // Best-effort stop in case the failed turn left a residual lock. - stopBackendTurn(server, turn.zcodeSid); + stopBackendTurn(server, turn.zcodeSid, turn.foregroundExecutionId); // Throw a TurnFailedError carrying the structured error so the caller // (prompt's retry loop) can classify transient vs fatal. The error // message is formatted for display when it ultimately reaches the user. @@ -1789,7 +1818,7 @@ export async function runEventTurn( } // 120s no progress: abandon. - stopBackendTurn(server, turn.zcodeSid); + stopBackendTurn(server, turn.zcodeSid, turn.foregroundExecutionId); return { stopReason: "max_turn_requests" }; } diff --git a/src/server.ts b/src/server.ts index 4e8016b..de513bc 100644 --- a/src/server.ts +++ b/src/server.ts @@ -35,6 +35,12 @@ export interface PendingTurn { cancelled: boolean; /** Set once session/stop has been fired for this turn, to avoid re-sending. */ stopSent?: boolean; + /** + * Foreground execution id from the backend's `turn.started` payload. The + * v4/command stop targets it — session/stop alone is ignored by the Aug-28 + * app-server (its abort controller is never registered; see AGENTS.md). + */ + foregroundExecutionId?: string; /** * Set when the turn was ended by the stall-recovery heuristic (backend * reported idle after a silence) rather than a real turn.completed event. diff --git a/tests/session-cancel.test.ts b/tests/session-cancel.test.ts index 9136c8e..1206ff8 100644 --- a/tests/session-cancel.test.ts +++ b/tests/session-cancel.test.ts @@ -38,7 +38,7 @@ function makeServer(turns: FakeTurn[], resolve: (sid: string) => string | undefi } describe("session/cancel handler", () => { - it("marks the matching turn cancelled and fires session/stop + session/close", async () => { + it("marks the matching turn cancelled and fires session/stop + v4 stop", async () => { const turn: FakeTurn = { zcodeSid: "sess_z", cancelled: false, stopSent: false }; const { server, sent } = makeServer([turn], (sid) => (sid === "acp_a" ? "sess_z" : undefined)); @@ -46,12 +46,11 @@ describe("session/cancel handler", () => { expect(turn.cancelled).toBe(true); expect(turn.stopSent).toBe(true); - // stop is the protocol formality; close is what actually kills the - // generation (the backend ignores session/stop — verified 0.16.5). - expect(sent).toEqual([ - { method: "session/stop", params: { sessionId: "sess_z" } }, - { method: "session/close", params: { sessionId: "sess_z" } }, - ]); + // session/stop is the protocol formality; the v4/command stop (the + // official app's own path) is what actually kills the generation on the + // Aug-28 app-server, which ignores session/stop entirely. + expect(sent.map((s) => s.method)).toEqual(["session/stop", "v4/command"]); + expect(sent[1]?.params).toMatchObject({ sessionId: "sess_z", type: "stop" }); expect(server.lastCancelledAt.get("sess_z")).toBeGreaterThan(0); }); @@ -68,12 +67,9 @@ describe("session/cancel handler", () => { expect(old.cancelled).toBe(true); expect(live.cancelled).toBe(true); expect(other.cancelled).toBe(false); - // The stale turn's stopSent guard skips the stop; close fires first (per - // cancel), then the live turn's stop. Both target the same session. - expect(sent).toEqual([ - { method: "session/close", params: { sessionId: "sess_z" } }, - { method: "session/stop", params: { sessionId: "sess_z" } }, - ]); + // The stale turn's stopSent guard skips its stop pair entirely; the live + // turn's fires the session/stop + v4/command pair once. + expect(sent.map((s) => s.method)).toEqual(["session/stop", "v4/command"]); }); it("no-ops for an unknown session id", async () => { @@ -143,8 +139,8 @@ describe("runEventTurn: cancel exits immediately (backend ignores session/stop)" ); expect(resp).toEqual({ stopReason: "cancelled" }); - // Guard-fired stop (stopSent was false), and NOT a single event consumed. - expect(f.sent).toEqual([{ method: "session/stop", params: { sessionId: "sess_z" } }]); + // Guard-fired stop pair (stopSent was false), and NOT a single event consumed. + expect(f.sent.map((s) => s.method)).toEqual(["session/stop", "v4/command"]); expect(f.pollEvent).not.toHaveBeenCalled(); }); @@ -177,7 +173,7 @@ describe("runEventTurn: cancel exits immediately (backend ignores session/stop)" ); expect(resp).toEqual({ stopReason: "cancelled" }); - expect(f.sent).toEqual([{ method: "session/stop", params: { sessionId: "sess_z" } }]); + expect(f.sent.map((s) => s.method)).toEqual(["session/stop", "v4/command"]); }); it("does not re-fire session/stop when the guard already sent one", async () => { diff --git a/tests/turn-attribution.test.ts b/tests/turn-attribution.test.ts index 849cc65..7b4a09f 100644 --- a/tests/turn-attribution.test.ts +++ b/tests/turn-attribution.test.ts @@ -223,14 +223,10 @@ describe("preemptInFlightTurn cancels ALL matching turns", () => { expect(pendingTurns.get(101)?.cancelled).toBe(true); expect(pendingTurns.get(102)?.cancelled).toBe(true); expect(pendingTurns.get(102)?.stopSent).toBe(true); - // stopBackendTurn fires once (stopSent guard dedupes across both turns), - // then the close escalation kills the runtime so the generation actually - // ends (the backend ignores session/stop — verified 0.16.5). The stale - // entry's stopSent guard skips its stop, so close lands first. - expect(sends).toEqual([ - { method: "session/close", sid: "zs_1" }, - { method: "session/stop", sid: "zs_1" }, - ]); + // stopBackendTurn fires once (stopSent guard dedupes across both turns): + // the session/stop formality plus the v4/command stop that actually kills + // the generation (the backend ignores session/stop — verified 0.16.5). + expect(sends.map((s) => s.method)).toEqual(["session/stop", "v4/command"]); }); it("returns false when no other turn exists for the session", () => { From e5cdc3577af72e99b32beb04f53e687303144c08 Mon Sep 17 00:00:00 2001 From: William Wang Date: Mon, 31 Aug 2026 19:20:08 +0800 Subject: [PATCH 11/14] fix: harden the cancel drain gate (resubscribe, re-baseline, steer report) --- AGENTS.md | 8 +- CHANGELOG.md | 9 +- src/handlers/session.ts | 190 +++++++++++++++++++++++------------ tests/drain-gate.test.ts | 134 ++++++++++++++++++++++++ tests/session-cancel.test.ts | 73 ++++++++++++++ 5 files changed, 349 insertions(+), 65 deletions(-) create mode 100644 tests/drain-gate.test.ts diff --git a/AGENTS.md b/AGENTS.md index 1d0016b..3c8674e 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -153,7 +153,13 @@ ZCode protocol types into ACP notifications directly — always translate. and a send after a recent cancel settles the backend first (drain gate: poll-until-idle, with a `session/close` escalation after a 5s grace if a generation somehow survives both stops — a mid-generation send is accepted - as steer input and silently dropped when the old turn ends). + as steer input and silently dropped when the old turn ends; the + `turn.steerQueued` event proves the swallow and the bridge reports it at + once instead of hanging). After a close-escalation reload the drain gate + must resubscribe the event stream (the reload revives the session but not + its push — the next turn would run deaf) and re-baseline the projection + differ (the abandoned turn committed messages while waiting — a stale + baseline replays that residue as the next reply). - **The backend rejects JSON-RPC frames carrying a `jsonrpc` field** (strict zod: "Unrecognized key: jsonrpc", code -32600). The bridge's backend client never sends one — keep it that way when hand-probing diff --git a/CHANGELOG.md b/CHANGELOG.md index 262081c..b00953d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -98,7 +98,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 probe then fails into a session reload). A visible `[上一个回复仍在生成,等待结束后发送…]` note explains the wait — bounded at 90s, still interruptible with `esc`, falling back to a direct send on - timeout or probe failure. + timeout or probe failure. Two edge paths found in review are also closed: + after a close-escalation reload the bridge re-subscribes the event stream + (the reload revives the session but not its push — without this the next + turn runs deaf until the watchdog) and re-baselines the projection differ + so the cancelled turn's residue is never replayed as the next reply; and a + send that does land mid-generation is reported at once via the backend's + `turn.steerQueued` event (`[消息被并入仍在生成的回合,将被丢弃,请重新发送]`) + instead of hanging silently until the 120s watchdog. - Pressing ↓ with no completion menu open no longer zombifies the whole UI: the setState updater dereferenced a null menu during render, unmounting React's tree under ink without any crash signal (found by review, diff --git a/src/handlers/session.ts b/src/handlers/session.ts index f8ca0e6..035e289 100644 --- a/src/handlers/session.ts +++ b/src/handlers/session.ts @@ -757,67 +757,21 @@ export async function prompt( const chunkMsgId = randomUUID(); // Drain gate: a recent cancel/preempt means the backend side needs - // settling before the send. Primary path: stopBackendTurn's v4/command - // stop kills the generation at once, so the first probe here already - // sees idle. Fallbacks: on a backend that honours session/stop we poll - // the projection until idle (a send that lands mid-generation is - // accepted as a steer whose input the backend silently DROPS when the - // old turn finishes); if the generation is STILL running after a grace - // period — both stops ignored — escalate to session/close, which tears - // down the runtime and kills it outright (probe then fails into the - // reload branch). A visible chunk tells the user why the send waits. - // Bounded: on timeout send anyway — the steer-drop risk returns, but - // blocking the prompt forever is worse. - const DRAIN_TIMEOUT_MS = 90_000; - const DRAIN_POLL_MS = 1000; - const V4_STOP_ESCALATE_MS = 5_000; + // settling before the send — see drainBackendAfterCancel. const cancelledRecently = server.lastCancelledAt.get(zcodeSid) !== undefined && - Date.now() - server.lastCancelledAt.get(zcodeSid)! < DRAIN_TIMEOUT_MS; + Date.now() - server.lastCancelledAt.get(zcodeSid)! < DRAIN_WINDOW_MS; if (cancelledRecently) { - const drainMonitor = new TurnMonitor(backend, zcodeSid, () => server.nextId()); - const drainT0 = Date.now(); - let noticed = false; - let escalated = false; - while (Date.now() - drainT0 < DRAIN_TIMEOUT_MS) { - if (turn.cancelled) { - stopBackendTurn(server, zcodeSid, turn.foregroundExecutionId); - return { stopReason: "cancelled" }; - } - const proj = await drainMonitor.pollOnce(); - if (!proj) { - // Probe failed — most likely the session was just closed by the - // escalation (close tears down the runtime). Reload it so the - // send below doesn't die on "session is not active". - try { - await reloadBackendSession(server, params.sessionId, zcodeSid); - } catch (e) { - warn( - `drain gate: reload after close failed: ${e instanceof Error ? e.message : String(e)}`, - ); - } - break; - } - if (proj.status === "idle") break; - if ( - proj.status === "running" && - !escalated && - Date.now() - drainT0 > V4_STOP_ESCALATE_MS - ) { - escalated = true; - closeBackendSession(server, zcodeSid); - } - if (!noticed) { - noticed = true; - await sendTextChunk( - cx, - params.sessionId, - "[上一个回复仍在生成,等待结束后发送…]", - randomUUID(), - ); - } - await sleep(DRAIN_POLL_MS); - } + const drained = await drainBackendAfterCancel(server, { + acpSid: params.sessionId, + zcodeSid, + turn, + listener, + monitor: new TurnMonitor(backend, zcodeSid, () => server.nextId()), + differ, + cx, + }); + if (drained === "cancelled") return { stopReason: "cancelled" }; } // Send the prompt, retrying while the backend reports it's still busy. @@ -970,6 +924,11 @@ export async function prompt( * abandoned turn may still be streaming its finalisation into the backend). */ const CANCEL_RESIDUE_WINDOW_MS = 120_000; +/** How long after a cancel/preempt a new prompt still runs the drain gate + * (drainBackendAfterCancel) before sending — same bound as the drain wait + * itself, so the gate never waits twice its window. */ +const DRAIN_WINDOW_MS = 90_000; + /** * `session/set_config_option` → dispatch model/mode/thought and emit the * resulting config_option_update (+ current_mode_update for mode). @@ -1085,8 +1044,10 @@ function stopBackendTurn( // app-server (its abort controller is never registered; backend log shows // `hadActivePrompt: false`), while this kills the generation instantly — // verified: turn.completed arrives the same instant the command lands. - // expectedForegroundExecutionId is optional and omitted: cancelling targets - // whatever is currently foreground for the session. + // expectedForegroundExecutionId is passed when known — it is captured from + // the turn's own turn.started, so it names the execution that is foreground + // at cancel time, letting the backend guard against stopping a newer one. + // Omitted when unknown, targeting whatever is currently foreground. try { server.ensureBackend().send("v4/command", { commandId: randomUUID(), @@ -1130,6 +1091,93 @@ function closeBackendSession(server: ZcodeAcpServer, zcodeSid: string): void { } } +/** Dependencies of drainBackendAfterCancel, injectable for tests. */ +interface DrainDeps { + acpSid: string; + zcodeSid: string; + turn: PendingTurn; + listener: EventStreamListener; + monitor: TurnMonitor; + differ: ProjectionDiffer; + cx: acp.AgentContext; + /** Test hook: override the close-escalation grace (default 5s). */ + escalateAfterMs?: number; +} + +/** + * Drain gate: a recent cancel/preempt means the backend side needs settling + * before the next send. Primary path: stopBackendTurn's v4/command stop kills + * the generation at once, so the first probe here already sees idle. + * Fallbacks: on a backend that honours session/stop we poll the projection + * until idle (a send that lands mid-generation is accepted as a steer whose + * input the backend silently DROPS when the old turn finishes); if the + * generation is STILL running after a grace period — both stops ignored — + * escalate to session/close, which tears down the runtime and kills it + * outright (the probe then fails into the reload branch). A visible chunk + * tells the user why the send waits. Bounded: on timeout send anyway — the + * steer-drop risk returns (the turn.steerQueued guard in runEventTurn reports + * it), but blocking the prompt forever is worse. + * + * Two post-drain repairs, both mirroring established patterns (prompt's + * eviction recovery / transient-retry re-baseline): + * - resubscribe: session/close killed the runtime this prompt subscribed to; + * the reload revives the session but not the event push, so re-arm it — + * without resubscribe the next turn runs deaf (no events at all, and stall + * recovery can't engage because it needs turn.started). + * - re-baseline: the abandoned turn committed messages to the session history + * while we waited (and close persisted its partial output); without markSeen + * the completion diff replays that residue as this turn's output. + * + * Returns "cancelled" when the turn was flagged cancelled during the drain + * (stop pair fired; caller resolves session/prompt at once). + */ +export async function drainBackendAfterCancel( + server: ZcodeAcpServer, + deps: DrainDeps, +): Promise<"cancelled" | "drained"> { + const { acpSid, zcodeSid, turn, listener, monitor, differ, cx } = deps; + const DRAIN_TIMEOUT_MS = 90_000; + const DRAIN_POLL_MS = 1000; + const escalateAfterMs = deps.escalateAfterMs ?? 5_000; + const drainT0 = Date.now(); + let noticed = false; + let escalated = false; + while (Date.now() - drainT0 < DRAIN_TIMEOUT_MS) { + if (turn.cancelled) { + stopBackendTurn(server, zcodeSid, turn.foregroundExecutionId); + return "cancelled"; + } + const proj = await monitor.pollOnce(); + if (!proj) { + // Probe failed — most likely the session was just closed by the + // escalation (close tears down the runtime). Reload it so the send + // below doesn't die on "session is not active", then re-arm the event + // push (see docstring). + try { + await reloadBackendSession(server, acpSid, zcodeSid); + await listener.resubscribe(() => server.nextId()); + } catch (e) { + warn( + `drain gate: reload after close failed: ${e instanceof Error ? e.message : String(e)}`, + ); + } + break; + } + if (proj.status === "idle") break; + if (proj.status === "running" && !escalated && Date.now() - drainT0 > escalateAfterMs) { + escalated = true; + closeBackendSession(server, zcodeSid); + } + if (!noticed) { + noticed = true; + await sendTextChunk(cx, acpSid, "[上一个回复仍在生成,等待结束后发送…]", randomUUID()); + } + await sleep(DRAIN_POLL_MS); + } + differ.markSeen(await fetchMessages(server, zcodeSid)); + return "drained"; +} + /** * Serialize a per-session critical section. Each section awaits the previous * one's promise before running, so concurrent prompts for the same session @@ -1661,6 +1709,22 @@ export async function runEventTurn( } lastProgress = Date.now(); + // Steer-swallow guard: a send accepted while the previous turn is still + // generating is queued as steer input, which the backend silently DROPS + // when that turn ends — no new turn ever starts (no turn.started), and + // the attribution gate below would discard the steerQueued event like + // any other residue, leaving this prompt to hang until the 120s watchdog + // with the message lost. turn.steerQueued is definitive proof of the + // swallow: report it at once so the user can resend immediately. + if (ev.type === "turn.steerQueued" && !translator.turnStarted && gateArmed) { + await sendTextChunk( + cx, + acpSid, + "[消息被并入仍在生成的回合,将被丢弃,请重新发送]", + chunkMsgId, + ); + return { stopReason: "max_turn_requests" }; + } // Turn-attribution gate: before this turn's own turn.started arrives, any // event is leftover from a prior turn (cancelled/preempted but still // finalising) that landed in the queue while send was retrying on a busy @@ -1678,9 +1742,9 @@ export async function runEventTurn( // returned — a prompt sent in that window reaches subscribe while the // residue is still arriving. Backend serialisation bounds the exposure: // a send is accepted only after the prior turn released the lock, so the - // residue can only arrive BEFORE this turn's turn.started. (In the rare - // steered-into backend-owned turn, where no turn.started ever arrives, - // the completion diff replays the dropped text at turn completion.) + // residue can only arrive BEFORE this turn's turn.started. (A send that + // lands mid-generation instead is a steer whose input is dropped — the + // turn.steerQueued guard above reports that at once.) if (shouldDropEventForTurnAttribution(ev, translator.turnStarted, gateArmed)) { continue; } diff --git a/tests/drain-gate.test.ts b/tests/drain-gate.test.ts new file mode 100644 index 0000000..1853949 --- /dev/null +++ b/tests/drain-gate.test.ts @@ -0,0 +1,134 @@ +/** + * Tests for drainBackendAfterCancel — the drain gate prompt() runs after a + * recent cancel/preempt — and its two post-drain repairs: + * - resubscribe after the close-escalation reload: the subscription died with + * the closed runtime; without re-arming, the next turn is deaf (no events + * at all, and stall recovery can't engage because it needs turn.started), + * - differ re-baseline: the abandoned turn committed messages to the session + * history while we waited; without markSeen the completion diff replays + * that residue as the next reply's output. + */ + +import { describe, expect, it, vi } from "vitest"; + +import type { TurnMonitor } from "../src/backend/listener.js"; +import { drainBackendAfterCancel } from "../src/handlers/session.js"; +import type { PendingTurn, ZcodeAcpServer } from "../src/server.js"; + +const fetchMessagesMock = vi.hoisted(() => vi.fn(async () => [])); + +vi.mock("../src/handlers/replay.js", async (importOriginal) => { + const actual = await importOriginal(); + return { ...actual, fetchMessages: fetchMessagesMock }; +}); + +interface DrainFixtures { + server: ZcodeAcpServer; + sent: Array<{ method: string; params: unknown }>; + pollOnce: ReturnType; + listener: { resubscribe: ReturnType }; + differ: { markSeen: ReturnType }; + cx: { notify: ReturnType }; + deps: Parameters[1]; +} + +function makeFixtures(turn: PendingTurn, escalateAfterMs = 0): DrainFixtures { + const sent: Array<{ method: string; params: unknown }> = []; + const server = { + ensureBackend: () => ({ + send: (method: string, params: unknown) => { + sent.push({ method, params }); + }, + request: async () => ({ result: {} }), // session/resume for the reload + }), + nextId: () => 1, + sessionCwds: new Map(), + markBackendLoaded: () => {}, + } as unknown as ZcodeAcpServer; + const pollOnce = vi.fn(); + const listener = { resubscribe: vi.fn(async () => true) }; + const differ = { markSeen: vi.fn() }; + const cx = { notify: vi.fn().mockResolvedValue(undefined) }; + const deps = { + acpSid: "acp_a", + zcodeSid: "sess_z", + turn, + listener, + monitor: { pollOnce } as unknown as TurnMonitor, + differ, + cx, + escalateAfterMs, + } as unknown as Parameters[1]; + return { + server, + sent, + pollOnce, + listener: listener as unknown as DrainFixtures["listener"], + differ: differ as unknown as DrainFixtures["differ"], + cx: cx as unknown as DrainFixtures["cx"], + deps, + }; +} + +describe("drainBackendAfterCancel", () => { + it("returns drained and re-baselines when the first probe already sees idle", async () => { + const f = makeFixtures({ zcodeSid: "sess_z", cancelled: false }); + f.pollOnce.mockResolvedValue({ status: "idle" }); + + const result = await drainBackendAfterCancel(f.server, f.deps); + + expect(result).toBe("drained"); + // Nothing to settle: no stop/close, no resubscribe. + expect(f.sent).toEqual([]); + expect(f.listener.resubscribe).not.toHaveBeenCalled(); + // Re-baseline always runs: the abandoned turn may have committed messages + // between the prompt's own baseline and this probe. + expect(f.differ.markSeen).toHaveBeenCalledTimes(1); + expect(fetchMessagesMock).toHaveBeenCalledWith(f.server, "sess_z"); + }); + + it("emits one wait note and keeps polling until idle (no close below the grace)", async () => { + const f = makeFixtures( + { zcodeSid: "sess_z", cancelled: false }, + 60_000, // grace far away: the escalation must not fire + ); + f.pollOnce.mockResolvedValueOnce({ status: "running" }).mockResolvedValue({ status: "idle" }); + + const result = await drainBackendAfterCancel(f.server, f.deps); + + expect(result).toBe("drained"); + expect(f.sent).toEqual([]); + const notes = f.cx.notify.mock.calls.filter((c) => + JSON.stringify(c).includes("等待结束后发送"), + ); + expect(notes).toHaveLength(1); + }); + + it("escalates to session/close, reloads, and re-arms the subscription when the probe dies", async () => { + const f = makeFixtures({ zcodeSid: "sess_z", cancelled: false }); + f.pollOnce + .mockResolvedValueOnce({ status: "running" }) + .mockResolvedValueOnce({ status: "running" }) + .mockResolvedValue(null); // probe fails: close tore down the runtime + + const result = await drainBackendAfterCancel(f.server, f.deps); + + expect(result).toBe("drained"); + expect(f.sent).toEqual([{ method: "session/close", params: { sessionId: "sess_z" } }]); + // The reload alone leaves the turn deaf — resubscribe must re-arm the push. + expect(f.listener.resubscribe).toHaveBeenCalledTimes(1); + expect(f.differ.markSeen).toHaveBeenCalledTimes(1); + }); + + it("returns cancelled at once when the turn is flagged during the drain", async () => { + const f = makeFixtures({ zcodeSid: "sess_z", cancelled: true }); + f.pollOnce.mockResolvedValue({ status: "running" }); + + const result = await drainBackendAfterCancel(f.server, f.deps); + + expect(result).toBe("cancelled"); + expect(f.sent.map((s) => s.method)).toEqual(["session/stop", "v4/command"]); + // Cancelled mid-drain: the next prompt's own drain re-baselines instead. + expect(f.differ.markSeen).not.toHaveBeenCalled(); + }); +}); diff --git a/tests/session-cancel.test.ts b/tests/session-cancel.test.ts index 1206ff8..a1959a8 100644 --- a/tests/session-cancel.test.ts +++ b/tests/session-cancel.test.ts @@ -195,4 +195,77 @@ describe("runEventTurn: cancel exits immediately (backend ignores session/stop)" expect(resp).toEqual({ stopReason: "cancelled" }); expect(f.sent).toEqual([]); }); + + it("reports a steered-and-dropped send at once instead of hanging (turn.steerQueued)", async () => { + const f = makeTurnFixtures(); + const turn: PendingTurn = { zcodeSid: "sess_z", cancelled: false }; + // The send landed mid-generation: the backend queued it as steer input + // (silently dropped when the old turn ends) and emitted steerQueued. No + // turn.started will ever arrive — report the swallow immediately rather + // than hanging until the 120s watchdog. + f.pollEvent.mockResolvedValue({ + sessionId: "sess_z", + seq: 1, + type: "turn.steerQueued", + payload: {}, + }); + + const resp = await runEventTurn( + f.server, + f.listener, + f.monitor, + f.differ, + f.cx, + "acp_a", + "m1", + turn, + true, // gate armed: cancel/preempt window + ); + + expect(resp).toEqual({ stopReason: "max_turn_requests" }); + // Nothing of ours is generating — no stop pair; the visible note tells + // the user to resend. + expect(f.sent).toEqual([]); + const note = f.cx.notify.mock.calls.find((c) => JSON.stringify(c).includes("重新发送")); + expect(note).toBeDefined(); + }); + + it("does not report a steerQueued event once the turn has started", async () => { + const f = makeTurnFixtures(); + const turn: PendingTurn = { zcodeSid: "sess_z", cancelled: false }; + f.pollEvent.mockResolvedValueOnce({ + sessionId: "sess_z", + seq: 1, + type: "turn.started", + payload: {}, + }); + f.pollEvent.mockImplementation(async () => { + turn.cancelled = true; // end the loop via the cancel path + return { + sessionId: "sess_z", + seq: 2, + type: "turn.steerQueued", + payload: {}, + }; + }); + + const resp = await runEventTurn( + f.server, + f.listener, + f.monitor, + f.differ, + f.cx, + "acp_a", + "m1", + turn, + true, + ); + + // turn.started already passed: a late steerQueued belongs to someone + // else's send — the translator ignores it and the loop does NOT fire the + // steer report; the cancel flag ends the turn instead. + expect(resp).toEqual({ stopReason: "cancelled" }); + expect(f.sent.map((s) => s.method)).toEqual(["session/stop", "v4/command"]); + expect(f.cx.notify.mock.calls.some((c) => JSON.stringify(c).includes("重新发送"))).toBe(false); + }); }); From cca2f66fc336aac691c1214a112a747498a8d303 Mon Sep 17 00:00:00 2001 From: William Wang Date: Mon, 31 Aug 2026 19:21:31 +0800 Subject: [PATCH 12/14] chore: release 0.14.0 --- CHANGELOG.md | 2 ++ package.json | 2 +- 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b00953d..bd6c872 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +## [0.14.0] - 2026-08-31 + ### Added - REPL prompt history: every submit is recorded per project diff --git a/package.json b/package.json index 2170d47..8a669c1 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "zcode-acp-server", - "version": "0.13.0", + "version": "0.14.0", "description": "Agent Client Protocol (ACP) server bridging headless ZCode to editors like Zed and JetBrains.", "type": "module", "license": "Apache-2.0", From 79cf3fdd2cde7881f05d1a64ed347dc5099e3202 Mon Sep 17 00:00:00 2001 From: William Wang Date: Mon, 31 Aug 2026 19:50:14 +0800 Subject: [PATCH 13/14] fix: flush prose buffers on segment transitions so the REPL tail keeps stream order --- src/repl/model.ts | 43 ++++++++++++++++++++----- tests/repl-model.test.ts | 68 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 104 insertions(+), 7 deletions(-) diff --git a/src/repl/model.ts b/src/repl/model.ts index 48f8275..62ccad7 100644 --- a/src/repl/model.ts +++ b/src/repl/model.ts @@ -222,6 +222,24 @@ function blockText(content: unknown): string { return ""; } +/** + * Fold pending stream buffers into entries. The live tail renders entries + * first and the streaming buffers last, so a segment that starts later must + * not stay buffered below entries that streamed earlier — flush on every + * transition to keep the tail in stream order. At most one buffer is + * non-empty between transitions; both are handled for safety. + */ +function flushBuffers(next: TurnState): void { + if (next.thinkBuf.trim()) { + next.entries = [...next.entries, { kind: "thinking", text: next.thinkBuf.trim() }]; + next.thinkBuf = ""; + } + if (next.textBuf.trim()) { + next.entries = [...next.entries, { kind: "assistant", text: next.textBuf.trim() }]; + next.textBuf = ""; + } +} + /** * Apply one update to the live turn state. Returns a NEW state object (safe * for React-style always-replace updates). Unknown update kinds are ignored. @@ -235,17 +253,22 @@ export function applyUpdate(state: TurnState, update: SessionUpdate): TurnState switch (update.sessionUpdate) { case "agent_message_chunk": { const chunk = blockText(update.content); - // A message chunk after thinking means the thought stream ended — flush - // it as its own dim entry before the prose starts. - if (chunk && next.thinkBuf) { - next.entries = [...next.entries, { kind: "thinking", text: next.thinkBuf.trim() }]; - next.thinkBuf = ""; - } + // Entering prose flushes the thought stream as its own dim entry. + // Continuation chunks must not re-flush, and the buffer side is + // trim-tested: a whitespace-only thought chunk must not count as a + // segment (it would otherwise shred every following prose chunk into + // its own entry). + if (chunk && next.thinkBuf.trim()) flushBuffers(next); next.textBuf = next.textBuf + chunk; return next; } case "agent_thought_chunk": { - next.thinkBuf = next.thinkBuf + blockText(update.content); + const chunk = blockText(update.content); + // Symmetric: entering thinking flushes the prose segment as its own + // entry — otherwise it stays pinned below every later entry until the + // turn ends and the tail renders out of stream order. + if (chunk.trim() && !next.thinkBuf.trim()) flushBuffers(next); + next.thinkBuf = next.thinkBuf + chunk; return next; } case "tool_call": @@ -264,6 +287,11 @@ export function applyUpdate(state: TurnState, update: SessionUpdate): TurnState next.entries = entries; } } else { + // A fresh row starts a non-prose segment: flush pending buffers so + // the row lands after everything streamed before it. In-place status + // updates keep buffers alone — a live thinking stream would shred + // into fragments on every status change. + flushBuffers(next); next.entries = [...next.entries, { kind: "tool", id, title, status }]; } return next; @@ -271,6 +299,7 @@ export function applyUpdate(state: TurnState, update: SessionUpdate): TurnState case "plan": { // First version renders plans as a one-line note; full plan UI is a // follow-up. Counts entries when the shape provides them. + flushBuffers(next); const items = Array.isArray(update.entries) ? update.entries.length : 0; next.entries = [ ...next.entries, diff --git a/tests/repl-model.test.ts b/tests/repl-model.test.ts index f6680d9..39668fe 100644 --- a/tests/repl-model.test.ts +++ b/tests/repl-model.test.ts @@ -58,6 +58,74 @@ describe("applyUpdate", () => { expect(s.thinkBuf).toBe(""); }); + it("flushes prose as an entry when thinking resumes after it", () => { + let s = createTurnState(); + s = applyUpdate(s, chunk("agent_message_chunk", "para one")); + s = applyUpdate(s, chunk("agent_thought_chunk", "hmm")); + expect(s.entries).toEqual([{ kind: "assistant", text: "para one" }]); + expect(s.textBuf).toBe(""); + expect(s.thinkBuf).toBe("hmm"); + }); + + it("keeps stream order across a prose → thinking → tool interleave", () => { + let s = createTurnState(); + s = applyUpdate(s, chunk("agent_message_chunk", "let me check")); + s = applyUpdate(s, chunk("agent_thought_chunk", "which file?")); + s = applyUpdate(s, { + sessionUpdate: "tool_call", + toolCallId: "t1", + title: "Read x.ts", + status: "in_progress", + } as SessionUpdateLike); + expect(s.entries).toEqual([ + { kind: "assistant", text: "let me check" }, + { kind: "thinking", text: "which file?" }, + { kind: "tool", id: "t1", title: "Read x.ts", status: "in_progress" }, + ]); + expect(s.textBuf).toBe(""); + expect(s.thinkBuf).toBe(""); + }); + + it("keeps buffers alone on in-place tool status updates", () => { + let s = createTurnState(); + s = applyUpdate(s, { + sessionUpdate: "tool_call", + toolCallId: "t1", + title: "Read x.ts", + status: "in_progress", + } as SessionUpdateLike); + s = applyUpdate(s, chunk("agent_thought_chunk", "still thinking")); + s = applyUpdate(s, { + sessionUpdate: "tool_call_update", + toolCallId: "t1", + status: "completed", + } as SessionUpdateLike); + // A status change mid-thinking must not shred the live thought stream. + expect(s.thinkBuf).toBe("still thinking"); + expect(s.entries).toEqual([ + { kind: "tool", id: "t1", title: "Read x.ts", status: "completed" }, + ]); + }); + + it("flushes prose before a plan note", () => { + let s = createTurnState(); + s = applyUpdate(s, chunk("agent_message_chunk", "here comes the plan")); + s = applyUpdate(s, { sessionUpdate: "plan", entries: [{}, {}] } as SessionUpdateLike); + expect(s.entries).toEqual([ + { kind: "assistant", text: "here comes the plan" }, + { kind: "note", text: "plan · 2 steps" }, + ]); + }); + + it("does not shred prose on whitespace-only thought chunks", () => { + let s = createTurnState(); + s = applyUpdate(s, chunk("agent_message_chunk", "para ")); + s = applyUpdate(s, chunk("agent_thought_chunk", "\n\n")); + s = applyUpdate(s, chunk("agent_message_chunk", "two")); + expect(s.entries).toEqual([]); + expect(s.textBuf).toBe("para two"); + }); + it("upserts tool rows by toolCallId and keeps last status", () => { let s = createTurnState(); s = applyUpdate(s, { From 13ca27748e5f6526903a358278e3f85e24aa782e Mon Sep 17 00:00:00 2001 From: William Wang Date: Mon, 31 Aug 2026 19:50:34 +0800 Subject: [PATCH 14/14] chore: release 0.14.1 --- CHANGELOG.md | 13 +++++++++++++ package.json | 2 +- 2 files changed, 14 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index bd6c872..174ce86 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +## [0.14.1] - 2026-08-31 + +### Fixed + +- REPL live turn: streamed prose now interleaves with thinking and tool + entries in stream order. Prose segments are flushed as entries whenever + thinking resumes, a fresh tool row starts, or a plan note arrives — + previously the whole reply accumulated in a single buffer pinned to the + bottom of the live-turn tail until the turn ended, rendering later + thinking/tool entries above earlier prose and letting long replies crowd + the tail. Whitespace-only thought chunks are ignored as segment + transitions so they cannot shred prose. + ## [0.14.0] - 2026-08-31 ### Added diff --git a/package.json b/package.json index 8a669c1..8e87480 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "zcode-acp-server", - "version": "0.14.0", + "version": "0.14.1", "description": "Agent Client Protocol (ACP) server bridging headless ZCode to editors like Zed and JetBrains.", "type": "module", "license": "Apache-2.0",