diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md new file mode 100644 index 000000000..037e2d204 --- /dev/null +++ b/ARCHITECTURE.md @@ -0,0 +1,232 @@ +# Codeg 架构设计 + +> 多智能体编码工作台：把 Claude Code、Codex、OpenCode、Gemini、OpenClaw、Cline、Grok、Hermes 等多种 agent CLI 聚合到同一工作区，做会话聚合与多智能体协作。支持桌面安装与服务器/Docker 部署。 + +本文描述整体设计与关键机制。代码标识符、路径、协议字段保留英文。 + +--- + +## 1. 最核心的设计决策：一份业务核心，三种二进制 + +整个架构围绕一个决策展开——用 Cargo feature flags 让**同一份代码**编译出三种形态，而不是维护三套逻辑： + +| 二进制 | feature | 形态 | +|---|---|---| +| `codeg` | `tauri-runtime`（默认） | 完整桌面应用（窗口管理、系统通知、自动更新等） | +| `codeg-server` | 无（`--no-default-features`） | 独立服务器（Axum HTTP + WebSocket + 静态服务） | +| `codeg-mcp` | 无 | per-launch stdio MCP 伴生进程（多智能体委托，见 §6.4） | + +支撑这一决策的四个关键抽象： + +- **`EventEmitter` 枚举**（`web/event_bridge.rs`）：`Tauri(AppHandle)` 或 `WebOnly(Arc)`。统一了"实时事件如何离开后端"这一分叉点，业务代码不感知运行模式。 +- **`_core` 后缀函数**（`commands/`）：接受普通引用参数（`&AppDatabase`、`&EventEmitter`），是同一份业务逻辑，同时被 `#[tauri::command]`（桌面）与 Axum handler（服务器）调用。条件编译约定：`#[cfg(feature = "tauri-runtime")]` 隔离桌面专属代码；`#[cfg_attr(feature = "tauri-runtime", tauri::command)]` 让函数两模式可用、仅桌面标记为命令。 +- **`AppState`**（`app_state.rs`）：共享状态容器——`db`、`connection_manager`、`terminal_manager`、`event_broadcaster`、`acp_event_bus`、`emitter`、`delegation_broker` 等。两种模式构造同一个 `AppState`。 +- **前端 `Transport`**（`src/lib/transport/`）：用 `__TAURI_INTERNALS__` 探测环境（`detect.ts`），桌面走 `invoke()`、浏览器走 `fetch()` + WebSocket。上层只面对统一的 `Transport` 接口（`call` / `subscribe` / `attach`）。 + +前端构建同样服务于此：`next.config.ts` 设 `output: "export"` **纯静态导出**，不使用动态路由（`[param]`），一律用查询参数替代——这样同一份前端既能被 Tauri webview 加载，也能被 server 静态服务。 + +--- + +## 2. 后端分层（`src-tauri/src/`） + +``` +┌──────────────────────────────────────────────────────────────┐ +│ lib.rs Tauri 命令注册 + 窗口/生命周期（桌面入口） │ +│ web/ Axum router + handlers + WS + 认证 + 静态服务（服务器入口）│ +│ main.rs / bin/ │ +├──────────────────────────────────────────────────────────────┤ +│ commands/ 业务逻辑层（_core 双模式共用） │ +│ conversations.rs · turn_window.rs · chat_* · delegation.rs │ +│ version_control.rs · work_task.rs · ... │ +├───────────────────────────┬──────────────────────────────────┤ +│ acp/ 实时运行侧（≈4.8 万行）│ parsers/ 历史读取侧 │ +│ connection.rs (1.3w) │ 14 个解析器（每 agent 一个） │ +│ manager.rs │ → 统一 ConversationSummary/Detail│ +│ session_state.rs │ summary_cache（mtime 指纹缓存） │ +│ background_watch.rs │ │ +│ event_stream / internal_bus │ +│ delegation/（子代理 broker）│ │ +├───────────────────────────┴──────────────────────────────────┤ +│ db/（SeaORM + SQLite） terminal/（PTY） process/ supervise/ │ +│ models/（共享数据结构，前端 types.ts 是其 TS 镜像） │ +└──────────────────────────────────────────────────────────────┘ +``` + +后端有**两条几乎独立的子系统**，是理解后端的关键： + +### 2.1 `parsers/` —— 历史读取侧（离线解析） + +每个 agent 一个解析器（`claude.rs`、`codex.rs`、`gemini.rs`、`opencode.rs`、`openclaw.rs`、`cline.rs`、`hermes.rs`、`codebuddy.rs`、`kimi_code.rs`、`pi.rs`、`grok.rs`、`cursor.rs`、`acp_native.rs`…），把散落在本地文件系统的会话文件（JSONL / 整文档 JSON / SQLite）解析成统一的 `ConversationSummary` / `ConversationDetail` / `MessageTurn` 模型（`models/`）。 + +- 出口模型统一，前端只面对一种数据结构。 +- `summary_cache.rs`：进程级缓存，用 `(mtime, size)` 指纹做命中/失效，避免每次列表都全量重解析；带 LRU 上限。 +- `turn_window.rs`：把"全量解析"与"传输窗口切片"分离（见 §6.5）。 + +### 2.2 `acp/` —— 实时运行侧（Agent Client Protocol） + +把 agent CLI 作为子进程拉起，通过 ACP 协议双向通信，是整个系统最大、最核心的子系统： + +| 模块 | 职责 | +|---|---| +| `connection.rs` | 单连接的 actor / 事件循环（读 agent stdout、发 prompt、收 session/update） | +| `manager.rs` | `ConnectionManager`：所有连接的生命周期（spawn/disconnect/idle sweep） | +| `session_state.rs` | 单会话实时状态：`live_message`、`active_tool_calls`、`pending_permission`、`to_snapshot()` | +| `background_watch.rs` | 对 transcript 做增量 tail（与详情解析复用同一个 `ClaudeRecordAccumulator`） | +| `event_stream.rs` | 每连接事件环形缓冲 + per-connection broadcast | +| `internal_bus.rs` | 进程内 ACP 事件总线（见 §6.2） | +| `delegation/` | 多智能体委托 broker（见 §6.4） | +| `process.rs` / `supervise.rs` | 子进程拉起与监督 | + +--- + +## 3. 前端分层（`src/`） + +``` +components/ UI（message/ composer、sidebar、settings、tasks、terminal、merge ...） +contexts/ React Context 横切状态 + acp-connections-context（连接/流式）· conversation-runtime-context + workspace / tab / terminal / delegation / alert ... +stores/ Zustand store（app-workspace-store · conversation-runtime-store · tab-store） +lib/ + transport/ Transport 抽象（tauri-transport vs web-transport；detect/ws-auth/web-auth） + adapters/ AI 响应 → 组件渲染的适配器 + api.ts 主 API 客户端 + types.ts Rust 模型的 TypeScript 镜像 +i18n/ 10 种语言（next-intl，i18n/messages/*.json） +``` + +- **状态管理**：Context（横切订阅/生命周期）+ Zustand（细粒度 store）。`conversation-runtime-store.ts` 管理每个会话的 detail/optimistic turns/live message 与窗口化加载。 +- **路径别名**：`@/*` → `./src/*`。 + +--- + +## 4. 两条数据通路 + +### 4.1 历史读路径（查看旧会话） + +``` +agent CLI 写的会话文件 + → parsers/ 解析成统一模型（summary_cache 加速） + → turn_window 按窗口切片（tail / fromIndex / page） + → 前端 conversation-runtime-store 窗口化装载 + 倒序无限滚动 +``` + +### 4.2 实时运行路径（正在跑的会话） + +``` +ConnectionManager 拉起 agent CLI 子进程 + ⇄ ACP 协议事件（session/update） + → connection.rs 事件循环 → session_state 更新 + → EventEmitter + 桌面: Tauri app.emit + 服务器: WebEventBroadcaster → WebSocket + → 前端 Transport.attach / subscribe + → acp-connections-context（16ms 合帧）→ stores → 渲染 +``` + +--- + +## 5. 数据库（`db/`） + +SeaORM + SQLite。`entities/`（conversation、folder、agent_setting、automation、chat_channel、delegation 等）、`migration/`（按日期命名的迁移）、`service/`（查询封装）。`models/` 与 DB 实体分离：`models/` 是面向传输/解析的共享结构，前端 `types.ts` 与其一一对应。 + +--- + +## 6. 关键深层机制 + +### 6.1 ACP 连接事件流 + +`connection.rs` 是每连接的 actor：从 agent 子进程 stdout 读 ACP `session/update`，转成内部 `AcpEvent`，更新 `SessionState`，再经 `emit_with_state` 发射。`SessionState::to_snapshot()` 产出 wire 友好的 `LiveSessionSnapshot`，供 attach 时的全量快照。 + +### 6.2 双事件总线（`internal_bus` vs `WebEventBroadcaster`） + +后端有**两条**总线，拆分原因写在 `internal_bus.rs` 头部： + +- **`InternalEventBus`**：携带 `Arc`，服务**进程内**消费者（lifecycle、pet_state_mapper、chat-channel 订阅者）。 typed 投递，消费者无需 `serde_json::from_value` 逐事件反序列化。 +- **`WebEventBroadcaster`**：携带 `Arc`，面向 WS 客户端的 JSON 投递。`Arc` 让 N 个订阅者只增加引用计数，不拷贝可能上 MB 的 JSON 树。 + +拆分收益：① 后端消费者省去每事件每订阅者的 JSON 解析；② ACP 事件从全局 firehose 移除后，每连接 attach 流成为唯一通路，前端不再需要按 `connectionId` 去重。 + +### 6.3 Attach 协议（Subscribe-with-Snapshot，`web/ws_attach.rs`） + +取代旧的"订阅全局 firehose + 单独 HTTP 拉快照"两步。客户端对某连接发 `attach`，服务器在 `SessionState` 读锁内**原子地**二选一： + +- **`snapshot`**：全量快照（`since_seq=None`，或缺口太大时）； +- **`replay`**：用每连接环形缓冲增量补齐（`since_seq` 已知时）。 + +之后该连接的实时事件以 `event` 帧沿同一 WebSocket 推送。环形缓冲（`event_stream.rs`）有界：`RECENT_BUFFER_MAX_BYTES=128KB`、`MAX_COUNT=128`、单事件 `64KB` 上限；消费跟不上时 broadcast `Lagged` → 客户端转为重新 attach（snapshot 兜底）。每连接出站 mpsc 容量 64，靠背压自然节流慢客户端。 + +### 6.4 多智能体委托（`acp/delegation/`） + +父 agent 的 LLM 调内建 MCP 工具 `delegate_to_agent`，即可拉起一个**可以是另一种 agent 类型**的全新 ACP 子会话，并把子 agent 首轮的最终 assistant 文本作为 MCP tool_result 返回： + +``` +parent LLM ─┐ ToolUse(delegate_to_agent, ...) + ▼ +parent CLI ──stdio──► codeg-mcp（per-launch 伴生进程） + │ UDS / named pipe（token 鉴权） + ▼ + DelegationBroker + │ ConnectionSpawner trait + ▼ + ConnectionManager.spawn_agent / send_prompt_linked + ▼ + child ACP session ── TurnComplete ──┐ +parent LLM ◄── MCP tool_result ◄── DelegationOutcome ◄┘ +``` + +要点（`broker.rs` / `mod.rs` 文档）： + +- **异步**：`delegate_to_agent` setup 完成即返回 `task_id` ack；LLM 之后用 `get_delegation_status`（可长轮询）取结果，或 `cancel_delegation` 取消。无阻塞 oneshot——运行中的任务只是 `running` map 里的一项，终态事件把它原子迁移进 `completed` 缓存并 `result_notify` 唤醒长轮询。 +- **生命周期**：`start_delegation` → 前置检查（开关、深度上限）→ `ConnectionSpawner.spawn` → 以首个 prompt 下发任务（尾部 `DelegationLink` 携带父 `tool_use_id` 与 broker 内部 `call_id`）→ 注册 `RunningTask` → 终态（`complete_call` / 各类 cancel）解析并拆除子会话。 +- **v1 一次性**：子会话首个 `TurnComplete` 后即 resolve + disconnect，不复用会话。 +- **结果不落库**：子输出不写入 codeg DB，broker 把完成文本缓存在 `completed`（按父作用域、FIFO 上限）。 + +前端侧对应 `delegation-context.tsx`：按 `parent_tool_use_id` 维护父↔子绑定，让父的 `delegate_to_agent` 工具卡内联渲染子会话（带 LRU 上限）。 + +### 6.5 窗口一致性协议（`commands/turn_window.rs`） + +长会话只在序列化前切片传输可视窗口。`tailTurns`/`fromIndex`/翻页都基于**完整解析+完整后处理**的 turn 列表切片，保证窗口响应与全量响应对应区域**字节一致**。每个窗口响应带： + +- `offset` / `total` / `assistant_before`； +- `prefix_hash`：`turns[0..offset)` 的结构指纹（FNV-1a，逐 turn 喂 `(role_tag, timestamp_millis)`）。刻意**不含 turn id**——id 在 turn 运行时会被改写，而 `(role, millis)` 在 id 改写与内容回填下不变，但插入/删除/移位（压缩/改写的真实形态）会改变它，前端据此检测前缀被改写并安全地重置窗口； +- `uncovered_prefix_max_ts`：未覆盖前缀的最大时间戳，驱动前端 overlay 退休判定。 + +`prefix_fingerprint` 由 Rust 与 TypeScript 双实现，并有共享测试向量锁定两端一致。 + +### 6.6 前端流式合帧（`acp-connections-context.tsx`） + +高频 token/tool 事件不直接逐条 setState： + +- 流式增量进入队列，`flushStreamingQueue` 以 **16ms** 定时批量 flush； +- `tool_call_update` 走 **RAF 合帧**（`BATCH_TOOL_CALL_UPDATES`）。 + +二者把"每 token 一次渲染"降为"每帧一次"，是长会话流畅度的关键。 + +--- + +## 7. 设计亮点速览 + +- **单一业务核心 + 特性开关分发**，桌面/服务器不漂移（`EventEmitter` / `_core` / `AppState` / `Transport`）。 +- **`Arc` 负载 + 双总线**：广播零拷贝；进程内消费者免 JSON 解析；前端免去重。 +- **每 agent 一个解析器**，出口模型统一。 +- **Subscribe-with-Snapshot**：原子化 snapshot/replay 抉择 + 有界环形缓冲 + Lagged 自愈。 +- **窗口一致性协议**：结构指纹让前端能检测压缩/改写，安全翻页。 + +--- + +## 8. 目录速查 + +| 路径 | 说明 | +|---|---| +| `src-tauri/src/lib.rs` | 桌面入口、Tauri 命令注册、模块声明 | +| `src-tauri/src/app_state.rs` | 共享状态 `AppState` | +| `src-tauri/src/web/` | Axum router/handlers/WS/认证/静态服务/event_bridge | +| `src-tauri/src/commands/` | 业务逻辑（`_core` 双模式共用） | +| `src-tauri/src/acp/` | ACP 实时运行侧（连接/状态/事件/委托） | +| `src-tauri/src/parsers/` | 每 agent 历史解析器 + summary_cache | +| `src-tauri/src/db/` | SeaORM 实体/迁移/服务 | +| `src-tauri/src/terminal/` `process.rs` `supervise.rs` | PTY 与子进程 | +| `src/lib/transport/` | 前端 Transport 抽象（tauri/web 自动切换） | +| `src/contexts/` `src/stores/` | 前端状态 | +| `src/lib/types.ts` | Rust 模型的 TS 镜像 | diff --git a/src-tauri/src/acp/session_state.rs b/src-tauri/src/acp/session_state.rs index a6711f90a..70ac29f08 100644 --- a/src-tauri/src/acp/session_state.rs +++ b/src-tauri/src/acp/session_state.rs @@ -102,9 +102,6 @@ pub struct ToolCallState { /// surviving page refresh without re-fetching from JSONL. #[serde(default)] pub images: Vec, - /// 流式拼接的 input chunks（serde 不输出，仅运行时用） - #[serde(skip)] - pub raw_input_chunks: Vec, } #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] @@ -1290,7 +1287,6 @@ impl SessionState { locations: None, meta: None, images: Vec::new(), - raw_input_chunks: Vec::new(), }); if let Some(k) = kind { entry.kind = parse_tool_kind(k); @@ -1305,15 +1301,12 @@ impl SessionState { entry.content = Some(c.to_string()); } if let Some(chunk) = raw_input { - entry.raw_input_chunks.push(chunk.to_string()); - // 后端目前发送的是已序列化的 JSON 文本（完整或正在累积）。 - // 对最新片段做尽力解析；解析失败则尝试拼接历史片段。 + // ACP `tool_call_update` 的 rawInput 是累积快照（携带当前已知的完整 + // 输入），不是增量片段——只有最新片段有意义。尽力解析进 `input`；若仍是 + // 未完成的前缀则保留上次成功解析的值。原先把每个快照都 push 再 join， + // 在会话写锁内是 O(k²)，且拼接多个前缀只会得到无效 JSON。 if let Ok(value) = serde_json::from_str::(chunk) { entry.input = Some(value); - } else if let Ok(value) = - serde_json::from_str::(&entry.raw_input_chunks.join("")) - { - entry.input = Some(value); } } if let Some(text) = raw_output { @@ -2987,7 +2980,6 @@ mod tests { }); let entry = s.active_tool_calls.get("tc-1").unwrap(); assert_eq!(entry.input, Some(serde_json::json!({"a": 1}))); - assert_eq!(entry.raw_input_chunks.len(), 2); let snapshot = s.to_snapshot(); assert_eq!(snapshot.connection_id, "conn-test"); @@ -3003,14 +2995,9 @@ mod tests { assert_eq!(snapshot.config_options.as_ref().map(|v| v.len()), Some(1)); assert_eq!(snapshot.active_tool_calls.len(), 1); - // Wire shape: raw_input_chunks must NOT be serialized. + // Wire shape: the latest raw_input snapshot is parsed into `input`. let json = serde_json::to_value(&snapshot).unwrap(); let tc_json = json["active_tool_calls"][0].clone(); - assert!( - tc_json.get("raw_input_chunks").is_none(), - "raw_input_chunks must be #[serde(skip)] (got {})", - tc_json - ); assert_eq!(tc_json["input"], serde_json::json!({"a": 1})); } diff --git a/src-tauri/src/parsers/claude.rs b/src-tauri/src/parsers/claude.rs index f3b7f9edd..23e46a44a 100644 --- a/src-tauri/src/parsers/claude.rs +++ b/src-tauri/src/parsers/claude.rs @@ -1,7 +1,9 @@ +use std::collections::HashMap; use std::fs; use std::io::{BufRead, BufReader}; use std::path::{Path, PathBuf}; -use std::sync::OnceLock; +use std::sync::{Mutex, OnceLock}; +use std::time::{Instant, SystemTime}; use chrono::{DateTime, Utc}; use regex::Regex; @@ -1415,28 +1417,8 @@ impl ClaudeParser { path: &PathBuf, conversation_id: &str, ) -> Result { - // Read the file fully up front: `transcript_watermark` must be EXACTLY - // the byte length this parse consumed. Stat-ing around a streaming read - // could over-claim (bytes appended mid-parse get counted but not read), - // and an over-claiming watermark makes the frontend retire background- - // overlay turns whose content this detail does NOT include — silent - // loss. An exact length risks at most a transient duplicate. - let bytes = fs::read(path)?; - let transcript_watermark = bytes.len() as u64; - - let mut acc = ClaudeRecordAccumulator::new(path.clone()); - for chunk in bytes.split(|b| *b == b'\n') { - // Mirror `BufReader::lines()`: a line that isn't valid UTF-8 is - // skipped (the old loop's per-line `Err(_) => continue`). - let Ok(line) = std::str::from_utf8(chunk) else { - continue; - }; - acc.feed_line(line); - } - acc.finalize_background_lifecycle(); - - let ClaudeRecordAccumulator { - messages, + let RawSessionParse { + transcript_watermark, cwd, git_branch, model, @@ -1445,13 +1427,18 @@ impl ClaudeParser { custom_title, first_timestamp, last_timestamp, - .. - } = acc; + mut turns, + } = raw_session_parse(path)?; let folder_path = cwd.clone(); let folder_name = folder_path.as_ref().map(|p| folder_name_from_path(p)); - let mut turns = group_into_turns(messages); + // Everything below re-runs on EVERY fetch (it is deliberately NOT in the + // cache): these passes resolve against external files that change + // independently of the transcript — patched source files + // (`resolve_patch_line_numbers`) and sub-agent transcripts + // (`attribute_subagent_usage`) — so their output stays fresh even when the + // grouped turns come from the cache. super::relocate_orphaned_tool_results(&mut turns); super::structurize_read_tool_output(&mut turns); super::resolve_patch_line_numbers(&mut turns, cwd.as_deref()); @@ -1503,6 +1490,171 @@ impl ClaudeParser { } } +/// The transcript-only portion of a detail parse: read the session file, run it +/// through the record accumulator, and group into turns. This is the expensive +/// step for a long session (a full read plus a per-line JSON parse), and — unlike +/// the post-processing passes in `parse_conversation_detail` — it is a pure +/// function of the transcript bytes, so its result is cached keyed by the file's +/// `(mtime, size)` fingerprint. A detail fetch for an unchanged file (repeat +/// viewer poll, or each "load older" page of a settled conversation) then skips +/// the full re-read + re-parse and only re-runs the cheap/external post-processing. +#[derive(Clone)] +struct RawSessionParse { + transcript_watermark: u64, + cwd: Option, + git_branch: Option, + model: Option, + title: Option, + ai_title: Option, + custom_title: Option, + first_timestamp: Option>, + last_timestamp: Option>, + turns: Vec, +} + +struct RawSessionCacheEntry { + fingerprint: (Option, u64), + parse: RawSessionParse, + last_used: Instant, +} + +/// Total memory budget across all cached raw parses. A cached entry holds a +/// conversation's grouped turns (its full message content), so this bounds the +/// cache's footprint; `transcript_watermark` (the transcript's byte length) is a +/// faithful per-entry proxy. A single transcript larger than the budget is never +/// cached (its detail is parsed fresh each fetch, exactly as before). +const RAW_SESSION_CACHE_BYTE_BUDGET: u64 = 64 * 1024 * 1024; + +fn raw_session_cache() -> &'static Mutex> { + RAW_SESSION_CACHE.get_or_init(|| Mutex::new(HashMap::new())) +} + +static RAW_SESSION_CACHE: OnceLock>> = OnceLock::new(); + +/// `(mtime, size)` fingerprint of the transcript; `None` (⇒ bypass the cache) +/// when the file can't be stat'd. +fn raw_session_fingerprint(path: &Path) -> Option<(Option, u64)> { + let meta = fs::metadata(path).ok()?; + Some((meta.modified().ok(), meta.len())) +} + +/// Return the raw parse for `path`, serving a cached clone when the file's +/// fingerprint is unchanged. Mirrors `summary_cache::get_or_parse`: the parse +/// runs with the lock released, and a result is only cached when the fingerprint +/// is unchanged across the parse (a racing/streaming write is not memoized). +fn raw_session_parse(path: &PathBuf) -> Result { + let Some(fp) = raw_session_fingerprint(path) else { + return parse_raw_session(path); + }; + + if let Ok(mut map) = raw_session_cache().lock() { + if let Some(entry) = map.get_mut(path) { + if entry.fingerprint == fp { + entry.last_used = Instant::now(); + return Ok(entry.parse.clone()); + } + } + } + + let parsed = parse_raw_session(path)?; + + // Cache only when the file didn't change mid-parse and it fits the budget. + if parsed.transcript_watermark <= RAW_SESSION_CACHE_BYTE_BUDGET + && raw_session_fingerprint(path) == Some(fp) + { + if let Ok(mut map) = raw_session_cache().lock() { + // Bytes already attributed to the entry this insert would replace. + let replaced = map + .get(path) + .map(|e| e.parse.transcript_watermark) + .unwrap_or(0); + let mut total: u64 = map + .values() + .map(|e| e.parse.transcript_watermark) + .sum::() + - replaced; + // Evict least-recently-used entries (never the one being inserted) + // until the new parse fits. + while total + parsed.transcript_watermark > RAW_SESSION_CACHE_BYTE_BUDGET { + let oldest = map + .iter() + .filter(|(p, _)| p.as_path() != path.as_path()) + .min_by_key(|(_, e)| e.last_used) + .map(|(p, _)| p.clone()); + match oldest { + Some(key) => { + if let Some(e) = map.remove(&key) { + total -= e.parse.transcript_watermark; + } + } + None => break, + } + } + map.insert( + path.clone(), + RawSessionCacheEntry { + fingerprint: fp, + parse: parsed.clone(), + last_used: Instant::now(), + }, + ); + } + } + Ok(parsed) +} + +/// Read + accumulate + group the transcript into turns. Pure function of the +/// transcript bytes; the watermark/patch/sub-agent-sensitive work lives in the +/// caller (`parse_conversation_detail`) so it stays fresh. +fn parse_raw_session(path: &PathBuf) -> Result { + // Read the file fully up front: `transcript_watermark` must be EXACTLY the + // byte length this parse consumed. Stat-ing around a streaming read could + // over-claim (bytes appended mid-parse get counted but not read), and an + // over-claiming watermark makes the frontend retire background-overlay turns + // whose content this detail does NOT include — silent loss. An exact length + // risks at most a transient duplicate. + let bytes = fs::read(path)?; + let transcript_watermark = bytes.len() as u64; + + let mut acc = ClaudeRecordAccumulator::new(path.clone()); + for chunk in bytes.split(|b| *b == b'\n') { + // Mirror `BufReader::lines()`: a line that isn't valid UTF-8 is skipped + // (the old loop's per-line `Err(_) => continue`). + let Ok(line) = std::str::from_utf8(chunk) else { + continue; + }; + acc.feed_line(line); + } + acc.finalize_background_lifecycle(); + + let ClaudeRecordAccumulator { + messages, + cwd, + git_branch, + model, + title, + ai_title, + custom_title, + first_timestamp, + last_timestamp, + .. + } = acc; + + let turns = group_into_turns(messages); + Ok(RawSessionParse { + transcript_watermark, + cwd, + git_branch, + model, + title, + ai_title, + custom_title, + first_timestamp, + last_timestamp, + turns, + }) +} + fn parse_timestamp(value: &serde_json::Value) -> Option> { value .get("timestamp") @@ -3025,6 +3177,64 @@ mod tests { detail } + #[test] + fn detail_parse_cache_hit_is_byte_identical_and_invalidates_on_growth() { + let path = std::env::temp_dir().join(format!( + "codeg-claude-detail-cache-{}.jsonl", + uuid::Uuid::new_v4() + )); + let write = |path: &PathBuf, lines: &[serde_json::Value]| { + let mut file = fs::File::create(path).expect("create temp jsonl"); + for line in lines { + writeln!(file, "{line}").unwrap(); + } + }; + let user = serde_json::json!({ + "type": "user", "sessionId": "cache-session", "timestamp": "2026-03-01T10:00:00Z", + "uuid": "u1", "cwd": "/tmp/demo", "gitBranch": "main", + "message": {"content": [{"type": "text", "text": "hi"}]} + }); + let assistant = serde_json::json!({ + "type": "assistant", "sessionId": "cache-session", "timestamp": "2026-03-01T10:00:05Z", + "uuid": "a1", + "message": {"model": "claude-sonnet-4-6", "content": [{"type": "text", "text": "ok"}]} + }); + + let parser = ClaudeParser { + base_dir: PathBuf::new(), + }; + + write(&path, &[user.clone(), assistant.clone()]); + let first = parser + .parse_conversation_detail(&path, "cache-session") + .expect("first parse"); + // Second parse of the unchanged file is served from the raw-parse cache; + // it must be byte-identical to the fresh parse. + let second = parser + .parse_conversation_detail(&path, "cache-session") + .expect("second parse"); + assert_eq!(first.turns.len(), 2); + assert_eq!( + serde_json::to_value(&first).unwrap(), + serde_json::to_value(&second).unwrap(), + "cache hit must be byte-identical to a fresh parse" + ); + + // Appending a turn changes the fingerprint → cache miss → fresh parse. + let assistant2 = serde_json::json!({ + "type": "assistant", "sessionId": "cache-session", "timestamp": "2026-03-01T10:00:09Z", + "uuid": "a2", + "message": {"model": "claude-sonnet-4-6", "content": [{"type": "text", "text": "more"}]} + }); + write(&path, &[user, assistant, assistant2]); + let third = parser + .parse_conversation_detail(&path, "cache-session") + .expect("third parse"); + assert_eq!(third.turns.len(), 3, "growth must invalidate the cache"); + + fs::remove_file(&path).expect("cleanup temp jsonl"); + } + fn total_usage_tokens(detail: &crate::models::ConversationDetail) -> u64 { detail .turns diff --git a/src-tauri/src/parsers/codebuddy.rs b/src-tauri/src/parsers/codebuddy.rs index 617f3867b..d1c7092de 100644 --- a/src-tauri/src/parsers/codebuddy.rs +++ b/src-tauri/src/parsers/codebuddy.rs @@ -150,6 +150,19 @@ impl CodeBuddyParser { &self, path: &Path, conversation_id: &str, + ) -> Result { + // Serve repeat fetches of an unchanged session file (viewer polls, "load + // older" pages) from the detail cache instead of re-parsing the whole + // file. See `summary_cache::detail_get_or_parse` for the contract. + super::summary_cache::detail_get_or_parse(AgentType::CodeBuddy, path, || { + self.parse_detail_uncached(path, conversation_id) + }) + } + + fn parse_detail_uncached( + &self, + path: &Path, + conversation_id: &str, ) -> Result { let reader = BufReader::new(fs::File::open(path)?); @@ -1903,4 +1916,43 @@ mod tests { std::fs::remove_dir_all(&root).ok(); } + + #[test] + fn detail_parse_cache_hit_is_byte_identical_and_invalidates_on_growth() { + let root = std::env::temp_dir().join(format!("codeg-cb-cache-{}", uuid::Uuid::new_v4())); + let sid = "sess-cache"; + let records = vec![ + json!({"type":"message","role":"user","timestamp":1781821844178i64,"cwd":"/Users/demo/app","sessionId":sid, + "content":[{"type":"input_text","text":"hello"}]}), + json!({"type":"message","role":"assistant","timestamp":1781821848958i64,"cwd":"/Users/demo/app","sessionId":sid, + "content":[{"type":"output_text","text":"hi"}]}), + ]; + write_session(&root, "Users-demo-app", sid, &records); + + let parser = CodeBuddyParser::with_base_dir(root.clone()); + let first = parser.get_conversation(sid).expect("first parse"); + // Second parse of the unchanged file is served from the detail cache. + let second = parser.get_conversation(sid).expect("second parse"); + assert!(!first.turns.is_empty()); + assert_eq!( + serde_json::to_value(&first).unwrap(), + serde_json::to_value(&second).unwrap(), + "cache hit must be byte-identical to a fresh parse" + ); + + // Appending a record changes the fingerprint → cache miss → fresh parse. + let mut grown = records.clone(); + grown.push(json!({"type":"message","role":"user","timestamp":1781821849999i64,"cwd":"/Users/demo/app","sessionId":sid, + "content":[{"type":"input_text","text":"again"}]})); + write_session(&root, "Users-demo-app", sid, &grown); + let third = parser.get_conversation(sid).expect("third parse"); + assert!( + third.turns.len() > first.turns.len(), + "growth must invalidate the cache ({} → {})", + first.turns.len(), + third.turns.len() + ); + + std::fs::remove_dir_all(&root).ok(); + } } diff --git a/src-tauri/src/parsers/codex.rs b/src-tauri/src/parsers/codex.rs index 3694f89ec..30792bd03 100644 --- a/src-tauri/src/parsers/codex.rs +++ b/src-tauri/src/parsers/codex.rs @@ -1838,6 +1838,20 @@ impl CodexParser { &self, path: &PathBuf, conversation_id: &str, + ) -> Result { + // Serve repeat fetches of an unchanged rollout (viewer polls, "load + // older" pages) from the process-global detail cache instead of + // re-reading + re-parsing the whole file. See + // `summary_cache::detail_get_or_parse` for the freshness contract. + super::summary_cache::detail_get_or_parse(AgentType::Codex, path, || { + self.parse_conversation_detail_uncached(path, conversation_id) + }) + } + + fn parse_conversation_detail_uncached( + &self, + path: &PathBuf, + conversation_id: &str, ) -> Result { let file = fs::File::open(path)?; let reader = BufReader::new(file); @@ -4256,6 +4270,57 @@ mod tests { detail } + #[test] + fn detail_parse_cache_hit_is_byte_identical_and_invalidates_on_growth() { + let nanos = SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("system time ok") + .as_nanos(); + let path: PathBuf = + env::temp_dir().join(format!("codeg-codex-detail-cache-{nanos}.jsonl")); + let base = concat!( + "{\"timestamp\":\"2026-03-01T10:00:00Z\",\"type\":\"session_meta\",\"payload\":{\"id\":\"cache-1\",\"cwd\":\"/tmp/demo\"}}\n", + "{\"timestamp\":\"2026-03-01T10:00:01Z\",\"type\":\"turn_context\",\"payload\":{\"model\":\"gpt-5-codex\"}}\n", + "{\"timestamp\":\"2026-03-01T10:00:02Z\",\"type\":\"event_msg\",\"payload\":{\"type\":\"user_message\",\"message\":\"hi\"}}\n", + "{\"timestamp\":\"2026-03-01T10:00:03Z\",\"type\":\"event_msg\",\"payload\":{\"type\":\"agent_message\",\"message\":\"hello\"}}\n", + ); + fs::write(&path, base).expect("write rollout"); + + let parser = CodexParser::new(); + let first = parser + .parse_conversation_detail(&path, "cache-1") + .expect("first parse"); + // Second parse of the unchanged file is served from the detail cache — + // it must be byte-identical to the fresh parse. + let second = parser + .parse_conversation_detail(&path, "cache-1") + .expect("second parse"); + assert!(!first.turns.is_empty()); + assert_eq!( + serde_json::to_value(&first).unwrap(), + serde_json::to_value(&second).unwrap(), + "cache hit must be byte-identical to a fresh parse" + ); + + // Appending a turn changes the fingerprint → cache miss → fresh parse. + let mut grown = base.to_string(); + grown.push_str( + "{\"timestamp\":\"2026-03-01T10:00:09Z\",\"type\":\"event_msg\",\"payload\":{\"type\":\"user_message\",\"message\":\"again\"}}\n", + ); + fs::write(&path, grown).expect("grow rollout"); + let third = parser + .parse_conversation_detail(&path, "cache-1") + .expect("third parse"); + assert!( + third.turns.len() > first.turns.len(), + "growth must invalidate the cache ({} → {})", + first.turns.len(), + third.turns.len() + ); + + let _ = fs::remove_file(&path); + } + #[test] fn every_model_round_trip_of_a_turn_is_counted() { // Codex emits a `token_count` after each model call, so one turn that diff --git a/src-tauri/src/parsers/gemini.rs b/src-tauri/src/parsers/gemini.rs index e8829cbd7..8c55117fe 100644 --- a/src-tauri/src/parsers/gemini.rs +++ b/src-tauri/src/parsers/gemini.rs @@ -118,7 +118,13 @@ impl GeminiParser { continue; } - let value: Value = serde_json::from_str(trimmed).ok()?; + let value: Value = match serde_json::from_str(trimmed) { + Ok(value) => value, + // A torn final line is normal for a session file the CLI is + // still writing; skip it instead of discarding every good + // line parsed so far (matches the other JSONL parsers). + Err(_) => continue, + }; let Some(object) = value.as_object() else { continue; }; @@ -959,6 +965,47 @@ mod tests { let _ = fs::remove_dir_all(base); } + #[test] + fn gemini_jsonl_skips_torn_final_line_instead_of_dropping_conversation() { + let nanos = SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("system time ok") + .as_nanos(); + let base: PathBuf = env::temp_dir().join(format!("codeg-gemini-torn-{nanos}")); + let chats_dir = base.join("tmp").join("codeg").join("chats"); + fs::create_dir_all(&chats_dir).expect("create chat dir"); + fs::write( + base.join("tmp").join("codeg").join(".project_root"), + "/Users/test/workspace/demo", + ) + .expect("write project root"); + + // Two well-formed records followed by a truncated final line — the + // shape of a session file read while the CLI is mid-write. + let file_path = chats_dir.join("session-2026-03-02T04-30-torn.jsonl"); + let content = "{\"sessionId\":\"jsonl-torn\",\"startTime\":\"2026-03-02T04:30:20.796Z\",\"lastUpdated\":\"2026-03-02T04:33:13.631Z\"}\n\ + {\"id\":\"u1\",\"timestamp\":\"2026-03-02T04:30:20.796Z\",\"type\":\"user\",\"content\":[{\"text\":\"hello\"}]}\n\ + {\"id\":\"a1\",\"timestamp\":\"2026-03-02T04:33:13.631Z\",\"type\":\"gemini\",\"content\":\"hi\"}\n\ + {\"id\":\"a2\",\"timestamp\":\"2026-03-02T04:34:00.000Z\",\"type\":\"gemini\",\"content\":\"truncat"; + fs::write(&file_path, content).expect("write chat file"); + + let parser = GeminiParser::with_base_dir(base.clone()); + let summaries = parser.list_conversations().expect("list conversations"); + assert_eq!(summaries.len(), 1, "torn line must not hide the session"); + assert_eq!(summaries[0].id, "jsonl-torn".to_string()); + + let detail = parser + .get_conversation("jsonl-torn") + .expect("get conversation"); + assert_eq!( + detail.turns.len(), + 2, + "good lines before the torn one survive" + ); + + let _ = fs::remove_dir_all(base); + } + #[test] fn gemini_prefers_update_topic_title_over_first_user_message() { let nanos = SystemTime::now() diff --git a/src-tauri/src/parsers/mod.rs b/src-tauri/src/parsers/mod.rs index e567d6fe9..5941bc4fa 100644 --- a/src-tauri/src/parsers/mod.rs +++ b/src-tauri/src/parsers/mod.rs @@ -16,6 +16,7 @@ mod summary_cache; use std::collections::{HashMap, HashSet}; use std::path::PathBuf; +use std::rc::Rc; use std::sync::OnceLock; /// A root of external agent-CLI transcript data, archived under @@ -885,6 +886,9 @@ pub fn strip_numbered_lines(text: &str) -> Option { /// file from disk and matches the context lines to calculate real line numbers. /// Falls back gracefully if the file doesn't exist or context doesn't match. pub fn resolve_patch_line_numbers(turns: &mut [MessageTurn], cwd: Option<&str>) { + // One cache for the whole pass: an edit-heavy session touches the same + // file from many patch blocks, and each block used to re-read it from disk. + let mut file_cache = FileLinesCache::new(); for turn in turns.iter_mut() { for block in turn.blocks.iter_mut() { if let ContentBlock::ToolUse { @@ -902,7 +906,9 @@ pub fn resolve_patch_line_numbers(turns: &mut [MessageTurn], cwd: Option<&str>) } if let Some(ref text) = input_preview { if text.contains("@@\n") || text.contains("@@\r\n") { - if let Some(resolved) = resolve_patch_text(text, cwd) { + if let Some(resolved) = + resolve_patch_text_with_cache(text, cwd, &mut file_cache) + { *input_preview = Some(resolved); } } @@ -914,9 +920,20 @@ pub fn resolve_patch_line_numbers(turns: &mut [MessageTurn], cwd: Option<&str>) /// Resolve a single patch text, replacing bare `@@` with `@@ -N,M +N,M @@`. pub fn resolve_patch_text(patch: &str, cwd: Option<&str>) -> Option { + resolve_patch_text_with_cache(patch, cwd, &mut FileLinesCache::new()) +} + +/// `resolve_patch_text` driven by a caller-supplied per-parse file cache, so a +/// full `resolve_patch_line_numbers` pass reads each patched file from disk at +/// most once no matter how many tool blocks touch it. +fn resolve_patch_text_with_cache( + patch: &str, + cwd: Option<&str>, + cache: &mut FileLinesCache, +) -> Option { let mut output = String::with_capacity(patch.len() + 256); let mut current_file_path: Option = None; - let mut file_lines: Option> = None; + let mut file_lines: Option>> = None; let mut any_resolved = false; let lines: Vec<&str> = patch.lines().collect(); @@ -934,7 +951,7 @@ pub fn resolve_patch_text(patch: &str, cwd: Option<&str>) -> Option { }; let path = line[marker_end..].trim(); current_file_path = Some(path.to_string()); - file_lines = load_file_lines(path, cwd); + file_lines = load_file_lines_cached(path, cwd, cache); output.push_str(line); output.push('\n'); i += 1; @@ -997,6 +1014,28 @@ pub fn load_file_lines(path: &str, cwd: Option<&str>) -> Option> { None } +/// Per-parse cache of file contents for patch line-number resolution, keyed by +/// the patch's file path. Caches hits AND misses, so a file patched across many +/// tool blocks (or a referenced-but-missing file) is read from disk at most once +/// per `resolve_patch_line_numbers` call instead of once per block. Resolution +/// reads a point-in-time snapshot of the file, so sharing one read across the +/// blocks of a single parse does not change the output. +type FileLinesCache = HashMap>>>; + +/// `load_file_lines` through a per-parse `FileLinesCache`. +fn load_file_lines_cached( + path: &str, + cwd: Option<&str>, + cache: &mut FileLinesCache, +) -> Option>> { + if let Some(cached) = cache.get(path) { + return cached.clone(); + } + let loaded = load_file_lines(path, cwd).map(Rc::new); + cache.insert(path.to_string(), loaded.clone()); + loaded +} + /// Collect lines belonging to a hunk (until next `@@` or `*** ` marker or end). fn collect_hunk_lines<'a>(lines: &'a [&'a str], start: usize) -> Vec<&'a str> { let mut result = Vec::new(); @@ -1165,8 +1204,9 @@ mod tests { use super::{ backfill_turn_durations, fold_reference_links, infer_context_window_max_tokens, - is_safe_subagent_id, latest_turn_total_usage_tokens, merge_context_window_stats, - path_eq_for_matching, title_from_user_text, + is_safe_subagent_id, latest_turn_total_usage_tokens, load_file_lines, + load_file_lines_cached, merge_context_window_stats, path_eq_for_matching, + resolve_patch_text, title_from_user_text, FileLinesCache, }; use crate::models::{MessageTurn, SessionStats, TurnRole, TurnUsage}; @@ -1537,4 +1577,57 @@ mod tests { "C:/Users/demo/workspace/codeg" )); } + + fn temp_dir(tag: &str) -> std::path::PathBuf { + let nanos = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .expect("system time") + .as_nanos(); + let dir = std::env::temp_dir().join(format!("codeg-patch-{tag}-{nanos}")); + std::fs::create_dir_all(&dir).expect("create temp dir"); + dir + } + + #[test] + fn resolve_patch_text_resolves_bare_hunk_against_file_on_disk() { + let dir = temp_dir("resolve"); + let file = dir.join("demo.txt"); + std::fs::write(&file, "line one\nline two\nline three\nline four\nline five\n") + .expect("write file"); + + let patch = format!( + "*** Update File: {}\n@@\n line two\n-line three\n+line THREE\n line four\n", + file.display() + ); + let resolved = resolve_patch_text(&patch, None).expect("should resolve"); + assert!(resolved.contains("@@ -2,3 +2,3 @@"), "got: {resolved}"); + + let _ = std::fs::remove_dir_all(&dir); + } + + #[test] + fn file_lines_cache_serves_repeated_reads_without_hitting_disk_again() { + let dir = temp_dir("cache"); + let file = dir.join("demo.txt"); + std::fs::write(&file, "alpha\nbeta\n").expect("write file"); + let path = file.to_string_lossy().to_string(); + + let mut cache = FileLinesCache::new(); + let first = load_file_lines_cached(&path, None, &mut cache); + assert!(first.is_some()); + + // Delete the file: a second cached read still succeeds (served from the + // cache), while a direct uncached read now fails. + std::fs::remove_file(&file).expect("remove file"); + let second = load_file_lines_cached(&path, None, &mut cache); + assert_eq!(first, second, "cache should serve the deleted file"); + assert!(load_file_lines(&path, None).is_none()); + + // Misses are cached as well. + let missing = dir.join("nope.txt").to_string_lossy().to_string(); + assert!(load_file_lines_cached(&missing, None, &mut cache).is_none()); + assert!(load_file_lines_cached(&missing, None, &mut cache).is_none()); + + let _ = std::fs::remove_dir_all(&dir); + } } diff --git a/src-tauri/src/parsers/pi.rs b/src-tauri/src/parsers/pi.rs index 77e196003..de533d37f 100644 --- a/src-tauri/src/parsers/pi.rs +++ b/src-tauri/src/parsers/pi.rs @@ -137,6 +137,19 @@ impl PiParser { &self, path: &Path, conversation_id: &str, + ) -> Result { + // Serve repeat fetches of an unchanged session file (viewer polls, "load + // older" pages) from the detail cache instead of re-parsing the whole + // file. See `summary_cache::detail_get_or_parse` for the contract. + super::summary_cache::detail_get_or_parse(AgentType::Pi, path, || { + self.parse_detail_uncached(path, conversation_id) + }) + } + + fn parse_detail_uncached( + &self, + path: &Path, + conversation_id: &str, ) -> Result { let parsed = parse_session(path); @@ -1103,4 +1116,40 @@ mod tests { Err(ParseError::ConversationNotFound(_)) )); } + + #[test] + fn detail_parse_cache_hit_is_byte_identical_and_invalidates_on_growth() { + let dir = tempdir().expect("tempdir"); + let base = dir.path(); + let id = "0f3c1d2e-1111-2222-3333-444455556666"; + let dashed = "--Users-demo-my-app--"; + let filename = "2026-06-27T10-00-00_0f3c1d2e.jsonl"; + write_session(base, dashed, filename, &sample_records(id)); + let path = base.join(dashed).join(filename); + + let parser = PiParser::with_base_dir(base.to_path_buf()); + let first = parser.parse_detail(&path, id).expect("first parse"); + // Second parse of the unchanged file is served from the detail cache — + // it must be byte-identical to the fresh parse. + let second = parser.parse_detail(&path, id).expect("second parse"); + assert!(!first.turns.is_empty()); + assert_eq!( + serde_json::to_value(&first).unwrap(), + serde_json::to_value(&second).unwrap(), + "cache hit must be byte-identical to a fresh parse" + ); + + // Appending a record changes the fingerprint → cache miss → fresh parse. + let mut records = sample_records(id); + records.push(json!({"type":"message","id":"m5","parentId":"m4","timestamp":"2026-06-27T10:00:11.000Z", + "message":{"role":"user","content":"thanks"}})); + write_session(base, dashed, filename, &records); + let third = parser.parse_detail(&path, id).expect("third parse"); + assert!( + third.turns.len() > first.turns.len(), + "growth must invalidate the cache ({} → {})", + first.turns.len(), + third.turns.len() + ); + } } diff --git a/src-tauri/src/parsers/summary_cache.rs b/src-tauri/src/parsers/summary_cache.rs index 858516541..1c90f2bff 100644 --- a/src-tauri/src/parsers/summary_cache.rs +++ b/src-tauri/src/parsers/summary_cache.rs @@ -56,10 +56,10 @@ use std::collections::HashMap; use std::path::{Path, PathBuf}; use std::sync::{Mutex, OnceLock}; -use std::time::SystemTime; +use std::time::{Instant, SystemTime}; use super::ParseError; -use crate::models::{AgentType, ConversationSummary}; +use crate::models::{AgentType, ConversationDetail, ConversationSummary}; /// File-content fingerprint: `(mtime, size)`. `mtime` is `Option` because a /// platform may not report it; a `None`/`None` comparison then falls back to @@ -70,6 +70,10 @@ struct CacheEntry { fingerprint: Fingerprint, /// Only positive summaries are stored — see `get_or_parse`. summary: ConversationSummary, + /// Last time this entry was served or written, for LRU eviction. The cache + /// is process-global and insert-only otherwise, so entries for long-deleted + /// session files would otherwise accumulate for the app's lifetime. + last_used: Instant, } /// Nested `AgentType → (path → entry)` so a lookup borrows `&Path` (no @@ -89,6 +93,30 @@ fn fingerprint(path: &Path) -> Option { Some((meta.modified().ok(), meta.len())) } +/// Bound on cached summaries per agent. Generous enough that eviction is rare in +/// practice (a heavy user has hundreds of sessions of one agent, and a dormant +/// hit costs only a `stat`), yet it stops the process-global map from growing +/// for the app's lifetime as sessions are created and deleted. Eviction is never +/// *incorrect*: the `(mtime, size)` fingerprint revalidates on the next access, +/// so an evicted entry is simply re-parsed. +const MAX_ENTRIES_PER_AGENT: usize = 1024; + +/// Make room for inserting `path` by evicting the least-recently-used entry when +/// `per_agent` is already at `cap`. No-op when `path` is an existing key (a +/// re-insert replaces in place and does not grow the map). +fn evict_lru_for_insert(per_agent: &mut HashMap, cap: usize, path: &Path) { + if per_agent.len() < cap || per_agent.contains_key(path) { + return; + } + if let Some(oldest) = per_agent + .iter() + .min_by_key(|(_, entry)| entry.last_used) + .map(|(key, _)| key.clone()) + { + per_agent.remove(&oldest); + } +} + /// Return the cached summary for `(agent_type, path)` if the file is unchanged /// since it was last parsed; otherwise run `parse`, cache a positive result, and /// return it. @@ -129,9 +157,10 @@ where // Fast path: a live fingerprint match returns the cached summary without // reading or parsing the file. - if let Ok(map) = cache().lock() { - if let Some(entry) = map.get(&agent_type).and_then(|m| m.get(path)) { + if let Ok(mut map) = cache().lock() { + if let Some(entry) = map.get_mut(&agent_type).and_then(|m| m.get_mut(path)) { if entry.fingerprint == fp_before { + entry.last_used = Instant::now(); return Ok(Some(entry.summary.clone())); } } @@ -147,11 +176,14 @@ where if let Some(summary) = &parsed { if fingerprint(path) == Some(fp_before) { if let Ok(mut map) = cache().lock() { - map.entry(agent_type).or_default().insert( + let per_agent = map.entry(agent_type).or_default(); + evict_lru_for_insert(per_agent, MAX_ENTRIES_PER_AGENT, path); + per_agent.insert( path.to_path_buf(), CacheEntry { fingerprint: fp_before, summary: summary.clone(), + last_used: Instant::now(), }, ); } @@ -160,6 +192,119 @@ where Ok(parsed) } +/// Process-global cache of FULL conversation details, keyed by `(AgentType, path)`. +/// +/// `get_conversation` re-parses a session transcript on every detail fetch — +/// viewer polls, and every "load older" page of the reverse-infinite scroll. +/// For a long session that is a full read + per-line parse each time. This cache +/// memoizes the parsed `ConversationDetail` per `(AgentType, path)`, invalidated +/// by the same cheap `(mtime, size)` fingerprint as the summary cache, so a +/// repeat fetch of an unchanged file skips the re-parse entirely. +/// +/// Memory is bounded by `DETAIL_CACHE_BYTE_BUDGET` across all entries, using the +/// transcript's byte length as a faithful per-entry footprint proxy; a single +/// transcript larger than the budget is never cached (parsed fresh each fetch, +/// exactly as before). Eviction is least-recently-used. +/// +/// Freshness contract (per parser): the fingerprint covers the transcript file. +/// Work that reads OTHER files must either be excluded from the cached region or +/// accepted as last-parse state — e.g. Codex's sub-agent capsule stats and +/// patch line numbers reflect the file as of the last parse; both are cosmetic +/// and self-correct the moment the transcript itself changes (an active collab +/// session appends to its rollout, so the fingerprint changes and a fresh parse +/// runs). Records lacking a timestamp fall back to `Utc::now()` during a parse; +/// a cached entry keeps the first such fallback, which keeps the window +/// `prefix_hash` STABLE across fetches of an unchanged file — without the cache, +/// each re-parse would mint a fresh `now()` and the frontend would misread the +/// unchanged prefix as rewritten. +const DETAIL_CACHE_BYTE_BUDGET: u64 = 64 * 1024 * 1024; + +struct DetailCacheEntry { + fingerprint: Fingerprint, + /// Transcript byte length — proxy for the detail's in-memory footprint. + size: u64, + detail: ConversationDetail, + last_used: Instant, +} + +/// Keyed by `(AgentType, path)` — namespaced by parser like the summary cache. +type DetailCache = HashMap<(AgentType, PathBuf), DetailCacheEntry>; + +fn detail_cache() -> &'static Mutex { + static CACHE: OnceLock> = OnceLock::new(); + CACHE.get_or_init(|| Mutex::new(HashMap::new())) +} + +/// Return the cached detail for `(agent_type, path)` when the transcript is +/// unchanged; otherwise run `parse` (lock released) and cache a real (non-empty) +/// result. The caller receives a CLONE, so it may freely mutate the result +/// (inject delegation meta, set counts) without disturbing the cached entry. +/// +/// Only a detail with at least one turn is cached: an empty/degenerate parse +/// derives fields like `started_at` from `Utc::now()`, which a cache would +/// otherwise freeze. +pub(crate) fn detail_get_or_parse( + agent_type: AgentType, + path: &Path, + parse: F, +) -> Result +where + F: FnOnce() -> Result, +{ + // Can't fingerprint ⇒ bypass the cache (parse fresh, store nothing). + let Some(fp) = fingerprint(path) else { + return parse(); + }; + let key = (agent_type, path.to_path_buf()); + + if let Ok(mut map) = detail_cache().lock() { + if let Some(entry) = map.get_mut(&key) { + if entry.fingerprint == fp { + entry.last_used = Instant::now(); + return Ok(entry.detail.clone()); + } + } + } + + let parsed = parse()?; + + // Cache only a real conversation, only if it fits the budget, and only if + // the file didn't change while we read it. + let size = fp.1; + if !parsed.turns.is_empty() && size <= DETAIL_CACHE_BYTE_BUDGET && fingerprint(path) == Some(fp) + { + if let Ok(mut map) = detail_cache().lock() { + let replaced = map.get(&key).map(|e| e.size).unwrap_or(0); + let mut total: u64 = map.values().map(|e| e.size).sum::() - replaced; + while total + size > DETAIL_CACHE_BYTE_BUDGET { + let oldest = map + .iter() + .filter(|(k, _)| **k != key) + .min_by_key(|(_, e)| e.last_used) + .map(|(k, _)| k.clone()); + match oldest { + Some(k) => { + if let Some(e) = map.remove(&k) { + total -= e.size; + } + } + None => break, + } + } + map.insert( + key, + DetailCacheEntry { + fingerprint: fp, + size, + detail: parsed.clone(), + last_used: Instant::now(), + }, + ); + } + } + Ok(parsed) +} + #[cfg(test)] mod tests { use super::*; @@ -382,4 +527,37 @@ mod tests { assert_eq!(cb2.id, "cb"); // served from cache, closure not run assert_eq!(cb_calls.get(), 0); } + + #[test] + fn evicts_the_least_recently_used_entry_at_capacity() { + let now = Instant::now(); + let mut per_agent: HashMap = HashMap::new(); + for (i, name) in ["a", "b", "c"].iter().enumerate() { + per_agent.insert( + PathBuf::from(name), + CacheEntry { + fingerprint: (None, i as u64), + summary: dummy(name), + last_used: now + std::time::Duration::from_millis(i as u64), + }, + ); + } + + // Full (cap 3), inserting a new key "d" evicts "a" (oldest last_used). + evict_lru_for_insert(&mut per_agent, 3, Path::new("d")); + assert!(!per_agent.contains_key(Path::new("a"))); + assert!(per_agent.contains_key(Path::new("b"))); + assert!(per_agent.contains_key(Path::new("c"))); + assert_eq!(per_agent.len(), 2); + + // Re-inserting an existing key never evicts (it replaces in place). + evict_lru_for_insert(&mut per_agent, 3, Path::new("b")); + assert_eq!(per_agent.len(), 2); + assert!(per_agent.contains_key(Path::new("b"))); + + // Below capacity, nothing is evicted. + evict_lru_for_insert(&mut per_agent, 3, Path::new("z")); + assert!(per_agent.contains_key(Path::new("b"))); + assert!(per_agent.contains_key(Path::new("c"))); + } } diff --git a/src-tauri/src/terminal/manager.rs b/src-tauri/src/terminal/manager.rs index c1dccfe89..9b7819bb4 100644 --- a/src-tauri/src/terminal/manager.rs +++ b/src-tauri/src/terminal/manager.rs @@ -502,7 +502,12 @@ fn read_loop( terminals: &Arc>>, ) { let output_event = format!("terminal://output/{}", terminal_id); - let mut buf = [0u8; 8192]; + // 64KB read buffer instead of 8KB. A blocking `read` returns as soon as any + // bytes are available — it does not wait to fill the buffer — so interactive + // latency is unchanged, while bulk output (build logs, `cat` of a large + // file) is batched into ~8x fewer broadcast events. Heap-allocated to keep + // the thread stack small. + let mut buf = vec![0u8; 64 * 1024]; loop { match reader.read(&mut buf) { @@ -513,7 +518,7 @@ fn read_loop( terminal_id: terminal_id.clone(), data, }; - crate::web::event_bridge::emit_event(emitter, &output_event, event.clone()); + crate::web::event_bridge::emit_event(emitter, &output_event, event); } Err(_) => break, } diff --git a/src/components/settings/web-service-settings.tsx b/src/components/settings/web-service-settings.tsx index b31f61f7f..50ab237bc 100644 --- a/src/components/settings/web-service-settings.tsx +++ b/src/components/settings/web-service-settings.tsx @@ -41,7 +41,7 @@ import { const DEFAULT_PORT = 3080 import { openUrl } from "@/lib/platform" -import { copyTextToClipboard } from "@/lib/utils" +import { copyTextToClipboard, randomUUID } from "@/lib/utils" import { useCopiedFlag } from "@/hooks/use-copied-flag" // Remembers which reachable address the user last chose to display/open. @@ -189,12 +189,10 @@ function AddressQrcodeDialog({ } function generateRandomToken() { - if (typeof crypto !== "undefined" && "randomUUID" in crypto) { - return crypto.randomUUID().replace(/-/g, "") - } - return Array.from({ length: 32 }, () => - Math.floor(Math.random() * 16).toString(16) - ).join("") + // randomUUID() falls back to crypto.getRandomValues() in non-secure contexts + // (the server served over plain HTTP on a LAN), so the token stays + // cryptographically random there instead of degrading to Math.random(). + return randomUUID().replace(/-/g, "") } function TokenEditor({ diff --git a/src/contexts/acp-connections-context.tsx b/src/contexts/acp-connections-context.tsx index 9a39881bc..82992e6c7 100644 --- a/src/contexts/acp-connections-context.tsx +++ b/src/contexts/acp-connections-context.tsx @@ -4660,6 +4660,7 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { reverseMapRef.current.delete(conn.connectionId) pendingUnmappedEventsRef.current.delete(conn.connectionId) lastActivityRef.current.delete(contextKey) + alertedErrorDetailsRef.current.delete(contextKey) dispatch({ type: "CONNECTION_REMOVED", contextKey }) return } @@ -4667,6 +4668,7 @@ export function AcpConnectionsProvider({ children }: { children: ReactNode }) { reverseMapRef.current.delete(conn.connectionId) teardownAttachSubscription(contextKey) lastActivityRef.current.delete(contextKey) + alertedErrorDetailsRef.current.delete(contextKey) pendingUnmappedEventsRef.current.delete(conn.connectionId) dispatch({ type: "CONNECTION_REMOVED", contextKey }) }, diff --git a/src/contexts/delegation-context.test.tsx b/src/contexts/delegation-context.test.tsx index bc433b3dd..055552bbd 100644 --- a/src/contexts/delegation-context.test.tsx +++ b/src/contexts/delegation-context.test.tsx @@ -4,6 +4,7 @@ import type { ReactNode } from "react" import { DelegationProvider, + MAX_DELEGATION_BINDINGS, useDelegation, } from "@/contexts/delegation-context" import type { EventEnvelope } from "@/lib/types" @@ -52,6 +53,28 @@ function renderProvider(children: ReactNode = null) { ) } +/** Probe that reports whether two specific parent_tool_use_ids resolve, so a + * test can assert the oldest binding was evicted while the newest survives. */ +function EvictProbe({ + oldestId, + newestId, +}: { + oldestId: string + newestId: string +}) { + const { findByParentToolUseId } = useDelegation() + return ( +
+
+ {findByParentToolUseId(oldestId)?.status ?? "none"} +
+
+ {findByParentToolUseId(newestId)?.status ?? "none"} +
+
+ ) +} + /** Wait until the provider has registered its `useAcpEvent` handler. The * capture is synchronous on mount, so this resolves on the first check; it * stays a waitFor for resilience and must run with REAL timers. */ @@ -232,4 +255,33 @@ describe("DelegationProvider", () => { // Detach was canceled by the re-arriving start event. expect(mockDetach).not.toHaveBeenCalled() }) + + it("evicts the oldest live bindings once MAX_DELEGATION_BINDINGS is exceeded", async () => { + // Long-session leak guard: bindings were previously never removed, so a + // delegation-heavy session grew the map (and the O(n) clone per event) + // without bound. The oldest entries must be evicted; the newest resolve. + const newestId = `pt-${MAX_DELEGATION_BINDINGS + 9}` + render( + + + + ) + await awaitHandlerCaptured() + + act(() => { + for (let i = 0; i < MAX_DELEGATION_BINDINGS + 10; i++) { + capturedHandler!({ + type: "delegation_started", + parent_connection_id: "p1", + parent_tool_use_id: `pt-${i}`, + child_connection_id: `c${i}`, + child_conversation_id: i, + agent_type: "codex", + } as unknown as EventEnvelope) + } + }) + + expect(screen.getByTestId("oldest")).toHaveTextContent("none") + expect(screen.getByTestId("newest")).toHaveTextContent("running") + }) }) diff --git a/src/contexts/delegation-context.tsx b/src/contexts/delegation-context.tsx index 1907b5d93..842edf398 100644 --- a/src/contexts/delegation-context.tsx +++ b/src/contexts/delegation-context.tsx @@ -73,6 +73,36 @@ export function useDelegation(): DelegationContextValue { * before falling through to the DB-persisted view. */ const CHILD_DETACH_GRACE_MS = 2_000 +/** Bound on live delegation bindings. A delegation-heavy session adds one + * entry per sub-agent and previously never removed any, so the map — and the + * O(n) `new Map` clone on every event — grew for the app's lifetime. + * Bindings are a *live* lookup: an evicted entry's parent card falls back to + * the persisted child via the tool-call snapshot meta (see + * `useDelegatedSubSession`'s `fallbackChildConversationId`), exactly as it + * already does for conversations resumed from disk. Keeping only the most + * recent bindings bounds memory without disturbing delegations the user is + * actually watching. */ +export const MAX_DELEGATION_BINDINGS = 200 + +/** Insert-or-update `key`, then evict the oldest entries beyond + * `MAX_DELEGATION_BINDINGS`. delete+set refreshes insertion order so `key` + * counts as most-recent; eviction drops the least-recently-touched bindings. */ +export function setDelegationBinding( + prev: ReadonlyMap, + key: string, + value: DelegationBinding +): Map { + const next = new Map(prev) + next.delete(key) + next.set(key, value) + while (next.size > MAX_DELEGATION_BINDINGS) { + const oldest = next.keys().next().value + if (oldest === undefined) break + next.delete(oldest) + } + return next +} + export function DelegationProvider({ children }: { children: ReactNode }) { const { attachDelegationChild, detachDelegationChild } = useAcpActions() const [byToolUseId, setByToolUseId] = useState< @@ -121,11 +151,9 @@ export function DelegationProvider({ children }: { children: ReactNode }) { task: envelope.task_preview ?? null, taskId: envelope.task_id ?? null, } - setByToolUseId((prev) => { - const m = new Map(prev) - m.set(envelope.parent_tool_use_id, next) - return m - }) + setByToolUseId((prev) => + setDelegationBinding(prev, envelope.parent_tool_use_id, next) + ) // Cancel any pending detach for this parent_tool_use_id — // delegation_started can be replayed after a partial flow // (e.g. reconnect), and an in-flight detach would tear the @@ -173,9 +201,11 @@ export function DelegationProvider({ children }: { children: ReactNode }) { status: "err", errorCode: envelope.result.error_code, } - const m = new Map(prev) - m.set(envelope.parent_tool_use_id, updated) - return m + return setDelegationBinding( + prev, + envelope.parent_tool_use_id, + updated + ) }) // Schedule detach of the synthetic child entry. We keep it diff --git a/src/hooks/use-agent-install-stream.test.tsx b/src/hooks/use-agent-install-stream.test.tsx new file mode 100644 index 000000000..8f738d4c7 --- /dev/null +++ b/src/hooks/use-agent-install-stream.test.tsx @@ -0,0 +1,109 @@ +import { act, renderHook } from "@testing-library/react" +import type { Mock } from "vitest" +import { beforeEach, describe, expect, it, vi } from "vitest" + +type InstallEvent = { task_id: string; kind: string; payload: string } +type Handler = (event: InstallEvent) => void + +vi.mock("@/lib/platform", () => ({ + subscribe: vi.fn(), +})) + +import { subscribe } from "@/lib/platform" +import { MAX_INSTALL_LOG_LINES } from "@/lib/install-stream" +import { useAgentInstallStream } from "./use-agent-install-stream" + +const mockSubscribe = subscribe as unknown as Mock + +let handlers: Handler[] +let unsubs: Array> + +beforeEach(() => { + handlers = [] + unsubs = [] + mockSubscribe.mockReset() + mockSubscribe.mockImplementation(async (_event: string, handler: Handler) => { + handlers.push(handler) + const unsub = vi.fn() + unsubs.push(unsub) + return unsub + }) +}) + +function emit(event: InstallEvent) { + for (const handler of handlers) handler(event) +} + +describe("useAgentInstallStream", () => { + it("caps logs at MAX_INSTALL_LOG_LINES, keeping the tail", async () => { + const { result } = renderHook(() => useAgentInstallStream()) + await act(async () => { + await result.current.start("task-1") + }) + + act(() => { + for (let i = 0; i < MAX_INSTALL_LOG_LINES + 100; i++) { + emit({ task_id: "task-1", kind: "log", payload: `line-${i}` }) + } + }) + + expect(result.current.logs).toHaveLength(MAX_INSTALL_LOG_LINES) + expect(result.current.logs[0]).toBe("line-100") + expect(result.current.logs[result.current.logs.length - 1]).toBe( + `line-${MAX_INSTALL_LOG_LINES + 99}` + ) + }) + + it("ignores events for other task ids", async () => { + const { result } = renderHook(() => useAgentInstallStream()) + await act(async () => { + await result.current.start("task-1") + }) + + act(() => { + emit({ task_id: "other-task", kind: "log", payload: "nope" }) + emit({ task_id: "task-1", kind: "log", payload: "yes" }) + }) + + expect(result.current.logs).toEqual(["yes"]) + }) + + it("unsubscribes on unmount mid-install (no leaked listener)", async () => { + const { result, unmount } = renderHook(() => useAgentInstallStream()) + await act(async () => { + await result.current.start("task-1") + }) + expect(unsubs).toHaveLength(1) + expect(unsubs[0]).not.toHaveBeenCalled() + + unmount() + + expect(unsubs[0]).toHaveBeenCalledTimes(1) + }) + + it("does not leak a subscription that resolves after unmount", async () => { + let resolveSub: ((unsub: () => void) => void) | null = null + mockSubscribe.mockImplementationOnce( + () => + new Promise<() => void>((resolve) => { + resolveSub = resolve + }) + ) + + const { result, unmount } = renderHook(() => useAgentInstallStream()) + let startPromise: Promise | undefined + act(() => { + startPromise = result.current.start("task-1") + }) + // Tearing down while subscribe() is still pending must not leak it. + unmount() + + const lateUnsub = vi.fn() + await act(async () => { + resolveSub?.(lateUnsub) + await startPromise + }) + + expect(lateUnsub).toHaveBeenCalledTimes(1) + }) +}) diff --git a/src/hooks/use-agent-install-stream.ts b/src/hooks/use-agent-install-stream.ts index 738937dcc..bbd243b6a 100644 --- a/src/hooks/use-agent-install-stream.ts +++ b/src/hooks/use-agent-install-stream.ts @@ -1,5 +1,6 @@ -import { useCallback, useRef, useState } from "react" +import { useCallback, useEffect, useRef, useState } from "react" import { subscribe } from "@/lib/platform" +import { appendInstallLogLine } from "@/lib/install-stream" import type { AgentInstallEvent, AgentInstallEventKind } from "@/lib/types" const AGENT_INSTALL_EVENT = "app://agent-install" @@ -19,8 +20,13 @@ export function useAgentInstallStream() { error: null, }) const unsubRef = useRef<(() => void) | null>(null) + // Flipped by reset()/unmount. Guards the gap between awaiting subscribe() and + // storing its unsubscribe fn: if the panel tore down meanwhile, we unsubscribe + // immediately instead of leaking the listener. + const cancelledRef = useRef(false) const start = useCallback(async (taskId: string) => { + cancelledRef.current = false setState({ status: "running", logs: [], error: null }) unsubRef.current?.() @@ -37,14 +43,14 @@ export function useAgentInstallStream() { case "log": setState((prev) => ({ ...prev, - logs: [...prev.logs, event.payload], + logs: appendInstallLogLine(prev.logs, event.payload), })) break case "completed": setState((prev) => ({ ...prev, status: "success", - logs: [...prev.logs, event.payload], + logs: appendInstallLogLine(prev.logs, event.payload), })) unsubRef.current?.() break @@ -53,7 +59,7 @@ export function useAgentInstallStream() { ...prev, status: "failed", error: event.payload, - logs: [...prev.logs, `ERROR: ${event.payload}`], + logs: appendInstallLogLine(prev.logs, `ERROR: ${event.payload}`), })) unsubRef.current?.() break @@ -61,14 +67,30 @@ export function useAgentInstallStream() { } ) + if (cancelledRef.current) { + // reset()/unmount ran while subscribe() was resolving — don't leak. + unsub() + return + } unsubRef.current = unsub }, []) const reset = useCallback(() => { + cancelledRef.current = true unsubRef.current?.() unsubRef.current = null setState({ status: "idle", logs: [], error: null }) }, []) + // Unsubscribe on unmount: a panel closed mid-install must not leak the + // global event subscription (or setState after unmount). + useEffect(() => { + return () => { + cancelledRef.current = true + unsubRef.current?.() + unsubRef.current = null + } + }, []) + return { ...state, start, reset } } diff --git a/src/hooks/use-officecli-install-stream.ts b/src/hooks/use-officecli-install-stream.ts index 173508cb2..b08df130a 100644 --- a/src/hooks/use-officecli-install-stream.ts +++ b/src/hooks/use-officecli-install-stream.ts @@ -1,5 +1,6 @@ -import { useCallback, useRef, useState } from "react" +import { useCallback, useEffect, useRef, useState } from "react" import { subscribe } from "@/lib/platform" +import { appendInstallLogLine } from "@/lib/install-stream" import type { OfficecliInstallEvent, OfficecliInstallEventKind, @@ -45,14 +46,14 @@ export function useOfficecliInstallStream() { case "log": setState((prev) => ({ ...prev, - logs: [...prev.logs, event.payload], + logs: appendInstallLogLine(prev.logs, event.payload), })) break case "completed": setState((prev) => ({ ...prev, status: "success", - logs: [...prev.logs, event.payload], + logs: appendInstallLogLine(prev.logs, event.payload), })) unsubRef.current?.() break @@ -61,7 +62,7 @@ export function useOfficecliInstallStream() { ...prev, status: "failed", error: event.payload, - logs: [...prev.logs, `ERROR: ${event.payload}`], + logs: appendInstallLogLine(prev.logs, `ERROR: ${event.payload}`), })) unsubRef.current?.() break @@ -84,5 +85,15 @@ export function useOfficecliInstallStream() { setState({ status: "idle", logs: [], error: null }) }, []) + // Unsubscribe on unmount: a panel closed mid-install must not leak the + // global event subscription (or setState after unmount). + useEffect(() => { + return () => { + cancelledRef.current = true + unsubRef.current?.() + unsubRef.current = null + } + }, []) + return { ...state, start, reset } } diff --git a/src/hooks/use-plugin-install-stream.ts b/src/hooks/use-plugin-install-stream.ts index a81021607..77e91ea55 100644 --- a/src/hooks/use-plugin-install-stream.ts +++ b/src/hooks/use-plugin-install-stream.ts @@ -1,5 +1,6 @@ -import { useCallback, useRef, useState } from "react" +import { useCallback, useEffect, useRef, useState } from "react" import { subscribe } from "@/lib/platform" +import { appendInstallLogLine } from "@/lib/install-stream" import type { PluginInstallEvent, PluginInstallEventKind } from "@/lib/types" const PLUGIN_INSTALL_EVENT = "app://opencode-plugin-install" @@ -19,8 +20,13 @@ export function usePluginInstallStream() { error: null, }) const unsubRef = useRef<(() => void) | null>(null) + // Flipped by reset()/unmount. Guards the gap between awaiting subscribe() and + // storing its unsubscribe fn: if the panel tore down meanwhile, we unsubscribe + // immediately instead of leaking the listener. + const cancelledRef = useRef(false) const start = useCallback(async (taskId: string) => { + cancelledRef.current = false setState({ status: "running", logs: [], error: null }) unsubRef.current?.() @@ -37,14 +43,14 @@ export function usePluginInstallStream() { case "log": setState((prev) => ({ ...prev, - logs: [...prev.logs, event.payload], + logs: appendInstallLogLine(prev.logs, event.payload), })) break case "completed": setState((prev) => ({ ...prev, status: "success", - logs: [...prev.logs, event.payload], + logs: appendInstallLogLine(prev.logs, event.payload), })) unsubRef.current?.() break @@ -53,7 +59,7 @@ export function usePluginInstallStream() { ...prev, status: "failed", error: event.payload, - logs: [...prev.logs, `ERROR: ${event.payload}`], + logs: appendInstallLogLine(prev.logs, `ERROR: ${event.payload}`), })) unsubRef.current?.() break @@ -61,14 +67,30 @@ export function usePluginInstallStream() { } ) + if (cancelledRef.current) { + // reset()/unmount ran while subscribe() was resolving — don't leak. + unsub() + return + } unsubRef.current = unsub }, []) const reset = useCallback(() => { + cancelledRef.current = true unsubRef.current?.() unsubRef.current = null setState({ status: "idle", logs: [], error: null }) }, []) + // Unsubscribe on unmount: a panel closed mid-install must not leak the + // global event subscription (or setState after unmount). + useEffect(() => { + return () => { + cancelledRef.current = true + unsubRef.current?.() + unsubRef.current = null + } + }, []) + return { ...state, start, reset } } diff --git a/src/lib/agent-plan.test.ts b/src/lib/agent-plan.test.ts index 12336de78..550bb1de2 100644 --- a/src/lib/agent-plan.test.ts +++ b/src/lib/agent-plan.test.ts @@ -134,4 +134,40 @@ describe("extractLatestPlanEntriesFromMessages", () => { { content: "Step two", status: "in_progress", priority: "medium" }, ]) }) + + const reasoningMsg = (id: string, text: string): AdaptedMessage => ({ + id, + role: "assistant", + timestamp: "2026-06-02T00:00:00.000Z", + content: [{ type: "reasoning", content: text, isStreaming: false }], + }) + + it("returns a stable shared empty reference for the no-plan case", () => { + const messages: AdaptedMessage[] = [ + reasoningMsg("a1", "just thinking\nnothing actionable"), + reasoningMsg("a2", "more reasoning, still no plan"), + ] + const first = extractLatestPlanEntriesFromMessages(messages) + const second = extractLatestPlanEntriesFromMessages(messages) + expect(first).toEqual([]) + // Same shared reference across calls — keeps the overlay memo stable + // across streaming batches instead of re-rendering on a fresh []. + expect(first).toBe(second) + }) + + it("extracts plan entries from reasoning text and memoizes per message", () => { + const messages: AdaptedMessage[] = [ + reasoningMsg( + "a1", + "- [pending] Write tests\n- [completed] Read code (high)" + ), + ] + const expected = [ + { content: "Write tests", status: "pending", priority: "medium" }, + { content: "Read code", status: "completed", priority: "high" }, + ] + expect(extractLatestPlanEntriesFromMessages(messages)).toEqual(expected) + // A second pass over the same message objects is served from the cache. + expect(extractLatestPlanEntriesFromMessages(messages)).toEqual(expected) + }) }) diff --git a/src/lib/agent-plan.ts b/src/lib/agent-plan.ts index b3bcfd9af..a5d1d2c16 100644 --- a/src/lib/agent-plan.ts +++ b/src/lib/agent-plan.ts @@ -10,6 +10,24 @@ import { } from "@/lib/plan-parse" import type { PlanEntryInfo } from "@/lib/types" +/** + * Shared empty result. Returning the same reference for the (common) no-plan + * case keeps the caller's `useMemo`/`React.memo` dependency stable across + * streaming batches instead of re-rendering on a fresh `[]` every time. + * Treated as read-only by all consumers. + */ +const EMPTY_PLAN_ENTRIES: PlanEntryInfo[] = [] + +/** + * Per-message memo for plan extraction. The turn adapter returns a STABLE + * `AdaptedMessage` reference for any unchanged (non-streaming) turn, so keying + * on the message object means a streaming batch only re-parses the one message + * still being streamed — without it, every 16ms batch re-ran the reasoning-text + * regex over every message in the conversation. Entries are auto-GC'd with the + * message objects (WeakMap), so memory tracks the live conversation. + */ +const messagePlanEntriesCache = new WeakMap() + function parseEntriesFromReasoningText(text: string): PlanEntryInfo[] { const lines = text .split("\n") @@ -89,25 +107,40 @@ function extractPlanEntriesFromPart(part: AdaptedContentPart): PlanEntryInfo[] { return [] } +/** Latest plan entries within a single message (empty if none), memoized by + * message identity. The expensive reasoning-text regex only re-runs when the + * message object itself is new (i.e. the turn still streaming). */ +function latestPlanEntriesInMessage(message: AdaptedMessage): PlanEntryInfo[] { + const cached = messagePlanEntriesCache.get(message) + if (cached !== undefined) return cached + + let result: PlanEntryInfo[] = EMPTY_PLAN_ENTRIES + for (let j = message.content.length - 1; j >= 0; j -= 1) { + const entries = extractPlanEntriesFromPart(message.content[j]) + if (entries.length > 0) { + result = entries + break + } + } + messagePlanEntriesCache.set(message, result) + return result +} + export function extractLatestPlanEntriesFromMessages( messages: AdaptedMessage[] ): PlanEntryInfo[] { - let planEntries: PlanEntryInfo[] = [] + let planEntries: PlanEntryInfo[] = EMPTY_PLAN_ENTRIES let planMessageIndex = -1 for (let i = messages.length - 1; i >= 0 && planMessageIndex === -1; i -= 1) { - const message = messages[i] - for (let j = message.content.length - 1; j >= 0; j -= 1) { - const entries = extractPlanEntriesFromPart(message.content[j]) - if (entries.length > 0) { - planEntries = entries - planMessageIndex = i - break - } + const entries = latestPlanEntriesInMessage(messages[i]) + if (entries.length > 0) { + planEntries = entries + planMessageIndex = i } } - if (planMessageIndex === -1) return [] + if (planMessageIndex === -1) return EMPTY_PLAN_ENTRIES // A fully completed plan that belongs to an earlier exchange is stale: once // the user has sent another message after it, a new turn has begun, so the @@ -121,7 +154,7 @@ export function extractLatestPlanEntriesFromMessages( const hasUserReplyAfterPlan = messages .slice(planMessageIndex + 1) .some((message) => message.role === "user") - if (hasUserReplyAfterPlan) return [] + if (hasUserReplyAfterPlan) return EMPTY_PLAN_ENTRIES } return planEntries diff --git a/src/lib/collab-tool.ts b/src/lib/collab-tool.ts index b531aba60..c69c3d734 100644 --- a/src/lib/collab-tool.ts +++ b/src/lib/collab-tool.ts @@ -201,6 +201,13 @@ function asText(v: unknown): string | null { export function isCodexCollabInput( rawInput: string | null | undefined ): boolean { + // Cheap gate: a collab input requires all three thread keys, so the absence of + // any one — checked here via `agentsStates`, the most collab-specific — means + // the payload can't be collab, and we skip the JSON.parse. This runs per + // tool_call on every 16ms streaming flush (via collapseLiveCollabBlocks), and + // the vast majority of tool calls (including every non-codex session) are not + // collab ops. + if (!rawInput || !rawInput.includes('"agentsStates"')) return false const parsed = tryParseObject(rawInput) if (!parsed) return false return ( diff --git a/src/lib/install-stream.test.ts b/src/lib/install-stream.test.ts new file mode 100644 index 000000000..4c64f1bc2 --- /dev/null +++ b/src/lib/install-stream.test.ts @@ -0,0 +1,25 @@ +import { describe, expect, it } from "vitest" +import { appendInstallLogLine, MAX_INSTALL_LOG_LINES } from "./install-stream" + +describe("appendInstallLogLine", () => { + it("appends while below the cap", () => { + expect(appendInstallLogLine(["a"], "b")).toEqual(["a", "b"]) + }) + + it("caps at MAX_INSTALL_LOG_LINES, keeping the most recent lines", () => { + let logs: string[] = [] + for (let i = 0; i < MAX_INSTALL_LOG_LINES + 250; i++) { + logs = appendInstallLogLine(logs, `line-${i}`) + } + expect(logs).toHaveLength(MAX_INSTALL_LOG_LINES) + // The first 250 lines were evicted; the tail survives. + expect(logs[0]).toBe("line-250") + expect(logs[logs.length - 1]).toBe(`line-${MAX_INSTALL_LOG_LINES + 249}`) + }) + + it("does not mutate the input array", () => { + const input = ["a"] + appendInstallLogLine(input, "b") + expect(input).toEqual(["a"]) + }) +}) diff --git a/src/lib/install-stream.ts b/src/lib/install-stream.ts new file mode 100644 index 000000000..954e552c2 --- /dev/null +++ b/src/lib/install-stream.ts @@ -0,0 +1,21 @@ +/** + * Shared helpers for the install-stream hooks (agent / opencode-plugin / + * officecli). Each hook appends one log line per backend event; a long + * install can emit thousands, and the consuming panels render every line as a + * DOM node. Without a cap the array (and the DOM list) grows unbounded and + * each append is an O(n) copy, so a long install is O(n²) in total. Keep only + * the tail — the recent lines are what a user actually scrolls to for + * progress and the error at the end. + */ +export const MAX_INSTALL_LOG_LINES = 1000 + +/** + * Append `line`, keeping at most the most recent `MAX_INSTALL_LOG_LINES`. + * Bounded to O(MAX) per append regardless of how many lines were streamed. + */ +export function appendInstallLogLine(logs: string[], line: string): string[] { + if (logs.length < MAX_INSTALL_LOG_LINES) { + return [...logs, line] + } + return [...logs.slice(logs.length - MAX_INSTALL_LOG_LINES + 1), line] +} diff --git a/src/lib/plan-parse.ts b/src/lib/plan-parse.ts index 7b5d8d91c..b63ea7ff1 100644 --- a/src/lib/plan-parse.ts +++ b/src/lib/plan-parse.ts @@ -205,6 +205,12 @@ export function kimiTodoWriteEntries( input: string | null | undefined ): PlanEntryInfo[] | null { if (!input) return null + // Cheap gate: a non-null result requires a top-level `todos` array, whose key + // always appears verbatim in the serialized input. This runs per tool_call on + // every 16ms streaming flush, and the overwhelming majority of tool calls + // (Read/Write/Bash/Edit — some with multi-KB `raw_input`) never carry one, so + // the substring scan lets us skip their JSON.parse entirely. + if (!input.includes('"todos"')) return null let parsed: unknown try { parsed = JSON.parse(input) diff --git a/src/lib/transport/remote-desktop-transport.ts b/src/lib/transport/remote-desktop-transport.ts index 3816c093c..be9fa4d8c 100644 --- a/src/lib/transport/remote-desktop-transport.ts +++ b/src/lib/transport/remote-desktop-transport.ts @@ -1,5 +1,6 @@ import { invoke } from "@tauri-apps/api/core" import { listen, type UnlistenFn } from "@tauri-apps/api/event" +import { randomUUID } from "@/lib/utils" import { WS_READY_CHANNEL } from "./constants" import type { AttachTransportHost } from "./web-event-stream" import { WebEventStream } from "./web-event-stream" @@ -103,7 +104,10 @@ export class RemoteDesktopTransport implements Transport { /// and `remote_ws_unsubscribe`. Because it is known before the invoke /// returns, destroy() can always issue a clean unsubscribe regardless of /// whether the subscribe invoke is still in-flight. - private readonly subscriptionId = crypto.randomUUID() + /// Uses the randomUUID() helper (crypto.getRandomValues fallback) rather than + /// a bare crypto.randomUUID() so construction can't throw on a webview that + /// isn't a secure context (e.g. Linux WebKitGTK). + private readonly subscriptionId = randomUUID() /// Null = not yet requested; true = subscribe invoke in-flight or done. private wsStarted = false /// Latched in `destroy()` so any in-flight `subscribe()` awaiters diff --git a/src/lib/utils.test.ts b/src/lib/utils.test.ts index 22d41a1b8..d4300c9d4 100644 --- a/src/lib/utils.test.ts +++ b/src/lib/utils.test.ts @@ -1,7 +1,7 @@ import type { MouseEvent } from "react" import { describe, expect, it, vi } from "vitest" -import { handleMiddleClickClose } from "./utils" +import { handleMiddleClickClose, randomUUID } from "./utils" function mouseEventWithButton(button: number) { const preventDefault = vi.fn() @@ -40,3 +40,27 @@ describe("handleMiddleClickClose", () => { expect(preventDefault).not.toHaveBeenCalled() }) }) + +const V4_UUID_RE = + /^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/ + +describe("randomUUID", () => { + it("returns a valid v4 UUID when crypto.randomUUID is available", () => { + expect(randomUUID()).toMatch(V4_UUID_RE) + }) + + it("falls back to crypto.getRandomValues in non-secure contexts", () => { + // Simulate the server-over-HTTP-on-LAN case: a non-secure context where + // crypto.randomUUID is undefined but crypto.getRandomValues still works. + const realCrypto = globalThis.crypto + vi.stubGlobal("crypto", { + getRandomValues: (arr: Uint8Array) => realCrypto.getRandomValues(arr), + }) + try { + const id = randomUUID() + expect(id).toMatch(V4_UUID_RE) + } finally { + vi.unstubAllGlobals() + } + }) +})