From 3712ddceccc39e880b48b8b689bec16c645879bf Mon Sep 17 00:00:00 2001 From: Nekono Nana KAKKO KARI <3267314+nananek@users.noreply.github.com> Date: Wed, 12 Aug 2026 00:43:10 +0900 Subject: [PATCH 1/3] =?UTF-8?q?Plan:=20handoff=20=E9=80=81=E5=8F=97?= =?UTF-8?q?=E4=BF=A1=E3=81=AE=E4=BF=A1=E9=A0=BC=E6=80=A7=E5=90=91=E4=B8=8A?= =?UTF-8?q?=E3=81=A8=20read=5Foutput=20=E3=81=AE=E7=94=BB=E9=9D=A2?= =?UTF-8?q?=E8=A1=A8=E7=A4=BA=E5=AF=BE=E5=BF=9C=20(=E8=AA=BF=E6=9F=BB?= =?UTF-8?q?=E3=83=BB=E5=AE=9F=E8=A3=85=E8=A8=88=E7=94=BB)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - handoff: groupManagerApi の getGroup 欠落 (repo_info 失敗の確定原因) と、 takeHandoff のウェイターが接続死活を知らないことによるイベント喪失レースを修正する計画。 - read_output: 生バイト連結では復元できない TUI 差分描画を、軽量仮想画面モデルと 画面変化ベースの idle 時間 (screen / screenIdleMs) で観測できるようにする計画。 --- tmp/handoff-reliability-plan.md | 188 ++++++++++++++++++++++++++++++++ tmp/read-output-screen-plan.md | 147 +++++++++++++++++++++++++ 2 files changed, 335 insertions(+) create mode 100644 tmp/handoff-reliability-plan.md create mode 100644 tmp/read-output-screen-plan.md diff --git a/tmp/handoff-reliability-plan.md b/tmp/handoff-reliability-plan.md new file mode 100644 index 0000000..08b2a9d --- /dev/null +++ b/tmp/handoff-reliability-plan.md @@ -0,0 +1,188 @@ +# Plan: handoff の送受信失敗をなくす (wait_for_handoff / handoff_to_orchestrator の信頼性向上) + +## 目的 + +オーケストレーターの `wait_for_handoff` とワーカーの `handoff_to_orchestrator` の間で、 +ハンドオフイベントが「送受信されない」ことがある問題を解決する。調査の手がかりとして +報告された `ccserver_repo_info` の失敗 (`deps.groupManager.getGroup is not a function`) は +**確定バグ**として特定した (後述)。これを直すだけでなく、ハンドオフイベント自体が +キューから失われるレースをコードレベルで塞ぐ。 + +## 調査結果 (コードを読んで裏取り済み) + +### 経路の全体像 + +``` +worker (opencode/claude) + └─ handoff_to_orchestrator (handoffソケット, buildHandoffMcpServer) + └─ tools.handoffToOrchestrator (mcpTools.js:166) + └─ groupManager.pushHandoff (groupManager.js:595) …同期的にキューへ + └─ group.handoffQueue + handoffEmitter.emit('handoff') + └─ takeHandoff (groupManager.js:621) のウェイターが shift + └─ wait_for_handoff (controlソケット, buildControlMcpServer) + └─ orchestrator が結果を受信 +``` + +- ブローカーは接続ごとに `buildServer()` で**新しい McpServer インスタンス**を作る + (mcpBroker.js:101-111 `settleConnection`)。ツールの deps は `startControlBroker` / + `startHandoffChannel` に渡された共通オブジェクトで、**接続ごとのクロージャを含まない**。 +- ワーカー側の `handoffToOrchestrator` の push は**サーバー内で同期完結**する + (mcpTools.js:174)。つまり ack が失われてもイベント自体はキューに残る。**イベント喪失の + 起点は「取り出し側」(takeHandoff) にしかない**。 + +### 原因1 (確定バグ): ブローカーの deps facade に `getGroup` が無い + +- `repoInfo` は `deps.groupManager.getGroup(deps.groupId)` を呼ぶ (mcpTools.js:293)。 +- 本番ではブローカーに `groupManagerApi` (groupManager.js:749-758) が渡されるが、 + **この facade には `getGroup` が含まれていない** (listGroupMembers / isSessionInGroup / + getRoleForSession / setCurrentTurn / pushHandoff / takeHandoff / addMember / removeMember のみ)。 +- 結果: 本番の `repo_info` は常に `TypeError: deps.groupManager.getGroup is not a function` を投げる。 + MCP SDK 1.30.0 はハンドラー例外を catch して `{ content:[...], isError:true }` を返す + (node_modules/@modelcontextprotocol/sdk/dist/esm/server/mcp.js:135-142) ためプロセスは落ちないが、 + オーケストレーターにはツール失敗として見える。これが報告エラーの正体。 +- **テストが検出できなかった理由**: `mcpTools.test.js` の `controlDeps` は**完全な + groupManager モジュール**を注入している (mcpTools.test.js:57-63)。本番は facade を渡すため、 + テストと本番で deps の形が異なり、ズレがすり抜けた。 + +### 原因2 (イベント喪失レース): takeHandoff のウェイターが接続の死活を知らない + +- `takeHandoff` はウェイターを `group.pendingTakes` に登録し、最大 15 分 (デフォルト + timeoutMs) 待つ。**このウェイターは「どの接続が待っているか」を知らない**。 +- クライアント側キャンセルでサーバー側のウェイターが生き残ることは、コードコメント自身が + 「既知の現実」として認めている (groupManager.js:612-620: "A client-side cancelled MCP + request leaves its takeHandoff promise -- and its listener -- alive server-side")。 +- これにより以下の経路で**イベントが消失**する: + + 1. オーケストレーターが `wait_for_handoff` を発行 (サーバー側ウェイター登録)。 + 2. 何らかの理由でこの要求が死ぬ/破棄される: + - クライアント側タイムアウト・キャンセル (接続は生きたまま、要求だけ破棄) + - 制御ブローカーの teardown (オーケストレーター pty exit → `onOrchestratorExit` + (groupManager.js:652-659) が `stopBroker` で接続を destroy。**この時 + `pendingTakes` は settle されず放置**) + - サーバー再起動・グループ破棄以外の要因で接続切断 + 3. ワーカーが `handoff_to_orchestrator` → `pushHandoff` → イベント到着。 + 4. **stale ウェイターがキューからイベントを shift し**、SDK が結果を + (死んでいる/破棄された) 接続へ `SocketTransport.send` (mcpServer.js:80-83)。 + 破棄済みソケットへの `write()` は握りつぶされる (ブローカーのエラーリスナー + mcpBroker.js:83 が握るだけ)。 + 5. オーケストレーターが `wait_for_handoff` を呼び直すとキューは空 → `{timedOut:true}` + が返り、イベントは永遠に失われる。 + +- 既存の「上書き (supersede)」機構 (takeHandoff 冒頭の stale settle, groupManager.js:624-627) + は、**新しいウェイターがイベント到着前に来た場合にしか救えない**。到着後に来た場合は + 取り返しがつかない。また stale への返値 `{orphaned: true}` は文書化されておらず、 + オーケストレーター (LLM) が「ハンドオフが来た」と誤解釈する余地がある + (`{timedOut:true}` と違い、既知の形状でない)。 + +### 原因3 (周辺): ワーカー側 ack 喪失 (イベント自体は失われない) + +- `addMember` のロール置換は `stopBroker(prevChannel)` (groupManager.js:523) で旧ワーカーの + handoff チャネルを破棄する。この瞬間に旧ワーカーの in-flight な + `handoff_to_orchestrator` があれば ack は失われるが、**push 自体はサーバー側で既に + 完了している**ためイベントはキューに残る。軽微。 + +## 実装方針 + +### フェーズ1: 原因1 (facade の getGroup 欠落) を修正 + +1. `server/ws/groupManager.js` — `groupManagerApi` (749-758) に `getGroup` を追加。 + - `getGroup` は既にモジュールから export されている (160)。facade に並べるだけ。 + - ついでに、facade と mcpTools が使う関数の対応表をコメントで維持し、以後の + 追加ツールは必ず facade に入れる旨を明記。 +2. **テストを本番と同じ形にする** (再発防止の本体): + - `mcpTools.test.js` の `controlDeps` / `handoffDeps` が注入する `groupManager` を、 + 完全モジュールではなく**実 facade** に切り替える。facade を取得できるように + `groupManager.js` にテスト用 export (例: `export function getGroupManagerApi()`) を + 追加し、`mcpTools.test.js` はそれを使う。 + - `mcpBroker.test.js` に wire レベルの `repo_info` テストを追加: 実 tmp リポジトリを + 作って control ソケット経由で `repo_info` を呼び、`error` ではなく基本情報が返ること、 + `isError` にならないことを検証。これが通れば facade 欠落の再発は即座に検出される。 + +### フェーズ2: 原因2 (takeHandoff のイベント喪失レース) を修正 + +原則: **イベントは「取り出し時に、その取り出し先の接続が生きていると確認できる場合のみ」 +キューから外す**。接続死やキャンセルで結果が届かない可能性がある場合、イベントは +キューに残し、次の `wait_for_handoff` が必ず受け取れるようにする。 + +3. `server/ws/groupManager.js` — `takeHandoff(groupId, timeoutMs, opts)` に `opts.isAlive` + (関数) を追加: + - `onHandoff` の dequeue 直前で `if (opts.isAlive && !opts.isAlive()) return;` — + 死んだ接続のウェイターはイベントを shift しない (キューに残る)。 + - `opts.isAlive` が無い場合 (テスト・既存呼び出し) は従来挙動。 +4. `server/ws/groupManager.js` — 上書き (supersede) の再キュー化: + - 各ウェイターは自身が shift したイベントを記録する。 + - 新しい `takeHandoff` が stale ウェイターを上書きする際、その stale が**直近にイベントを + 消費していたら**、そのイベントをキューの先頭に戻す (配信が怪しいので次ウェイターが + 確実に受け取れるように)。同一イベントの二重 re-queue は参照一致で防ぐ。 + - 併発する 2 接続 (旧接続がまだ生きている) のケースで二重配信になる可能性は、 + 「旧接続が生きているのに要求だけキャンセルされる」パターンより遥かに稀で、 + 喪失より二重のほうが安全 (オーケストレーターが同一ハンドオフを 2 回受けるだけで、 + キューは空にならない)。 + - stale への返値を `{orphaned: true}` → **`{timedOut: true}` に変更** (オーケストレーターが + 既知の形状のみ受け取る)。コメントも更新。 +5. `server/ws/groupManager.js` — `onOrchestratorExit` (652-659) で `pendingTakes` を + `{timedOut:true}` で settle する (制御ブローカー停止時に 15 分のゾンビウェイターを残さない。 + イベントはキューに残るため、再起動後の次の wait で受信される)。 +6. `server/ws/mcpBroker.js` — control / handoff サーバーに**接続ごとの死活関数**を注入: + - `settleConnection` で `SocketTransport` を先に生成し、 + `buildServer(identity, connectionIsAlive)` の形で per-connection クロージャ + (`() => !socket.destroyed && !transport._closed`) を渡す。 + - `startControlBroker` は `buildControlMcpServer({ ...deps, connectionIsAlive })` に + 展開し、control サーバーの deps が接続ごとに固有になるようにする + (現在は全接続で共通 deps のため、ここを必ず直す)。 +7. `server/ws/mcpServer.js` / `server/ws/mcpTools.js` — `waitForHandoff` が + `deps.connectionIsAlive` を `takeHandoff` の `opts.isAlive` に渡す。 + - control サーバーの `wait_for_handoff` の説明文に追記: + 「タイムアウト・接続断でもイベントは失われない。`{timedOut:true}` なら単に + もう一度呼べばよい」。 +8. テスト: + - `groupManager.test.js`: + - supersede 時の再キュー (stale がイベントを消費済み → 新しい wait がそのイベントを受信)。 + - `isAlive: () => false` のウェイターはイベントを消費しない (次の生きた wait が受信)。 + - `onOrchestratorExit` が pendingTakes を `{timedOut:true}` で settle する。 + - 既存の `{orphaned: true}` 検証テスト (166-199) を `{timedOut: true}` に更新。 + - `mcpBroker.test.js` (wire レベル): + - **wait 中の接続が死んでもイベントは失われない**: クライアント A が + `wait_for_handoff` を発行 → A のソケットを破棄 → ワーカーが handoff → 新クライアント B の + `wait_for_handoff` が**そのイベントを受信**することを検証。 + - 接続クローズで pending ウェイターが残らないこと。 + - フェーズ1 の `repo_info` wire テスト。 + - `mcpTools.test.js`: facade 経由の deps で全ツールが正常動作すること (1 の切り替えで + カバーされる)。 + +### フェーズ3: ドキュメント更新 + +9. `server/routes/groups.js` `DEFAULT_ORCHESTRATOR_TEMPLATE` — ハンドオフ規律に追記: + - `wait_for_handoff` の `{timedOut:true}` は「そのままもう一度呼ぶ」で安全。 + - 接続再確立後も未受信ハンドオフはキューに残り、次の `wait_for_handoff` で届く。 +10. `README.md` — handoff の MCP ツール説明に上記の保証を追記 (該当箇所を探して修正)。 + +## 変更しない範囲 + +- `pushHandoff` の同期 push / キューの FIFO と 100 件キャップ (喪失の起点ではない)。 +- ハンドオフイベントの形状 (`fromSessionId/fromRole/summary/status/nextRole/at`)。 +- `handoff_to_orchestrator` のデフォルト引数・権限境界 (identity はクロージャ由来)。 +- チャネル/ブローカーのソケットパス決定規則 (listenMcp の既存挙動)。 +- MCP SDK / トランスポート層のフレーミング。 + +## 検証コマンド + +```bash +npm test # workspace=server の全ユニット/ワイヤーテスト +npm run build --workspace=client +npm run test:e2e # 回帰確認 +``` + +- 手動確認: 実サーバーでコンボ起動 → オーケストレーターから `repo_info` が成功すること、 + ワーカー→handoff→`wait_for_handoff` の往復が (再起動・再接続を挟んでも) 届くこと。 + +## 完了条件 + +- `repo_info` が本番 (facade) 経由で失敗しない。テストが facade を使用する形になり、 + 同種のズレが再発しない。 +- 待機中にクライアント接続が死んでも、ハンドオフイベントはキューに残り、次回の + `wait_for_handoff` で必ず受信される (wire テストで検証済み)。 +- 上書きされたウェイターへの返値が `{timedOut:true}` に統一され、未知の結果形状 + (`{orphaned:true}`) がオーケストレーターへ届かない。 +- 制御ブローカー停止時に 15 分のゾンビウェイターが残らない。 +- サーバー単体テスト・E2E が通る。 diff --git a/tmp/read-output-screen-plan.md b/tmp/read-output-screen-plan.md new file mode 100644 index 0000000..948745f --- /dev/null +++ b/tmp/read-output-screen-plan.md @@ -0,0 +1,147 @@ +# Plan: read_output でスピナー状態を誤判しない (仮想画面 + 画面ベースの活動判定) + +## 目的 + +`read_output` でワーカーのスピナー (作業中インジケータ) の状態を見ようとすると誤判する +問題を解決する。スピナーは TUI 特有の描画 (カーソル移動・行消去・代替画面への差分描画) +で更新されるため、生の xterm 描画ストリームを上から追う形では「今見えている状態」を +復元できない。**セッションごとに軽量な仮想画面 (ANSI 解釈) を維持し、read_output に +「現在の画面」と「画面が最後に変化した時刻」を提供**して、スピナーの有無・作業中か +静止か (busy/idle) を安定して観測できるようにする。 + +## 調査結果 (コードを読んで裏取り済み) + +### 現状の read_output の実装 + +- `readOutput` (mcpTools.js:50-75) は以下しか返さない: + - `raw` — `session.outputBuffer.slice(-n).join('')` (**生バイトの連結**) + - `text` — `stripAnsi(raw)` (正規表現 `ANSI_RE` による削除、mcpTools.js:20) + - `truncated` — 16KB キャップの有無 +- outputBuffer は pty の `onData` を chunk 単位で追記するだけ (sessionManager.js:313-318, + `appendToBuffer` 1103-1111)。**サーバー側に画面状態 (仮想端末) のモデルは存在しない**。 +- 代替画面 (DECSET 1049) の扱い: claude には + `CLAUDE_CODE_DISABLE_ALTERNATE_SCREEN=1` を注入する (sessionManager.js:255-258) が、 + **opencode は TUI のまま** (コメント 259-262 で明示的に alt 画面を許可)。opencode の + スピナー/ステータス行は代替画面内のカーソル位置固定描画。 + +### なぜ誤判するか + +1. **バイト連結では「今見えている画面」を復元できない**: + opencode の TUI はスピナーを `\r`/CSI カーソル移動 + 行消去 (CSI K 等) + フレーム文字の + 書き換えで描画する。`raw` は「フレーム1, フレーム2, ...」の全フレームが順番に並んだ + ストリームであり、`text` はそれらを単に連結したものになる (例: `⠋ 分析中…⠙ 分析中…⠹ + 分析中…`)。どのフレームが「現在画面に残っているか」はバイト列からは決まらない。 +2. **スピナーが止まった直後も誤った印象を与える**: 完了時に行を消去して別の表示に切り + 替えると、tail に残るのは古いスピナーフレーム。逆に静止プロンプト表示中でも、直前の + スピナーフレームが tail に残っていれば「まだ作業中」に見える。 +3. **バイトベースの idleForMs はスピナーで常に更新される**: `get_tab_status` の + `idleForMs` は `lastOutputAt` (バイト出力時刻) ベース (mcpTools.js:150, sessionManager + .js:297)。スピナーが回っている間は**内容が同じでも**バイトが出続けるため、 + 「実はモデルが待機/停滞しているのに作業中と誤判」する。 +4. **付随する正確性の問題**: + - `text` の 16KB slice (`raw.slice(-MAX_READOUTPUT_CHARS)`) がエスケープシーケンスの + 途中で切れると、`stripAnsi` が取りこぼして制御文字が `text` に漏れる。 + - pty chunk 境界で UTF-8 マルチバイト文字が分割されると mojibake になる + (outputBuffer は文字列として連結するため、跨ぎは join では修復できない)。 + +### 既存の周辺資産 + +- サーバー側に ANSI パーサは無い (`ANSI_RE` は正規表現のみ)。クライアントは xterm.js + が画面を持つ (ccserver 側の再利用は想定しない — サーバーは Node で動く軽量モデルを新設)。 +- read_output の呼び出し側 (オーケストレーター) は、テンプレート + (routes/groups.js `DEFAULT_ORCHESTRATOR_TEMPLATE` 135-141) で「read_output で + アイドル/完了プロンプトかどうか確認」する運用を既にしているため、**この判定を + 安定させる API が必要**。 + +## 実装方針 + +### フェーズ1: 軽量仮想画面モデル `server/ws/screenModel.js` (新規・純モジュール) + +1. 新モジュール `screenModel.js` を追加。node --test で直接ユニットテスト可能な + 純関数/ファクトリ構成にする (mcpTools.js と同じ「アプリの mutable モジュールを + import しない」制約を踏襲)。 + - 状態: 可視行リスト (最近 N 行、既定 200、1 行は幅 W=80 で wrap)、カーソル位置、 + 代替画面フラグ、**部分エスケープ保持バッファ** (chunk 跨ぎのシーケンスを安全に継続)。 + - 処理する制御 (サブセットで十分。未対応シーケンスは無害に無視): + - 表示文字 (改行 wrap)、CR / LF / BS / TAB + - CSI: CUP/H, CUU/A, CUD/B, CUF/C, CUB/D, CHA/G, EL/K (0/1/2), ED/J (0/2/3), + SGR (描画属性は無視して破棄), 25l/25h (カーソル表示は無視) + - 代替画面: CSI ?1049h/l, ?47h/l (切り替え時は内容を保持し、フラグで公開) + - OSC / その他の ESC シーケンス: 破棄 + - 出力: `screenRows()` (現在の可視行), `altScreenActive()`, `version()` (1 画面変更ごと + に増える整数 — 変化検知用)。 +2. サイズ上限: 可視行 200 行 × 80 文字 ≒ 16KB 相当でメモリは有界 (outputBuffer 512KB と同格)。 +3. `server/ws/sessionManager.js` — `appendToBuffer` (1103-1111) で + `session.screen = screenModel` を並行維持: + - `session.screen` が無ければ作成 (createSession で初期化)。 + - chunk 追記時に `screen.feed(data)` を呼び、画面が実際に変化したら + `session.screenLastChangeAt = Date.now()` を更新。 + - UTF-8 の chunk 跨ぎ: chunk を `Buffer` 経由ではなく `TextDecoder` の stream モードで + 連結し、末尾の未完了文字を次 chunk へ持ち越す (mojibake を防ぐ)。decode 済み文字列を + outputBuffer と screen の両方に渡す。 + - 既存の outputBuffer / autoYesBuf 等の挙動は変えない (screen は追加物)。 + +### フェーズ2: read_output / get_tab_status への画面ベース情報の追加 + +4. `server/ws/mcpTools.js` — `readOutput` の結果に追加 (raw/text/truncated は後方互換で維持): + - `screen` — 現在の可視画面の末尾 (例: 最大 40 行、1 行 80 文字、行数が多い場合は + 末尾を返し `screenTruncated: true`)。生バイト tail ではなく「今見えている状態」。 + - `screenAlt` — 代替画面使用中か。 + - `screenIdleMs` — 画面が最後に変化してからの経過 (スピナーが回っていれば小さい、 + 静止プロンプトなら大きい)。**バイト出力ではなく画面変化ベース**なので、 + 「スピナーだけ回って中身は停滞」も busy として正しく判定される。 + - サイズ: `screen` は 16KB キャップを独立に適用 (コンテキスト肥大化防止の既存思想)。 +5. `server/ws/mcpTools.js` — `getTabStatus` にも `screenIdleMs` を追加 (read_output を + 呼ばずに判定できるように)。`idleForMs` (バイトベース) は互換のため残す。 +6. `server/ws/mcpServer.js` — ツール説明文を更新: + - `read_output`: 「`screen` と `screenIdleMs` を優先せよ。`text` は後方互換の + バイトストリーム。スピナー等の動的描画は生ストリームでは判定できない」旨を明記。 + - `get_tab_status`: `screenIdleMs` の説明を追記。 +7. `server/routes/groups.js` `DEFAULT_ORCHESTRATOR_TEMPLATE` — 「stuck 判定は + `read_output` の `screen` / `screenIdleMs` で行う」旨に更新 + (現在の「read_output でアイドル/完了プロンプトか確認」の記述を具体化)。 + +### フェーズ3: テスト + +8. `server/ws/screenModel.test.js` (新規): + - スピナー描画列 (フレーム + 行消去 + カーソル位置固定) を feed → `screenRows()` が + 最新フレームの 1 行だけになる。 + - CR 上書き / 行消去 / 画面消去 / 代替画面 1049 切替 / スクロール (行数超過) / + 未対応 CSI の無視。 + - エスケープシーケンスが chunk 境界で分割されても正しく処理される (partial 保持)。 + - UTF-8 マルチバイト文字の chunk 跨ぎ分割で mojibake しない (TextDecoder stream)。 +9. `server/ws/mcpTools.test.js` — readOutput の統合テスト追加 (fake session に + outputBuffer + screen を注入): + - スピナー風出力を注入した fake セッションで、`screen` が最新 1 行、 + `screenIdleMs` が画面変化時刻と整合すること。 + - `get_tab_status` の `screenIdleMs` が返ること。 + +## 変更しない範囲 + +- `raw` / `text` / `truncated` の既存フィールドと意味 (後方互換)。 +- outputBuffer の chunk 構造と 512KB キャップ (既存の attach/replay/resume 経路に影響 + させない)。 +- クライアント側の xterm.js 描画 (サーバー側モデルは read_output 専用)。 +- 代替画面の注入方針 (claude は無効化、opencode は TUI のまま — 変えない)。 + +## 検証コマンド + +```bash +npm test # workspace=server の全ユニット (screenModel / read_output 含む) +npm run build --workspace=client +npm run test:e2e # 回帰確認 +``` + +- 手動確認: 実サーバーで opencode ワーカーを起動し、作業中 (スピナー表示) と待機中 + (静止プロンプト) のそれぞれで `read_output` を呼び、`screen` が実際の見た目に一致し、 + `screenIdleMs` が busy/idle を正しく区別することを確認。 + +## 完了条件 + +- read_output が「現在見えている画面」(`screen`) と「画面変化ベースの idle 時間」 + (`screenIdleMs`) を返し、スピナー/動的描画を安定して判定できる。 +- エスケープシーケンスの chunk 跨ぎ・UTF-8 マルチバイト跨ぎで破損しない + (ユニットテストで検証済み)。 +- `text` の 16KB slice でシーケンスが途中切断されない (chunk 粒度で切る修正を含む)。 +- オーケストレーターのテンプレートとツール説明が新 API を指すよう更新されている。 +- サーバー単体テスト・E2E が通る。 From e6ce41ab0d321dfaa40021da301e26ce8678871a Mon Sep 17 00:00:00 2001 From: Nekono Nana KAKKO KARI <3267314+nananek@users.noreply.github.com> Date: Wed, 12 Aug 2026 01:44:55 +0900 Subject: [PATCH 2/3] Implement handoff reliability + read_output screen view (per plan 3712ddc) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit handoff (問題1): - groupManagerApi に getGroup を追加 (repo_info の TypeError 確定原因の修正)。 facade とツールの対応をコメントで明示し、テストは実 facade (getGroupManagerApi) 経由に変更 -- 本番/テストの deps 形状ズレの再発を防止。 - takeHandoff に opts.isAlive (接続死活) を追加: 死んだ接続のウェイターは イベントを dequeue しない。mcpBroker が接続ごとの死活クロージャを注入 (SocketTransport 生成後に buildServer(identity, connectionIsAlive))。 - supersede 時に stale が消費済みのイベントをキューの先頭へ戻す再キュー化 (参照一致で二重 re-queue 防止)、stale への返値を {orphaned:true} から {timedOut:true} へ統一。onOrchestratorExit で pendingTakes を settle (15 分のゾンビウェイターを残さない。イベントはキューに残る)。 - テスト: facade 経由 deps 化、supersede 再キュー / isAlive 非消費 / onOrchestratorExit settle、wire レベルの接続死後イベント到達 + repo_info (isError なし) を追加。 read_output (問題2): - 軽量仮想画面モデル server/ws/screenModel.js を新設 (ANSI サブセット解釈、 200x80 有界、chunk 跨ぎエスケープ保持、TextDecoder stream で UTF-8 跨ぎ対応)。 - sessionManager が出力 chunk ごとに screen を並行維持し、 screenLastChangeAt (画面変化ベース) を更新。read_output に screen / screenAlt / screenTruncated / screenIdleMs を追加、get_tab_status に screenIdleMs を追加 (idleForMs は互換のため残す)。 - text の 16KB キャップをエスケープシーケンスを割らない境界でカット (plan の chunk 粒度案より厳密: 途中切断による制御文字漏れを除去)。 - テンプレート/ツール説明/README を新 API と handoff 保証に更新。 検証: npm test 318 tests / 317 pass / 0 fail (1 skipped)、 npm run build --workspace=client 成功。実シェルセッションで screen が スピナー最終フレームを表示し screenIdleMs が増加することを確認。 E2E (playwright) はこの環境でブラウザバイナリが取得できず起動不可 (MS CDN ゲートウェイエラー、全失敗がブラウザ起動エラー) のため未実施。 --- README.md | 6 + server/routes/groups.js | 17 ++- server/ws/groupManager.js | 112 +++++++++++++--- server/ws/groupManager.test.js | 61 ++++++++- server/ws/mcpBroker.js | 17 ++- server/ws/mcpBroker.test.js | 84 +++++++++++- server/ws/mcpServer.js | 6 +- server/ws/mcpTools.js | 120 ++++++++++++++++- server/ws/mcpTools.test.js | 158 +++++++++++++++++++++-- server/ws/screenModel.js | 229 +++++++++++++++++++++++++++++++++ server/ws/screenModel.test.js | 129 +++++++++++++++++++ server/ws/sessionManager.js | 17 +++ 12 files changed, 906 insertions(+), 50 deletions(-) create mode 100644 server/ws/screenModel.js create mode 100644 server/ws/screenModel.test.js diff --git a/README.md b/README.md index 4d4963d..de10050 100644 --- a/README.md +++ b/README.md @@ -170,6 +170,11 @@ GitHub Copilot を選んだ場合: - シェルエラーが出てしまった場合、それを収拾しようとして空入力や Ctrl+C 相当の入力を送ってはいけません。継続入力待ちのシェルに対しては EOF のように作用し、**シェルプロセスごと終了させてしまうことがあります** (実際に一度そうなりました)。`get_tab_status` で `exited: true` を確認したら、そのタブは諦めて `close_tab` → `open_tab` で作り直してください。 - 新規に開いたタブに何かを送る前には、`read_output` で実際にアプリの TUI が描画されていることを確認してから送ってください。 +#### control MCP ツールの信頼性保証 (handoff と read_output) + +- **ハンドオフは失われません**: `wait_for_handoff` はタイムアウト (`{timedOut:true}`) 時に**そのままもう一度呼ぶだけで安全**です。誰も待っていない間に届いたハンドオフはキューに残り、また**待機中に接続が切れても**イベントを消費しないため、再接続後の次の `wait_for_handoff` が必ず受け取ります。サーバー再起動後も未受信ハンドオフは残っています。 +- **`read_output` の `screen` / `screenAlt` / `screenIdleMs` を使う**: ワーカーのスピナー等の動的描画はカーソル移動と行消去でその場を書き換えるため、生のバイト列 (`raw` / `text`) からは「今見えている画面」を復元できません。サーバーはセッションごとに軽量な仮想画面 (ANSI 解釈) を維持しており、`screen` が現在の可視画面、`screenIdleMs` が**画面が最後に変化してからの経過** (スピナーが回っていれば小さい値、静止プロンプトなら大きい値) です。stuck/busy 判定は `text` や `idleForMs` (バイトベース) よりこれらを優先してください。`get_tab_status` の `screenIdleMs` も同様です。 + #### オーケストレーターから見えるのは repo_info の基本情報だけ オーケストレーターのサンドボックスにワーカーのディレクトリは**マウントされません** (プロジェクトファイルへの直接アクセスは不可)。代わりに、control MCP サーバーのツール `repo_info` がグループのプロジェクト (cwd) の**基本情報だけ**を返します: トップレベルの構成 (ディレクトリ/ファイル名のみ、100 エントリ上限)、README の先頭 ~8KB、`package.json` の要約 (name/version/description と scripts/dependencies/devDependencies の**キー一覧のみ**、値は返さない、各 50 キー上限)、git 状態 (現在ブランチ / short HEAD / 直近 5 コミットの件名 / 変更ファイル数)。 @@ -319,6 +324,7 @@ ccserver/ │ ├── mcpServer.js # control / handoff / notify 各 MCP サーバー (SocketTransport 含む) │ ├── mcpBroker.js # Unix-socket MCP ブローカー (control/handoff はグループ毎、notify はプロセス毎 1 つ) │ ├── mcpTools.js # control/handoff ツールの実装 (deps 注入) +│ ├── screenModel.js # read_output 用の軽量仮想画面 (ANSI 解釈 + 変化検知) │ ├── sandbox.js # bwrap + rootless docker サンドボックス構築 │ ├── sandbox-entrypoint.sh │ ├── sandbox-gh-wrapper.cjs # サンドボックス内 gh をブローカー中継に差し替え diff --git a/server/routes/groups.js b/server/routes/groups.js index c584d20..9993b23 100644 --- a/server/routes/groups.js +++ b/server/routes/groups.js @@ -58,11 +58,16 @@ the MCP server "ccserver" that is already configured in this session. Each worker is a full terminal session you can inspect and control: - list_group_sessions -- see the members of this group. -- read_output -- read a member's recent terminal output (fallback for - inspecting a stuck member; avoid polling it). +- read_output -- read a member's current screen / recent terminal output + (fallback for inspecting a stuck member; avoid polling it). Use its + \`screen\` and \`screenIdleMs\` fields for stuck/busy judgments -- a static + screen (large screenIdleMs) means the member is idle even if its byte + stream is noisy; a small screenIdleMs means it is actively redrawing + (spinner or progress). - send_input -- type text into a member's terminal (submit defaults to true). - open_tab / close_tab -- add or terminate worker sessions. -- get_tab_status -- quick status of a member. +- get_tab_status -- quick status of a member (including screenIdleMs, the + screen-change-based idle signal). - repo_info -- the repository's basic facts (top-level layout, README, package.json summary, git state). Shallow by design: it never returns source-file contents, takes no path arguments, and is capped in size. @@ -130,6 +135,12 @@ orchestrator should catch this itself. - Every instruction sent via \`send_input\` MUST end with an explicit reminder to call \`handoff_to_orchestrator\` once done, blocked, or in need of input. +- \`wait_for_handoff\` returning \`{timedOut:true}\` is NOT an error: it + simply means no handoff arrived within the timeout. Call it again. A + handoff is never lost to a timeout or a disconnect -- an event that + arrives while nobody is waiting stays queued, and even if your + connection dies mid-wait, the next \`wait_for_handoff\` (after the + reconnect) receives it. - After sending a step, don't just trust \`wait_for_handoff\` to eventually notify you -- it only returns once the worker actually calls the tool, and nothing forces that to happen. When you get any other opportunity to diff --git a/server/ws/groupManager.js b/server/ws/groupManager.js index 9ecab55..2604d63 100644 --- a/server/ws/groupManager.js +++ b/server/ws/groupManager.js @@ -610,48 +610,110 @@ export function pushHandoff(groupId, event) { // Resolves with the next handoff event, or { timedOut: true } when timeoutMs // elapses with the queue still empty (timeoutMs <= 0 means never). // +// Reliability contract (the orchestrator's wait_for_handoff depends on it): +// an event is only ever dequeued by a waiter that can be reasonably expected +// to deliver it. A waiter whose client connection is dead or whose request +// was cancelled must not remove an event from the queue -- the event stays +// queued and the next wait_for_handoff receives it. +// // Only one waiter per group is ever meaningful (the orchestrator calls // wait_for_handoff one at a time). A client-side cancelled MCP request leaves // its takeHandoff promise -- and its listener -- alive server-side for up to // timeoutMs (15 min by default), and such a "zombie" listener, being older, // would consume the next pushHandoff before the real waiter ever sees it. // So a new takeHandoff first settles every still-pending waiter for the same -// group as { orphaned: true } (each finish tears its own listener/timer down), -// then registers the fresh waiter as the sole consumer. -export function takeHandoff(groupId, timeoutMs) { +// group as { timedOut: true } (each finish tears its own listener/timer +// down), then registers the fresh waiter as the sole consumer. +// +// opts.isAlive (a function, optional): checked right before a dequeue. When +// it returns false the waiter leaves the queue alone -- the event belongs to +// the next waiter whose connection is actually alive. The waiter itself is +// left pending (it cannot consume anything) until superseded or timed out. +// +// Dequeue is not the same as delivery: the waiter claims an event, then +// commits the delivery on the next macrotask. A supersede arriving in the +// same turn can still reclaim the claimed event (its connection may have died +// or its request been cancelled between the claim and the send), so the +// event is re-queued instead of being lost with the stale waiter. The same +// reclaim runs when the orchestrator exits (onOrchestratorExit) or a timeout +// fires while an event is claimed. +export function takeHandoff(groupId, timeoutMs, opts = {}) { const group = groups.get(groupId); if (!group) return Promise.resolve({ error: 'group-not-found' }); - for (const stale of [...group.pendingTakes]) { - console.warn(`[groupManager] takeHandoff(${groupId}): superseding a still-pending waiter`); - stale({ orphaned: true }); + if (group.pendingTakes.size > 0) { + console.warn(`[groupManager] takeHandoff(${groupId}): superseding ${group.pendingTakes.size} still-pending waiter(s)`); } + settlePendingTakes(group, { timedOut: true }); return new Promise((resolve) => { + const waiter = { consumed: null, finish: null, onHandoff: null }; let settled = false; const finish = (val) => { if (settled) return; settled = true; clearTimeout(timer); - group.pendingTakes.delete(finish); - group.handoffEmitter.off('handoff', onHandoff); + group.pendingTakes.delete(waiter); + group.handoffEmitter.off('handoff', waiter.onHandoff); resolve(val); }; - const onHandoff = () => { - if (group.handoffQueue.length > 0) finish(group.handoffQueue.shift()); + waiter.finish = finish; + waiter.onHandoff = () => { + if (group.handoffQueue.length === 0 || waiter.consumed) return; + if (opts.isAlive && !opts.isAlive()) return; + waiter.consumed = group.handoffQueue.shift(); + // Commit the delivery on the next macrotask, not inline: a supersede + // (a newer takeHandoff in the same turn) must be able to reclaim the + // event from this waiter, so it is never delivered to a connection + // whose request may already be gone. + setTimeout(() => finish(waiter.consumed), 0); }; const timer = timeoutMs > 0 - ? setTimeout(() => finish({ timedOut: true }), timeoutMs) + ? setTimeout(() => { + reclaimConsumed(group, waiter); + finish({ timedOut: true }); + }, timeoutMs) : null; - group.pendingTakes.add(finish); - group.handoffEmitter.on('handoff', onHandoff); - onHandoff(); + group.pendingTakes.add(waiter); + group.handoffEmitter.on('handoff', waiter.onHandoff); + waiter.onHandoff(); }); } +// Give back an event a (still-pending) waiter claimed but has not committed: +// its delivery is suspect (dead connection, cancelled request), so the event +// must reach the next waiter. Reference-guarded against re-queueing an event +// that already sits in the queue. +function reclaimConsumed(group, waiter) { + if (!waiter.consumed) return; + if (!group.handoffQueue.includes(waiter.consumed)) { + group.handoffQueue.unshift(waiter.consumed); + } + waiter.consumed = null; +} + +// Settle every pending waiter for the group with `val`, reclaiming any event +// a waiter already claimed. Used by supersede (a newer takeHandoff) and by +// onOrchestratorExit (the control broker went away: no zombie waiter may +// linger for the full timeout). +function settlePendingTakes(group, val) { + for (const stale of [...group.pendingTakes]) { + reclaimConsumed(group, stale); + stale.finish(val); + } +} + // Stop only the control broker (orchestrator exited) -- the workers stay -// alive so the human can keep working in them. +// alive so the human can keep working in them. Pending wait_for_handoff +// waiters (created by the now-destroyed control connections) are settled +// with { timedOut: true } so no 15-minute zombie survives the broker +// teardown; the events themselves stay in the queue (any claimed-but- +// undelivered event is reclaimed by settlePendingTakes), so the next +// orchestrator's wait_for_handoff still receives them. export function onOrchestratorExit(groupId) { const group = groups.get(groupId); if (!group) return; + if (group.pendingTakes.size > 0) { + settlePendingTakes(group, { timedOut: true }); + } if (group.controlBroker) { stopBroker(group.controlBroker); group.controlBroker = null; @@ -666,8 +728,8 @@ export function onOrchestratorExit(groupId) { export function destroyGroup(groupId) { const group = groups.get(groupId); if (!group) return; - for (const finish of [...group.pendingTakes]) { - finish({ error: 'group-destroyed' }); + for (const waiter of [...group.pendingTakes]) { + waiter.finish({ error: 'group-destroyed' }); } group.pendingTakes.clear(); for (const sessionId of [...group.members.values()]) { @@ -745,8 +807,15 @@ function cleanupMemberChannels(group, sessionId) { } // Public facade passed into broker servers (avoids exposing the module -// namespace's internals / keeps tool deps explicit). +// namespace's internals / keeps tool deps explicit). This IS the shape the +// production MCP tools receive -- keep it in sync with what mcpTools.js +// calls: a function used by a tool but missing here fails in production +// (TypeError) while the full-module tests stay green. Every tool added to +// mcpServer/mcpTools must have its backing groupManager function in this +// facade, and tests must inject this facade (getGroupManagerApi), not the +// full module. const groupManagerApi = { + getGroup, listGroupMembers, isSessionInGroup, getRoleForSession, @@ -757,6 +826,13 @@ const groupManagerApi = { removeMember, }; +// Test seam: returns the exact facade the broker servers receive. Unit tests +// must inject this -- never the full module -- so a facade/real mismatch +// (a missing function) is caught by the tests, not only in production. +export function getGroupManagerApi() { + return groupManagerApi; +} + // Session-manager facade. A `let` so tests can swap in fakes (see // setSessionApiForTests) to exercise addMember's spawn/teardown paths without // real ptys. All references go through this binding (function bodies only), diff --git a/server/ws/groupManager.test.js b/server/ws/groupManager.test.js index cfd2c6d..d899010 100644 --- a/server/ws/groupManager.test.js +++ b/server/ws/groupManager.test.js @@ -170,14 +170,15 @@ test('a newer takeHandoff supersedes a still-pending one (no zombie listener)', const waitA = groupManager.takeHandoff(gid, 0); // Call B: the real waiter arriving while A is still unresolved. Under the // pre-fix implementation A's listener would stay attached and consume the - // next pushHandoff, leaving B stuck until timeout; now A is orphaned first. + // next pushHandoff, leaving B stuck until timeout; now A is superseded + // first. const waitB = groupManager.takeHandoff(gid, 0); const event = { type: 'done', from: 'workerA' }; assert.equal(groupManager.pushHandoff(gid, event), true); const [resA, resB] = await Promise.all([waitA, waitB]); - assert.deepEqual(resA, { orphaned: true }, 'superseded waiter settles as orphaned, not by stealing the event'); + assert.deepEqual(resA, { timedOut: true }, 'superseded waiter settles as timedOut, not by stealing the event'); assert.deepEqual(resB, event, 'the latest waiter receives the pushed event'); }); @@ -190,14 +191,66 @@ test('a superseded waiter is removed from pendingTakes (no zombie listener left // The newer waiter supersedes A, which must not linger in pendingTakes -- // otherwise its listener would consume the next pushHandoff before waitB. const waitB = groupManager.takeHandoff(gid, 0); - assert.equal(groupManager.getGroup(gid).pendingTakes.size, 1, 'orphaned A must not linger'); - assert.deepEqual(await waitA, { orphaned: true }); + assert.equal(groupManager.getGroup(gid).pendingTakes.size, 1, 'superseded A must not linger'); + assert.deepEqual(await waitA, { timedOut: true }); groupManager.pushHandoff(gid, { type: 'first' }); assert.deepEqual(await waitB, { type: 'first' }); assert.equal(groupManager.getGroup(gid).pendingTakes.size, 0, 'resolved waiter cleans up'); }); +// The supersede reclaim: a waiter that already dequeued an event (its +// delivery is committed only on the next macrotask) gives it back to the +// queue when superseded -- the event must reach the fresh waiter instead of +// being lost with the stale one. +test('supersede reclaims an event a stale waiter already consumed', async () => { + const gid = await makeGroup(); + + const waitA = groupManager.takeHandoff(gid, 0); + const event = { type: 'done', from: 'workerA', summary: 'E1' }; + groupManager.pushHandoff(gid, event); // A dequeues it (delivery not yet committed) + + const waitB = groupManager.takeHandoff(gid, 0); // supersedes A, reclaiming the event + const [resA, resB] = await Promise.all([waitA, waitB]); + assert.deepEqual(resA, { timedOut: true }, 'the stale waiter settles as timedOut without the event'); + assert.deepEqual(resB, event, 'the reclaimed event reaches the new waiter'); +}); + +// The core no-loss guarantee: a waiter whose connection is dead must not +// dequeue anything -- the event stays queued for the next (live) waiter. +test('a dead (isAlive:false) waiter never consumes; the next live waiter receives the event', async () => { + const gid = await makeGroup(); + + const deadWait = groupManager.takeHandoff(gid, 0, { isAlive: () => false }); + groupManager.pushHandoff(gid, { type: 'done', from: 'workerA', summary: 'survives death' }); + // The dead waiter has not consumed: the queue still holds the event and a + // live waiter supersedes the dead one and receives it. + const liveWait = groupManager.takeHandoff(gid, 0, { isAlive: () => true }); + const [resDead, resLive] = await Promise.all([deadWait, liveWait]); + assert.deepEqual(resDead, { timedOut: true }); + assert.deepEqual(resLive, { type: 'done', from: 'workerA', summary: 'survives death' }); +}); + +test('onOrchestratorExit settles pending waiters as timedOut (no 15-min zombie)', async () => { + const gid = await makeGroup(); + + const wait = groupManager.takeHandoff(gid, 0); // never times out on its own + assert.equal(groupManager.getGroup(gid).pendingTakes.size, 1); + groupManager.onOrchestratorExit(gid); + assert.equal(groupManager.getGroup(gid).pendingTakes.size, 0, 'waiters settled on orchestrator exit'); + const res = await Promise.race([ + wait, + new Promise((r) => setTimeout(() => r('still-pending'), 500)), + ]); + assert.deepEqual(res, { timedOut: true }); + + // The queue is untouched: a worker handoff after the exit is still + // received by the next waiter. + groupManager.pushHandoff(gid, { summary: 'after exit' }); + const next = await groupManager.takeHandoff(gid, 200); + assert.deepEqual(next, { summary: 'after exit' }); +}); + test('destroyGroup settles pending takeHandoff waiters', async () => { const gid = randomUUID(); await groupManager.createGroup({ groupId: gid, cwd: '/srv/proj', orchestratorDir: '/srv/orch' }); diff --git a/server/ws/mcpBroker.js b/server/ws/mcpBroker.js index 4c3bafe..3d13ff7 100644 --- a/server/ws/mcpBroker.js +++ b/server/ws/mcpBroker.js @@ -108,8 +108,16 @@ async function listenMcp({ groupId, tag, buildServer, sockPath }) { // after the frame read is buffered by the paused socket and replayed // to the transport's own handler once it starts. socket.pause(); - const mcp = buildServer(identity); const transport = new SocketTransport(socket, seed); + // Per-connection liveness oracle for tools that must not act on behalf + // of a connection whose client is gone (e.g. wait_for_handoff must not + // dequeue an event for a dead socket -- the event would be lost). The + // transport is created before the server so the closure can observe its + // close state; buildServer receives it as the second argument (control/ + // handoff servers thread it into their deps; the notify server ignores + // it). + const connectionIsAlive = () => !socket.destroyed && !transport._closed; + const mcp = buildServer(identity, connectionIsAlive); // mcp.connect() is async (transport.start() + the MCP initialize // handshake). A rejected promise here must NOT become an unhandled // rejection (Node's default --unhandled-rejections=throw would crash the @@ -196,7 +204,10 @@ export async function startControlBroker(deps) { return listenMcp({ groupId: deps.groupId, tag: 'control', - buildServer: () => buildControlMcpServer(deps), + // Per-connection deps: the liveness closure differs per accepted socket, + // so the server is built with a connection-specific deps object, not the + // shared one (a shared deps could never carry per-connection state). + buildServer: (identity, connectionIsAlive) => buildControlMcpServer({ ...deps, connectionIsAlive }), }); } @@ -205,7 +216,7 @@ export async function startHandoffChannel(deps) { return listenMcp({ groupId: deps.groupId, tag: `handoff-${deps.role}`, - buildServer: () => buildHandoffMcpServer(deps), + buildServer: (identity, connectionIsAlive) => buildHandoffMcpServer({ ...deps, connectionIsAlive }), }); } diff --git a/server/ws/mcpBroker.test.js b/server/ws/mcpBroker.test.js index bc58bf8..339f54d 100644 --- a/server/ws/mcpBroker.test.js +++ b/server/ws/mcpBroker.test.js @@ -6,7 +6,7 @@ import { test, before, after } from 'node:test'; import assert from 'node:assert/strict'; import net from 'node:net'; -import { mkdtempSync, rmSync, existsSync } from 'node:fs'; +import { mkdtempSync, rmSync, existsSync, mkdirSync, writeFileSync } from 'node:fs'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { randomUUID } from 'node:crypto'; @@ -86,6 +86,13 @@ async function callTool(client, name, args) { return JSON.parse(result.content[0].text); } +// Raw variant: returns the full tools/call result so tests can inspect +// isError / non-JSON error payloads (e.g. a handler exception surfacing as +// { content: [...], isError: true }). +async function callToolRaw(client, name, args) { + return client.call('tools/call', { name, arguments: args }); +} + test('control socket: MCP initialize handshake works over the UDS', async () => { const c = mcpClient(control.sockPath); await c.connected; @@ -514,3 +521,78 @@ test('notify broker: a non-ccserver first line is replayed, never treated as ide broker.stopBroker(notify); } }); + +// --- handoff reliability: events survive a dead wait (Issue: handoff loss) +// --------------------------------------------------------------------------- + +// The root-cause regression test: production brokers inject the groupManager +// FACADE (not the full module), and repo_info calls deps.groupManager.getGroup. +// A facade missing getGroup made repo_info fail with a TypeError on every +// production call while the (full-module) unit tests stayed green. Over the +// wire this surfaces as an isError tools/call result -- assert it does not. +test('control socket: repo_info succeeds over the wire (facade carries getGroup)', async () => { + const dir = mkdtempSync(join(tmpdir(), 'ccserver-repo-wire-')); + mkdirSync(join(dir, 'src')); + writeFileSync(join(dir, 'README.md'), '# Wire repo'); + const gid = randomUUID(); + await groupManager.createGroup({ groupId: gid, cwd: dir, orchestratorDir: join(dir, '..', 'wire-orch') }); + try { + const ctrl = groupManager.getGroup(gid).controlBroker; + const c = mcpClient(ctrl.sockPath); + await c.connected; + const result = await callToolRaw(c, 'repo_info', {}); + assert.equal(result.isError, undefined, 'repo_info must NOT surface as a tool error'); + const out = JSON.parse(result.content[0].text); + assert.equal(out.error, undefined, out.content?.[0]?.text || 'no error field'); + assert.equal(out.cwd, dir); + assert.ok(out.root.dirs.includes('src'), 'root listing returned'); + assert.equal(out.readme.file, 'README.md'); + assert.equal(out.readme.text, '# Wire repo'); + c.close(); + } finally { + groupManager.destroyGroup(gid); + try { rmSync(dir, { recursive: true, force: true }); } catch { /* ignore */ } + } +}); + +// The core no-loss scenario: an orchestrator waits, its connection dies, a +// worker hands off afterwards -- the event must NOT be consumed by the dead +// waiter (its response would be written to a destroyed socket and lost). +// The next orchestrator connection receives it. +test('a handoff is not lost when the waiting connection dies mid-wait', async () => { + const waitA = mcpClient(control.sockPath); + await waitA.connected; + const deadWait = callTool(waitA, 'wait_for_handoff', { timeoutMs: 5000 }); + // Give the server a beat to register the waiter, then kill the connection. + await new Promise((r) => setTimeout(r, 100)); + waitA.raw.destroy(); + // Wait until the close has propagated server-side, so the dead waiter's + // liveness check is guaranteed to see the connection gone before the + // handoff arrives. + await new Promise((resolve) => waitA.raw.on('close', resolve)); + await new Promise((r) => setTimeout(r, 50)); + + // The worker hands off only AFTER the orchestrator's connection died. + const worker = mcpClient(handoff.sockPath); + await worker.connected; + const handoffRes = await callTool(worker, 'handoff_to_orchestrator', { + summary: 'survives the dead wait', + status: 'done', + }); + assert.deepEqual(handoffRes, { ok: true }); + worker.close(); + + // A fresh orchestrator connection receives the event. + const waitB = mcpClient(control.sockPath); + await waitB.connected; + const ev = await callTool(waitB, 'wait_for_handoff', { timeoutMs: 3000 }); + assert.equal(ev.error, undefined); + assert.equal(ev.summary, 'survives the dead wait'); + assert.equal(ev.fromRole, 'workerA'); + waitB.close(); + + // The dead waiter was superseded: no waiter may linger in the group. + assert.equal(groupManager.getGroup(groupId).pendingTakes.size, 0, 'no zombie waiter remains'); + // (The dead client never receives a response -- deadWait stays pending + // client-side by design; the server-side waiter was settled.) +}); diff --git a/server/ws/mcpServer.js b/server/ws/mcpServer.js index 839d769..6237114 100644 --- a/server/ws/mcpServer.js +++ b/server/ws/mcpServer.js @@ -100,7 +100,7 @@ export function buildControlMcpServer(deps) { server.tool( 'read_output', - 'Read the recent terminal output of a group member session. Returns raw bytes and ANSI-stripped text. tail is a count of output chunks (default 200; the server buffers up to ~512KB of the most recent output, chunked), not characters. The returned text is capped at 16KB (the buffer tail) with truncated:true when the cap is hit. This is a fallback for inspecting a possibly-stuck member -- for normal flow, prefer wait_for_handoff.', + 'Read the recent terminal output of a group member session. Returns raw bytes and ANSI-stripped text plus a screen view: screen (the member\'s current visible screen -- its latest rows, capped at 40 lines of 80 chars; the raw byte stream cannot show this because TUI spinners redraw in place via cursor moves and line erases), screenAlt (whether an alternate screen is active), screenTruncated (when the screen view was cut to its cap) and screenIdleMs (ms since the screen last visibly changed -- a spinner keeps this small, a static prompt makes it grow; prefer screenIdleMs over idleForMs for busy/idle judgments, since bytes can keep flowing while the screen is unchanged). tail is a count of output chunks (default 200; the server buffers up to ~512KB of the most recent output, chunked), not characters. The returned text is capped at 16KB (the buffer tail) with truncated:true when the cap is hit. This is a fallback for inspecting a possibly-stuck member -- for normal flow, prefer wait_for_handoff.', { sessionId: z.string(), tail: z.number().optional() }, async (args) => ({ content: [{ type: 'text', text: JSON.stringify(tools.readOutput(deps, args)) }] }), ); @@ -134,7 +134,7 @@ export function buildControlMcpServer(deps) { server.tool( 'get_tab_status', - 'Return the live status of a group member session (exited, connected, cwd, app) plus autoYes (whether automatic permission-approval is currently enabled), lastOutputAt (epoch ms of its last output; null if none yet) and idleForMs (ms since then). A large idleForMs on a member that should be working may mean it is stuck -- check with read_output to confirm.', + 'Return the live status of a group member session (exited, connected, cwd, app) plus autoYes (whether automatic permission-approval is currently enabled), lastOutputAt (epoch ms of its last output; null if none yet), idleForMs (ms since then -- byte-based) and screenIdleMs (ms since the screen last visibly changed; null when no live screen exists). screenIdleMs is the better stuck/busy signal: a spinner keeps redrawing the screen (small screenIdleMs) even while the model is stalled, while a static screen (large screenIdleMs) means the member is genuinely idle. A large idleForMs on a member that should be working may mean it is stuck -- check with read_output to confirm.', { sessionId: z.string() }, async (args) => ({ content: [{ type: 'text', text: JSON.stringify(tools.getTabStatus(deps, args)) }] }), ); @@ -148,7 +148,7 @@ export function buildControlMcpServer(deps) { server.tool( 'wait_for_handoff', - 'Block until a worker calls handoff_to_orchestrator, or the timeout elapses. Returns the structured handoff event (worker, summary, status) -- or {timedOut:true} on timeout, in which case simply call wait_for_handoff again. Call this once per turn instead of polling read_output.', + 'Block until a worker calls handoff_to_orchestrator, or the timeout elapses. Returns the structured handoff event (worker, summary, status) -- or {timedOut:true} on timeout, in which case simply call wait_for_handoff again. Handoffs are never lost: a handoff that arrives while no one is waiting stays queued, and even a connection that dies mid-wait does not consume it -- the next wait_for_handoff (after reconnect) receives it. Call this once per turn instead of polling read_output.', { timeoutMs: z.number().optional() }, async (args) => { const result = await tools.waitForHandoff(deps, args); diff --git a/server/ws/mcpTools.js b/server/ws/mcpTools.js index b14cd1e..df14b67 100644 --- a/server/ws/mcpTools.js +++ b/server/ws/mcpTools.js @@ -37,15 +37,85 @@ export function listGroupSessions(deps) { // orchestrator's context. This is a fallback for stuck-member inspection -- // the recommended flow is wait_for_handoff. // +// The raw byte stream cannot show what the member's screen currently looks +// like (TUI spinners redraw in place via cursor moves/line erases), so the +// server also keeps a lightweight virtual screen per session: `screen` is +// the current visible screen (tail of the screen model's rows), +// `screenAlt` whether an alternate screen is active, and `screenIdleMs` +// the time since the screen last visibly changed (bytes can keep flowing +// while the screen is static -- a spinner keeps writing frames; a screen +// that stopped changing means the member is idle). Prefer `screen` + +// `screenIdleMs` for stuck/busy judgments over `text`/`raw`. +// // Cost control: this feature exists to keep the orchestrator's context // small, so a default call must not balloon it. `tail` counts output chunks // (default 200 -- the server buffers up to ~512KB in chunks), and the // returned text is hard-capped at MAX_READOUTPUT_CHARS; when the cap bites, // the tail of the buffer is returned and `truncated: true` is set so the -// caller knows the head of the output was dropped. +// caller knows the head of the output was dropped. The text cap cuts at a +// boundary that never splits an escape sequence (a split one would leak +// bare control bytes through stripAnsi). The `screen` view gets its own cap +// (a row count well under the char cap by construction). const DEFAULT_OUTPUT_TAIL_CHUNKS = 200; const MAX_OUTPUT_TAIL_CHUNKS = 100000; const MAX_READOUTPUT_CHARS = 16 * 1024; +// The screen view is capped independently of the text cap: at most this many +// of the newest rows, each of which is at most SCREEN_COLS chars, so the +// returned screen stays well under MAX_READOUTPUT_CHARS. +const MAX_SCREEN_ROWS = 40; + +// Cut `text` to at most `maxChars` chars at a boundary that does not split +// an escape sequence, keeping the tail. stripAnsi() only removes *complete* +// sequences, so a plain `.slice(-maxChars)` can land mid-sequence and leak +// bare control bytes into the text view. Walk the stream from the front, +// skip complete sequences, and cut at the last clean position at or before +// the cap -- when the cap splits a sequence, cut right after that sequence +// (the tail then starts clean and stays at or under the cap). +function cleanTextCut(text, maxChars) { + if (text.length <= maxChars) return text; + const limit = text.length - maxChars; + let cut = limit; + let i = 0; + while (i <= limit && i < text.length) { + if (text[i] === '\x1b') { + const end = ansiSequenceEnd(text, i); + if (end === -1) break; // dangling sequence to the end -- cut at the limit + if (end > limit) { // the cap splits this sequence + cut = end; + break; + } + i = end; + } else { + i++; + } + } + return text.slice(cut); +} + +// End index (exclusive) of the escape sequence starting at `start` (which +// must be an ESC byte), or -1 when the sequence is incomplete at the end of +// the input. Mirrors the ANSI_RE grammar (CSI/OSC/charset/single-char). +function ansiSequenceEnd(text, start) { + const next = text[start + 1]; + if (next === '[') { + let j = start + 2; + while (j < text.length && /[0-9;?]/.test(text[j])) j++; + if (j >= text.length) return -1; + return j + 1; // final byte 0x40-0x7E (anything else still terminates it) + } + if (next === ']') { + let j = start + 2; + while (j < text.length && text[j] !== '\x07' && !(text[j] === '\x1b' && text[j + 1] === '\\')) j++; + if (j >= text.length) return -1; + return text[j] === '\x07' ? j + 1 : j + 2; + } + if (next === '(' || next === ')' || next === '=' || next === '>' || next === '#') { + if (text.length < start + 3) return -1; + return start + 3; + } + if (next === undefined) return -1; + return start + 2; +} export function readOutput(deps, { sessionId, tail }) { const t = Number.isFinite(tail) ? tail : DEFAULT_OUTPUT_TAIL_CHUNKS; @@ -57,10 +127,11 @@ export function readOutput(deps, { sessionId, tail }) { if (!session) { return { error: 'not-found', message: 'session not found' }; } - let raw = session.outputBuffer.slice(-n).join(''); + const joined = session.outputBuffer.slice(-n).join(''); + let raw = joined; let truncated = false; - if (raw.length > MAX_READOUTPUT_CHARS) { - raw = raw.slice(-MAX_READOUTPUT_CHARS); + if (joined.length > MAX_READOUTPUT_CHARS) { + raw = joined.slice(-MAX_READOUTPUT_CHARS); truncated = true; } return { @@ -69,8 +140,33 @@ export function readOutput(deps, { sessionId, tail }) { app: session.app, exited: !!session.exited, raw, - text: stripAnsi(raw), + // The text view cuts the FULL stream at a sequence-safe boundary (raw + // stays backward-compatible byte tail); see cleanTextCut. + text: stripAnsi(cleanTextCut(joined, MAX_READOUTPUT_CHARS)), truncated, + ...screenView(session), + }; +} + +// The screen-model view of a session (see readOutput's doc comment). Null +// fields when the session has no screen model (e.g. a fake session in +// tests). +function screenView(session) { + const screen = session.screen; + if (!screen) { + return { screen: null, screenAlt: null, screenTruncated: null, screenIdleMs: null }; + } + let rows = screen.screenRows(); + let screenTruncated = false; + if (rows.length > MAX_SCREEN_ROWS) { + rows = rows.slice(-MAX_SCREEN_ROWS); + screenTruncated = true; + } + return { + screen: rows.join('\n'), + screenAlt: screen.altScreenActive(), + screenTruncated, + screenIdleMs: session.screenLastChangeAt != null ? Date.now() - session.screenLastChangeAt : null, }; } @@ -148,6 +244,11 @@ export function getTabStatus(deps, { sessionId }) { autoYes: !!session.autoYes, lastOutputAt: session.lastOutputAt, idleForMs: session.lastOutputAt != null ? Date.now() - session.lastOutputAt : null, + // Screen-change-based idle time (ms since the visible screen last + // changed; null when the session has no screen model). Unlike idleForMs + // (bytes-based), a spinner that keeps redrawing keeps this small -- a + // large value means the screen is genuinely static. + screenIdleMs: session.screenLastChangeAt != null ? Date.now() - session.screenLastChangeAt : null, }; } @@ -156,8 +257,15 @@ export function getTabStatus(deps, { sessionId }) { // the orchestrator can simply call wait_for_handoff again). This is the // recommended wait primitive: one structured call instead of polling // read_output. +// +// deps.connectionIsAlive (a per-connection function, when provided) is +// forwarded to takeHandoff: an event is never dequeued for a connection +// whose socket is dead, so a handoff is never lost to a disconnected wait -- +// it stays queued and the next wait_for_handoff receives it. export function waitForHandoff(deps, { timeoutMs = 900000 }) { - return deps.groupManager.takeHandoff(deps.groupId, Math.max(Number(timeoutMs) || 0, 0)); + const opts = {}; + if (typeof deps.connectionIsAlive === 'function') opts.isAlive = deps.connectionIsAlive; + return deps.groupManager.takeHandoff(deps.groupId, Math.max(Number(timeoutMs) || 0, 0), opts); } // Handoff (worker-only): notify the orchestrator that the worker's task is diff --git a/server/ws/mcpTools.test.js b/server/ws/mcpTools.test.js index 001f753..7ddb3f9 100644 --- a/server/ws/mcpTools.test.js +++ b/server/ws/mcpTools.test.js @@ -18,6 +18,7 @@ import { randomUUID } from 'node:crypto'; let runtimeDir; let groupManager; let tools; +let screenModel; let groupsToDestroy = []; // Real on-disk repo fixtures for repo_info (see the repoInfo tests below). let tmpRepos = []; @@ -31,6 +32,7 @@ before(async () => { process.env.CCSERVER_GROUPS_PATH = join(runtimeDir, 'saved-groups.json'); groupManager = await import('./groupManager.js'); tools = await import('./mcpTools.js'); + screenModel = await import('./screenModel.js'); }); after(() => { @@ -53,11 +55,15 @@ async function makeGroupAsync() { return id; } -// deps the way mcpServer would build them for the control socket +// deps the way mcpServer would build them for the control socket. The +// groupManager injected here is the REAL facade the production brokers +// receive (getGroupManagerApi) -- NOT the full module -- so a missing +// facade function (like the historical getGroup gap that broke repo_info in +// production) fails these tests instead of slipping past them. function controlDeps(groupId) { return { groupId, - groupManager, + groupManager: groupManager.getGroupManagerApi(), sessionManager: { getSession: () => null, writeToSession: () => false, waitUntilSettled: async () => ({ settled: true }) }, }; } @@ -69,7 +75,7 @@ function handoffDeps(groupId, role, sessionId) { groupId, role, getSessionId: () => sessionId, - groupManager, + groupManager: groupManager.getGroupManagerApi(), sessionManager: { getSession: () => null, writeToSession: () => false, waitUntilSettled: async () => ({ settled: true }) }, }; } @@ -225,7 +231,7 @@ test('sendInput moves the current turn to the targeted member', async () => { // A working writeToSession (the default controlDeps always returns false). const deps = { groupId: g, - groupManager, + groupManager: groupManager.getGroupManagerApi(), sessionManager: { getSession: () => ({}), writeToSession: () => true, waitUntilSettled: async () => ({ settled: true }) }, }; @@ -246,7 +252,7 @@ test('sendInput: holds the write until the settle gate opens (fresh session)', a const gate = new Promise((r) => { releaseGate = r; }); const deps = { groupId: g, - groupManager, + groupManager: groupManager.getGroupManagerApi(), sessionManager: { waitUntilSettled: async () => { await gate; @@ -272,7 +278,7 @@ test('sendInput: still writes when the settle gate times out, reporting settled: let writeCalls = 0; const deps = { groupId: g, - groupManager, + groupManager: groupManager.getGroupManagerApi(), sessionManager: { waitUntilSettled: async () => ({ settled: false, timedOut: true }), writeToSession: () => { writeCalls++; return true; }, @@ -291,7 +297,7 @@ test('sendInput: an already-settled session writes without waiting (no latency r let gateWaited = false; const deps = { groupId: g, - groupManager, + groupManager: groupManager.getGroupManagerApi(), sessionManager: { waitUntilSettled: async () => { gateWaited = true; return { settled: true }; }, writeToSession: () => true, @@ -324,7 +330,7 @@ test('sendInput (real session): holds the write until the idle gap opens the set const writes = []; const deps = { groupId: g, - groupManager, + groupManager: groupManager.getGroupManagerApi(), sessionManager: { getSession: (id) => sm.getSession(id), writeToSession: (id, text, opts) => { writes.push(text); return sm.writeToSession(id, text, opts); }, @@ -356,7 +362,7 @@ test('getTabStatus: reports lastOutputAt and the derived idleForMs', async () => const fakeSession = { cwd: '/srv/proj', app: 'claude', exited: false, socket: {}, lastOutputAt, autoYes: true }; const deps = { groupId: g, - groupManager, + groupManager: groupManager.getGroupManagerApi(), sessionManager: { getSession: (id) => (id === 'sess-a1' ? fakeSession : null), writeToSession: () => false }, }; const r = tools.getTabStatus(deps, { sessionId: 'sess-a1' }); @@ -372,7 +378,7 @@ test('getTabStatus: no output yet (lastOutputAt null) yields idleForMs null', as const fakeSession = { cwd: '/srv/proj', app: 'claude', exited: false, lastOutputAt: null, autoYes: false }; const deps = { groupId: g, - groupManager, + groupManager: groupManager.getGroupManagerApi(), sessionManager: { getSession: (id) => (id === 'sess-a1' ? fakeSession : null), writeToSession: () => false }, }; const r = tools.getTabStatus(deps, { sessionId: 'sess-a1' }); @@ -544,7 +550,7 @@ test('readOutput: authorized live member returns raw + stripped text (tail 0 cla }; const deps = { groupId: g, - groupManager, + groupManager: groupManager.getGroupManagerApi(), sessionManager: { getSession: (id) => (id === 'sess-a1' ? fakeSession : null), writeToSession: () => false }, }; // tail: 0 must NOT silently fall back to the 4000 default -- it clamps to @@ -572,7 +578,7 @@ test('readOutput: default tail stays small and output is hard-capped with trunca }; const deps = { groupId: g, - groupManager, + groupManager: groupManager.getGroupManagerApi(), sessionManager: { getSession: (id) => (id === 'sess-a1' ? fakeSession : null), writeToSession: () => false }, }; const out = tools.readOutput(deps, { sessionId: 'sess-a1' }); @@ -594,6 +600,134 @@ test('readOutput: default tail stays small and output is hard-capped with trunca assert.ok(small.raw.endsWith('x'.repeat(190))); }); +// --- screen view (Issue: read_output must show the CURRENT screen, not the +// byte stream -- spinners redraw in place and cannot be read from raw bytes) +// --------------------------------------------------------------------------- + +// A fake session carrying a real screen model, the way sessionManager would +// have one. +function makeScreenSession(screen, screenLastChangeAt = Date.now() - 100) { + return { + cwd: '/srv/project-x', + app: 'claude', + exited: false, + outputBuffer: ['\x1b[2K\r⠋ analyzing\r'], + screen, + screenLastChangeAt, + }; +} + +test('readOutput: screen shows the current view (spinner frames collapse to the latest line)', async () => { + const g = await makeGroupAsync(); + groupManager.registerMember(g, 'workerA', 'sess-a1'); + // A spinner redrawing one line in place: every frame is a CR + line erase + // + new frame char. Only the last frame must be on the screen. + const screen = screenModel.createScreenModel(); + screen.feed('\r\x1b[2K⠋ analyzing…\r\x1b[2K⠙ analyzing…\r\x1b[2K⠹ analyzing…'); + const deps = { + groupId: g, + groupManager: groupManager.getGroupManagerApi(), + sessionManager: { getSession: (id) => (id === 'sess-a1' ? makeScreenSession(screen, Date.now() - 5000) : null), writeToSession: () => false }, + }; + const out = tools.readOutput(deps, { sessionId: 'sess-a1' }); + assert.equal(out.error, undefined); + assert.equal(out.screen, '⠹ analyzing…', 'the screen view holds only the latest frame'); + assert.equal(out.screenAlt, false); + assert.ok(out.screenIdleMs >= 5000, `screenIdleMs derives from the screen change time (got ${out.screenIdleMs})`); +}); + +test('readOutput: screen honors the row cap with screenTruncated:true', async () => { + const g = await makeGroupAsync(); + groupManager.registerMember(g, 'workerA', 'sess-a1'); + const screen = screenModel.createScreenModel(); + for (let i = 0; i < 100; i++) screen.feed(`line ${i}\r\n`); + const deps = { + groupId: g, + groupManager: groupManager.getGroupManagerApi(), + sessionManager: { getSession: (id) => (id === 'sess-a1' ? makeScreenSession(screen) : null), writeToSession: () => false }, + }; + const out = tools.readOutput(deps, { sessionId: 'sess-a1' }); + assert.equal(out.screenTruncated, true); + assert.equal(out.screen.split('\n').length, 40, 'screen view capped at the newest 40 rows'); + assert.ok(out.screen.endsWith('line 99\n'), 'the newest row survives the cap'); +}); + +test('readOutput: session without a screen model yields null screen fields (no crash)', async () => { + const g = await makeGroupAsync(); + groupManager.registerMember(g, 'workerA', 'sess-a1'); + const deps = { + groupId: g, + groupManager: groupManager.getGroupManagerApi(), + sessionManager: { + getSession: (id) => (id === 'sess-a1' + ? { cwd: '/srv/project-x', app: 'claude', exited: false, outputBuffer: ['plain'] } + : null), + writeToSession: () => false, + }, + }; + const out = tools.readOutput(deps, { sessionId: 'sess-a1' }); + assert.equal(out.error, undefined); + assert.equal(out.text, 'plain'); + assert.equal(out.screen, null); + assert.equal(out.screenIdleMs, null); +}); + +test('getTabStatus: reports screenIdleMs from the screen model', async () => { + const g = await makeGroupAsync(); + groupManager.registerMember(g, 'workerA', 'sess-a1'); + const screenLastChangeAt = Date.now() - 9000; + const fakeSession = { + cwd: '/srv/proj', + app: 'claude', + exited: false, + socket: {}, + lastOutputAt: Date.now() - 500, + autoYes: true, + screen: screenModel.createScreenModel(), + screenLastChangeAt, + }; + const deps = { + groupId: g, + groupManager: groupManager.getGroupManagerApi(), + sessionManager: { getSession: (id) => (id === 'sess-a1' ? fakeSession : null), writeToSession: () => false }, + }; + const r = tools.getTabStatus(deps, { sessionId: 'sess-a1' }); + assert.equal(r.error, undefined); + assert.ok(r.screenIdleMs >= 9000 && r.screenIdleMs <= 10000, `screenIdleMs must be the time since the screen changed (got ${r.screenIdleMs})`); + // idleForMs stays byte-based (backward compatible): here bytes kept + // flowing recently while the screen is old -- the two signals diverge on + // purpose (a spinner writes bytes without changing the screen). + assert.ok(r.idleForMs <= r.screenIdleMs, 'a static screen with flowing bytes: idleForMs < screenIdleMs'); +}); + +test('readOutput: the 16KB text cap never splits an escape sequence (no control-byte leak)', async () => { + const g = await makeGroupAsync(); + groupManager.registerMember(g, 'workerA', 'sess-a1'); + // Layout the stream so the 16KB-from-the-end cut lands INSIDE an escape + // sequence: 9999 plain chars, then a SGR sequence starting at index 9999 + // (the cut position is raw.length - 16384 = 10000, inside the sequence), + // then a visible word and filler. A naive `.slice(-16K)` would start the + // text mid-sequence ("[31mcolored...") and stripAnsi would leave the + // residue -- control bytes leak into `text`. + const raw = 'a'.repeat(9999) + '\x1b[31m' + 'colored' + 'b'.repeat(16373); + const fakeSession = { + cwd: '/srv/project-x', + app: 'claude', + exited: false, + outputBuffer: [raw], + }; + const deps = { + groupId: g, + groupManager: groupManager.getGroupManagerApi(), + sessionManager: { getSession: (id) => (id === 'sess-a1' ? fakeSession : null), writeToSession: () => false }, + }; + const out = tools.readOutput(deps, { sessionId: 'sess-a1' }); + assert.equal(out.truncated, true); + assert.ok(out.text.length <= 16 * 1024, `text must stay capped (got ${out.text.length})`); + assert.ok(!out.text.includes('\x1b'), 'no bare ESC may leak through the cap'); + assert.ok(out.text.startsWith('colored'), 'the cut lands after the split sequence: clean text follows'); +}); + test('handoff queue is capped: overflow drops the oldest entries', async () => { const g = await makeGroupAsync(); groupManager.registerMember(g, 'workerA', 'sess-a1'); diff --git a/server/ws/screenModel.js b/server/ws/screenModel.js new file mode 100644 index 0000000..5301560 --- /dev/null +++ b/server/ws/screenModel.js @@ -0,0 +1,229 @@ +// Lightweight virtual screen model for read_output (see mcpTools.js). The +// server previously only buffered raw pty bytes, which cannot show what a +// member's screen currently looks like: TUI spinners redraw in place via +// cursor moves, line erases and alternate-screen diffs, so the byte stream +// is "frame 1, frame 2, ..." with no way to tell which frame is on screen. +// This module interprets a practical subset of the xterm stream per session +// and exposes the current visible screen plus a change counter. +// +// Pure module (no app imports, Node builtins only), unit-testable directly +// with node --test. Bounded memory: at most `rows` (default 200) visible +// rows of `cols` (default 80) chars each -- roughly 16KB, the same order as +// the output buffer cap. +// +// Supported control subset (unhandled sequences are dropped harmlessly): +// - printable text with wrapping, CR/LF/BS/TAB +// - CSI: CUP/H, CUU/A, CUD/B, CUF/C, CUB/D, CHA/G, EL/K (0/1/2), +// ED/J (0/2/3), SGR (attributes ignored), ?25 l/h (cursor hidden) +// - alternate screen: CSI ?1049 h/l, ?47 h/l (content kept, flag exposed) +// - OSC and other ESC sequences: discarded +// +// UTF-8: feed() accepts a string (the pty layer already delivers cleanly +// decoded text) or bytes, which are decoded through a per-model TextDecoder +// in stream mode so a multi-byte character split across chunks never +// mojibakes. + +export const SCREEN_COLS = 80; +export const SCREEN_ROWS = 200; + +export function createScreenModel({ cols = SCREEN_COLS, rows = SCREEN_ROWS } = {}) { + const capCols = Math.max(cols, 1); + const capRows = Math.max(rows, 1); + const decoder = new TextDecoder('utf-8', { fatal: false }); + + const lines = []; // visible rows, oldest first, each at most capCols chars + let cursorRow = 0; + let cursorCol = 0; + let alt = false; + let version = 0; + let pending = ''; // partial escape sequence awaiting the next chunk + + // --- internal mutations --------------------------------------------------- + + const bump = () => { version++; }; + + // Grow rows until the cursor row exists, scrolling the oldest off the top + // when the cap is reached (the cursor then stays at the same screen line). + const ensureRow = () => { + while (cursorRow >= lines.length) { + lines.push(''); + if (lines.length > capRows) { + lines.shift(); + cursorRow--; + } + } + }; + + const setChar = (ch) => { + if (cursorCol >= capCols) { + cursorRow++; + cursorCol = 0; + } + ensureRow(); + let line = lines[cursorRow]; + if (line.length < cursorCol) line = line.padEnd(cursorCol, ' '); + lines[cursorRow] = (line.slice(0, cursorCol) + ch + line.slice(cursorCol + 1)).replace(/\s+$/, ''); + cursorCol++; + bump(); + }; + + // --- control sequences ---------------------------------------------------- + + const csiParams = (body) => body.split(';').map((p) => (p === '' ? 0 : Number(p) || 0)); + + const eraseLine = (mode) => { + ensureRow(); + const line = lines[cursorRow]; + if (mode === 0) { + lines[cursorRow] = line.slice(0, cursorCol).replace(/\s+$/, ''); + } else if (mode === 1) { + lines[cursorRow] = (' '.repeat(Math.min(cursorCol, line.length)) + line.slice(cursorCol)).replace(/\s+$/, ''); + } else { + lines[cursorRow] = ''; + } + bump(); + }; + + const eraseDisplay = (mode) => { + if (mode === 2 || mode === 3) { + lines.length = 0; + cursorRow = 0; + cursorCol = 0; + lines.push(''); + } else if (mode === 1) { + // BOL of screen through the cursor -- rare; clear the rows above and + // the current row's head. + for (let r = 0; r < cursorRow; r++) lines[r] = ''; + eraseLine(1); + return; + } else { + // mode 0: cursor through the end of the screen. + eraseLine(0); + lines.length = cursorRow + 1; + } + bump(); + }; + + const cursorPos = (r, c) => { + const before = lines.length; + cursorRow = Math.max(0, (Number.isFinite(r) && r >= 1 ? r : 1) - 1); + ensureRow(); // positions below the current bottom scroll down like xterm + if (lines.length !== before) bump(); // a new row appeared on screen + cursorCol = Math.max(0, Math.min(Number.isFinite(c) && c >= 1 ? c - 1 : 0, capCols - 1)); + }; + + const csi = (paramsStr, final) => { + const priv = paramsStr.startsWith('?'); + const parts = csiParams(priv ? paramsStr.slice(1) : paramsStr); + const p0 = parts[0] || 0; + switch (final) { + case 'H': + case 'f': + cursorPos(parts[0] || 1, parts[1] || 1); + return; + case 'A': cursorRow = Math.max(0, cursorRow - (p0 || 1)); return; + case 'B': { + const before = lines.length; + cursorRow += (p0 || 1); + ensureRow(); + if (lines.length !== before) bump(); + return; + } + case 'C': cursorCol = Math.min(capCols - 1, cursorCol + (p0 || 1)); return; + case 'D': cursorCol = Math.max(0, cursorCol - (p0 || 1)); return; + case 'G': cursorCol = Math.max(0, Math.min((p0 || 1) - 1, capCols - 1)); return; + case 'K': eraseLine(p0); return; + case 'J': eraseDisplay(p0); return; + case 'h': + case 'l': + // Alternate screen only; cursor visibility (25) and other modes are + // ignored (they do not change visible content). + if (priv && (parts[0] === 1049 || parts[0] === 47)) { + alt = final === 'h'; + bump(); + } + return; + default: + return; // SGR (m) and the rest: attributes are discarded + } + }; + + // --- character / sequence dispatch ---------------------------------------- + + const text = (ch) => { + const code = ch.charCodeAt(0); + if (code === 0x0d) { cursorCol = 0; return; } // CR + if (code === 0x0a || code === 0x0c || code === 0x0b) { + const before = lines.length; + cursorRow++; + ensureRow(); + if (lines.length !== before) bump(); // a new row appeared on screen + return; + } // LF/FF/VT + if (code === 0x08) { cursorCol = Math.max(0, cursorCol - 1); return; } // BS + if (code === 0x09) { cursorCol = Math.min(capCols - 1, ((cursorCol >> 3) + 1) << 3); return; } // TAB + if (code < 0x20 || code === 0x7f) return; // other C0 controls / DEL + setChar(ch); + }; + + // Parse the escape sequence starting at input[start] (an ESC byte). + // Returns { end } (exclusive) when complete, { needsMore: true } when it + // runs off the end of the input (the caller keeps the tail pending). + const escapeSequence = (input, start) => { + const next = input[start + 1]; + if (next === '[') { + let j = start + 2; + while (j < input.length && '0123456789;?'.includes(input[j])) j++; + if (j >= input.length) return { needsMore: true }; + const final = input[j]; + if (final >= '@' && final <= '~') { + csi(input.slice(start + 2, j), final); + return { end: j + 1 }; + } + return { end: j + 1 }; // malformed CSI -- skip the final byte + } + if (next === ']') { + let j = start + 2; + while (j < input.length && input[j] !== '\x07' && !(input[j] === '\x1b' && input[j + 1] === '\\')) j++; + if (j >= input.length) return { needsMore: true }; + return { end: input[j] === '\x07' ? j + 1 : j + 2 }; + } + if (next === '(' || next === ')' || next === '=' || next === '>' || next === '#') { + if (input.length < start + 3) return { needsMore: true }; + return { end: start + 3 }; + } + if (next === undefined) return { needsMore: true }; + return { end: start + 2 }; + }; + + return { + feed(data) { + const input = pending + (typeof data === 'string' ? data : decoder.decode(data, { stream: true })); + pending = ''; + let i = 0; + while (i < input.length) { + const ch = input[i]; + if (ch === '\x1b') { + const seq = escapeSequence(input, i); + if (seq.needsMore) { + pending = input.slice(i); + return; + } + i = seq.end; + } else { + text(ch); + i++; + } + } + }, + screenRows() { + return lines.slice(); + }, + altScreenActive() { + return alt; + }, + version() { + return version; + }, + }; +} diff --git a/server/ws/screenModel.test.js b/server/ws/screenModel.test.js new file mode 100644 index 0000000..c507f37 --- /dev/null +++ b/server/ws/screenModel.test.js @@ -0,0 +1,129 @@ +// Unit tests for the lightweight virtual screen model (screenModel.js). +// No MCP SDK / bwrap / agent CLIs needed -- pure module tests. + +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { createScreenModel } from './screenModel.js'; + +test('plain text lands on the screen line by line', () => { + const s = createScreenModel(); + s.feed('hello\r\nworld'); + assert.deepEqual(s.screenRows(), ['hello', 'world']); +}); + +test('CR overwrites the current line (the spinner pattern)', () => { + const s = createScreenModel(); + s.feed('⠋ analyzing…'); + const v1 = s.version(); + s.feed('\r⠙ analyzing…'); + assert.ok(s.version() > v1, 'a new frame is a visible change'); + s.feed('\r⠹ analyzing…'); + assert.ok(s.version() > v1); + assert.deepEqual(s.screenRows(), ['⠹ analyzing…'], 'only the latest frame survives'); +}); + +test('spinner drawn with line erase + fixed cursor leaves a single line', () => { + const s = createScreenModel(); + s.feed('line 1\n'); + s.feed('\r\x1b[2K⠋ working\r\x1b[2K⠙ working\r\x1b[2K⠹ working'); + assert.deepEqual(s.screenRows(), ['line 1', '⠹ working']); +}); + +test('line erase modes: K 0 clears cursor-to-EOL, K 2 the whole line', () => { + const s = createScreenModel(); + s.feed('0123456789\r\n'); + s.feed('\x1b[5G'); // CHA: cursor to column 5 (1-based) + s.feed('ab'); + assert.deepEqual(s.screenRows(), ['0123456789', ' ab']); + s.feed('\r\x1b[5G\x1b[K'); + assert.deepEqual(s.screenRows(), ['0123456789', ''], 'K0 erases from the cursor to EOL'); + s.feed('\r\x1b[2K'); + assert.deepEqual(s.screenRows(), ['0123456789', ''], 'K2 erases the whole line'); +}); + +test('cursor positioning: CUP moves the write target, overwriting in place', () => { + const s = createScreenModel(); + s.feed('row one\r\nrow two\r\nrow three'); + s.feed('\x1b[2;1Hreplaced'); + assert.deepEqual(s.screenRows(), ['row one', 'replaced', 'row three']); +}); + +test('display erase: ED 2 clears everything; ED 0 clears cursor to screen end', () => { + const s = createScreenModel(); + s.feed('a\r\nb\r\nc'); + s.feed('\x1b[2J'); + assert.deepEqual(s.screenRows(), ['']); + s.feed('x\r\ny\r\nz'); + s.feed('\x1b[2;2H\x1b[J'); // cursor at (2,2), erase to end + assert.deepEqual(s.screenRows(), ['x', 'y']); +}); + +test('alternate screen: ?1049 h/l toggles the flag without clearing content', () => { + const s = createScreenModel(); + s.feed('main screen'); + assert.equal(s.altScreenActive(), false); + s.feed('\x1b[?1049h'); + assert.equal(s.altScreenActive(), true); + assert.deepEqual(s.screenRows(), ['main screen'], 'content is kept across the switch'); + s.feed('\x1b[?1049l'); + assert.equal(s.altScreenActive(), false); +}); + +test('scrolling: rows beyond the cap drop the oldest (bounded memory)', () => { + const s = createScreenModel({ rows: 5 }); + for (let i = 0; i < 20; i++) s.feed(`line ${i}\r\n`); + assert.equal(s.screenRows().length, 5); + assert.deepEqual(s.screenRows(), ['line 16', 'line 17', 'line 18', 'line 19', '']); +}); + +test('line wrap: text wider than the width wraps to the next row', () => { + const s = createScreenModel({ cols: 8 }); + s.feed('1234567890'); + assert.deepEqual(s.screenRows(), ['12345678', '90']); +}); + +test('unknown / ignored CSI sequences are dropped harmlessly (SGR, cursor hide, OSC)', () => { + const s = createScreenModel(); + s.feed('\x1b[31m\x1b[1mred text\x1b[0m'); + s.feed('\x1b[?25l'); + s.feed('\x1b]0;title\x07'); + s.feed(' visible'); + assert.deepEqual(s.screenRows(), ['red text visible']); +}); + +test('escape sequences split across chunk boundaries are joined correctly', () => { + const s = createScreenModel(); + const full = 'first\r\x1b[2Kline\r\x1b[31mred\x1b[0m end'; + // Feed one byte at a time -- every sequence boundary is a chunk boundary. + for (const ch of full) s.feed(ch); + assert.deepEqual(s.screenRows(), ['red end']); +}); + +test('OSC split across chunks (BEL terminator in a later chunk)', () => { + const s = createScreenModel(); + s.feed('before\x1b]0;long '); + s.feed('title\x07after'); + assert.deepEqual(s.screenRows(), ['beforeafter']); +}); + +test('UTF-8 multibyte characters split across byte chunks never mojibake', () => { + const s = createScreenModel(); + const bytes = new TextEncoder().encode('分析中… done'); + // Split at a byte boundary inside the second character ('析' = 3 bytes). + s.feed(bytes.slice(0, 4)); + s.feed(bytes.slice(4, 8)); + s.feed(bytes.slice(8)); + assert.deepEqual(s.screenRows(), ['分析中… done']); +}); + +test('version() counts visible changes; cursor-only movement does not', () => { + const s = createScreenModel(); + s.feed('abc'); + const v = s.version(); + s.feed('\r'); // CR alone: no visible change + assert.equal(s.version(), v); + s.feed('\x1b[2C'); // cursor right: no visible change + assert.equal(s.version(), v); + s.feed('d'); // a real change + assert.equal(s.version(), v + 1); +}); diff --git a/server/ws/sessionManager.js b/server/ws/sessionManager.js index 2e48fb0..5be8d45 100644 --- a/server/ws/sessionManager.js +++ b/server/ws/sessionManager.js @@ -7,6 +7,7 @@ import { fileURLToPath } from 'node:url'; import { buildSandboxSpawn, resolveApp, sandboxAvailable, loadSandboxConfig } from './sandbox.js'; import { buildMcpConfigArgsAndEnv } from './mcpConfig.js'; import { shouldInjectNotify, notifyEnabled, getNotifySockPath, notifyBrokerRunning } from './notify.js'; +import { createScreenModel, SCREEN_ROWS } from './screenModel.js'; import { isValidApp, appResumeArgs, @@ -308,6 +309,13 @@ export function createSession({ cwd, cols, rows, claudeSessionId, shell, sandbox startedClaudeSessionId: claudeSessionId || null, scheduleId: null, // key into the module-level `schedules` map, if any pendingInjection: null, // { text, at } — scheduled prompt awaiting a freshly-resumed session + // Lightweight virtual screen (see screenModel.js): fed every output + // chunk, exposing the current visible screen and a change counter so + // read_output can tell "spinner still drawing" from "static screen". + // screenLastChangeAt is stamped when the screen visibly changes (not on + // every byte) -- the basis of read_output's screenIdleMs / get_tab_status. + screen: createScreenModel({ cols, rows: SCREEN_ROWS }), + screenLastChangeAt: null, }; ptyProcess.onData((rawData) => { @@ -316,6 +324,15 @@ export function createSession({ cwd, cols, rows, claudeSessionId, shell, sandbox // the agent-only idle detection below). Pure activity bookkeeping. session.lastOutputAt = Date.now(); appendToBuffer(session, data); + // Keep the virtual screen model in parallel with the buffer: it only + // stamps screenLastChangeAt when the visible screen actually changes, + // so a spinner redrawing the same line registers as activity while a + // byte flow that leaves the screen static does not. + const screenVersion = session.screen.version(); + session.screen.feed(data); + if (session.screen.version() !== screenVersion) { + session.screenLastChangeAt = Date.now(); + } if (session.socket && session.socket.readyState === 1) { try { From 817c72dff288da51758182b5fa51b4f739292f34 Mon Sep 17 00:00:00 2001 From: Nekono Nana KAKKO KARI <3267314+nananek@users.noreply.github.com> Date: Wed, 12 Aug 2026 02:48:42 +0900 Subject: [PATCH 3/3] Review fix: read_output text never leaks a dangling escape past the cap cleanTextCut cut at a sequence-safe boundary, but a stream that ends mid-sequence (a pty chunk boundary split the last escape) still leaked the bare partial sequence through stripAnsi into the text view. Trim the dangling escape from the tail in both the capped and uncapped paths; regression tests cover both. --- server/ws/mcpTools.js | 41 ++++++++++++++++++++++++-------------- server/ws/mcpTools.test.js | 32 +++++++++++++++++++++++++++++ 2 files changed, 58 insertions(+), 15 deletions(-) diff --git a/server/ws/mcpTools.js b/server/ws/mcpTools.js index df14b67..68fb043 100644 --- a/server/ws/mcpTools.js +++ b/server/ws/mcpTools.js @@ -72,24 +72,35 @@ const MAX_SCREEN_ROWS = 40; // the cap -- when the cap splits a sequence, cut right after that sequence // (the tail then starts clean and stays at or under the cap). function cleanTextCut(text, maxChars) { - if (text.length <= maxChars) return text; - const limit = text.length - maxChars; - let cut = limit; - let i = 0; - while (i <= limit && i < text.length) { - if (text[i] === '\x1b') { - const end = ansiSequenceEnd(text, i); - if (end === -1) break; // dangling sequence to the end -- cut at the limit - if (end > limit) { // the cap splits this sequence - cut = end; - break; + if (text.length > maxChars) { + const limit = text.length - maxChars; + let cut = limit; + let i = 0; + while (i <= limit && i < text.length) { + if (text[i] === '\x1b') { + const end = ansiSequenceEnd(text, i); + if (end === -1) break; // dangling sequence to the end -- cut at the limit + if (end > limit) { // the cap splits this sequence + cut = end; + break; + } + i = end; + } else { + i++; } - i = end; - } else { - i++; } + text = text.slice(cut); } - return text.slice(cut); + // The stream itself may end mid-sequence (a pty chunk boundary split it), + // even when the cap did not: trim a dangling escape from the tail so bare + // control bytes never leak through stripAnsi. Only the last sequence can + // dangle (a dangling sequence runs to the end of the input). + for (let k = 0; k < text.length; k++) { + if (text[k] === '\x1b' && ansiSequenceEnd(text, k) === -1) { + return text.slice(0, k); + } + } + return text; } // End index (exclusive) of the escape sequence starting at `start` (which diff --git a/server/ws/mcpTools.test.js b/server/ws/mcpTools.test.js index 7ddb3f9..a128ee2 100644 --- a/server/ws/mcpTools.test.js +++ b/server/ws/mcpTools.test.js @@ -728,6 +728,38 @@ test('readOutput: the 16KB text cap never splits an escape sequence (no control- assert.ok(out.text.startsWith('colored'), 'the cut lands after the split sequence: clean text follows'); }); +test('readOutput: a dangling escape at the end of the stream never leaks into text', async () => { + const g = await makeGroupAsync(); + groupManager.registerMember(g, 'workerA', 'sess-a1'); + // The buffer tail itself ends mid-sequence (a pty chunk boundary split the + // sequence): with the cap biting, the returned tail would end with a bare + // "\x1b[31" that stripAnsi cannot remove -- the text view must not contain + // it. Also cover the no-cap case (short stream, dangling escape at the end). + const raw = 'a'.repeat(16381) + '\x1b[31'; + const fakeSession = { + cwd: '/srv/project-x', + app: 'claude', + exited: false, + outputBuffer: [raw], + }; + const deps = { + groupId: g, + groupManager: groupManager.getGroupManagerApi(), + sessionManager: { getSession: (id) => (id === 'sess-a1' ? fakeSession : null), writeToSession: () => false }, + }; + const out = tools.readOutput(deps, { sessionId: 'sess-a1' }); + assert.equal(out.truncated, true); + assert.ok(out.text.length <= 16 * 1024, `text must stay capped (got ${out.text.length})`); + assert.ok(!out.text.includes('\x1b'), 'the dangling escape must be trimmed, not leaked'); + assert.ok(out.text.endsWith('a'.repeat(100)), 'the visible tail survives'); + + const short = tools.readOutput({ + ...deps, + sessionManager: { getSession: (id) => (id === 'sess-a1' ? { ...fakeSession, outputBuffer: ['plain\x1b[3'] } : null), writeToSession: () => false }, + }, { sessionId: 'sess-a1' }); + assert.equal(short.text, 'plain', 'a dangling escape at the end of a short stream is trimmed too'); +}); + test('handoff queue is capped: overflow drops the oldest entries', async () => { const g = await makeGroupAsync(); groupManager.registerMember(g, 'workerA', 'sess-a1');