diff --git a/README.md b/README.md index 717363d3..153396d8 100644 --- a/README.md +++ b/README.md @@ -1,13 +1,13 @@ # cc-remote -**在手机和浏览器里，继续使用你机器上的 Claude Code 和 Codex。** +**在手机、浏览器和终端里，继续使用你机器上的 Claude Code 和 Codex。** -自托管 · 多会话 · 多设备 · 实时工具过程 · Code / Work · PWA +自托管 · 多会话 · 多设备 · 实时工具过程 · Code / Work · Web / PWA / TUI **产品版本：v3.0.0** · Wire protocol v71 [English](README_en.md) · [功能对照](#引擎与功能) · [快速开始](#快速开始) · -[安装与升级](#安装与升级) · [文档](#文档) · [更新记录](CHANGELOG_zh.md) +[终端工作台](#terminal-workspace) · [安装与升级](#安装与升级) · [文档](#文档) · [更新记录](CHANGELOG_zh.md) cc-remote 把本机 agent 的会话、工具过程、文件和运行控制带到远端。你可以在电脑上 开始任务，用手机查看进度、回答询问、补充指令，再回到原来的会话继续工作。 @@ -32,6 +32,9 @@ cc-remote 把本机 agent 的会话、工具过程、文件和运行控制带到 可配置独立账号，分别使用原生登录、会话和扩展目录。 - **适合手机使用**：紧凑的引擎菜单、明暗主题、图片缩放、PWA 和可选后台通知。 通知默认只显示通用状态，显示会话名称需要主动开启。 +- **也能在终端继续工作**：全屏 TUI 支持 Claude/Codex 的 Code / Work，共享网页 + 会话、运行状态和队列；提供 Vim 风格操作、会话树、标签与工具详情。 + [安装与启动](#terminal-workspace)。 ### Code 与 Work @@ -51,6 +54,7 @@ cc-remote 把本机 agent 的会话、工具过程、文件和运行控制带到 |---|---|---| | 接入方式 | 日常 Claude CLI + Agent SDK | 官方 app-server，共享 daemon | | Code / Work | 两者支持 | 两者支持 | +| 全屏终端 TUI | 支持 | 支持 | | 模型与思考 | 原生模型与支持的档位 | 原生模型、思考强度、服务档位 | | Plan | 原生 Plan 模式 | 原生 Plan 协作模式 | | Goal | 完成条件、检查轮次、最近检查结果、Token 用量和耗时 | 目标、可选 Token 预算、暂停／继续、完成和清除 | @@ -127,6 +131,7 @@ Relay 不落盘原文件或预览结果。Viewer 默认复用主站地址，不 ```mermaid flowchart LR browser["手机 / 浏览器"] <-->|HTTPS / WSS| relay["Relay + Web 静态文件"] + terminal["终端 TUI"] <-->|WSS| relay subgraph device["你的机器"] wrapper["Wrapper"] <--> engine["Claude SDK / Codex app-server"] end @@ -266,18 +271,45 @@ Code 默认权限较宽；Work 的私有目录策略不能替代独立系统用 | Codex App 接入：[macOS](docs/codex-desktop-launcher.md)／[Linux](docs/codex-desktop-linux.md) | 可选桌面 App、日常 CLI 与 Wrapper 共用 daemon | | [Codex App 工具](docs/codex-app-tools.md) | 可选 App-control MCP | | [定时消息 UI](docs/timed-messages.md) | 定时发送入口、消息标签、光圈与下次时间 | +| [终端工作台](docs/tui_zh.md) | TUI 安装、连接、Vim 操作、会话树、文件预览与终端限制 | | [更新记录](CHANGELOG_zh.md) | 版本变化与迁移记录 | + + ## 终端工作台（预览） -不习惯网页时，可以使用内置的 Python/Textual 全屏 TUI。它与网页共享 -relay/wrapper 会话，提供会话标签、Space e 目录树及搜索、Vim 风格阅读与输入 -模式、消息跳转，以及不改变阅读位置的复制和引用。 -支持思考/工具/进程详情、运行计时、Goal/Plan、 -用量、队列编辑及共享会话控制；支持 Markdown 预览和兼容终端的图片显示， -其他图形内容保留明确的网页入口。 -安装、快捷键及终端边界见 -[终端工作台中文指南](docs/tui_zh.md)。原逐行客户端可用 `--line-mode` 启动。 +内置的 Python/Textual 全屏 TUI 支持 **Claude/Codex 的 Code / Work**，连接与网页 +相同的 Relay 和 Wrapper。会话、运行状态和队列来自同一控制链路；关闭 TUI +不会停止已运行的任务或已接收的排队消息。原生客户端的会话写入权限仍然生效。 + +完成上面的源码依赖安装后，再安装可选终端依赖： + +```bash +.venv/bin/python -m pip install -r requirements-tui.txt +# 离线体验，不连接服务器，也不调用模型。 +./scripts/cc-remote-tui --demo +# 连接自己的 Relay；按提示输入登录口令，替换示例域名。 +./scripts/cc-remote-tui --engine codex --url wss://cc.example.com/ws +``` + +`--engine claude` 切换引擎，`--space work` 进入 Work；可附加会话 ID， +或用 `--machine ` 选择已授权设备。TUI 与 Relay、Wrapper 必须使用相同的 +源码／协议版本。登录配置、本机自动认证及固定启动命令见 +[终端工作台中文指南](docs/tui_zh.md)。 + +| 操作 | 默认按键 | +|---|---| +| 会话目录树与搜索 | Normal 模式下 `Space e`，树内 `/` 搜索 | +| 切换已打开会话 | Normal 模式下 `H` / `L`；`Space ,` 搜索标签 | +| 切换阅读区／输入框 | `Ctrl+k` / `Ctrl+j` | +| 输入与发送 | `i` 进入 Insert，`Esc` 回 Normal，Normal 下 `Enter` 发送 | +| 排队／停止当前轮 | `Ctrl+e` / `Ctrl+x` | +| 查看当前快捷键 | `Space h` | + +支持 Vim 选择、复制和引用，按需展开思考／工具／进程详情，以及运行计时、 +Goal / Plan、用量与队列编辑。运行状态文字从左向右扫光，完成或中断后停止； +`TEXTUAL_ANIMATIONS=none` 可关闭动画。Markdown 可在终端预览，图片取决于终端 +图形协议；其他内容提供网页入口。原逐行客户端仍可通过 `--line-mode` 启动。 ## 开发 diff --git a/README_en.md b/README_en.md index 6be3435a..fd6c11a3 100644 --- a/README_en.md +++ b/README_en.md @@ -1,13 +1,13 @@ # cc-remote -**Use Claude Code and Codex on your machine from your phone or browser.** +**Use Claude Code and Codex on your machine from your phone, browser or terminal.** -Self-hosted · Multiple sessions and devices · Live tool activity · Code / Work · PWA +Self-hosted · Multiple sessions and devices · Live tool activity · Code / Work · Web / PWA / TUI **Product version: v3.0.0** · Wire protocol v71 [中文](README.md) · [Engine comparison](#engines-and-features) · [Quick start](#quick-start) · -[Install and upgrade](#install-and-upgrade) · [Documentation](#documentation) · [Changelog](CHANGELOG.md) +[Terminal workspace](#terminal-workspace) · [Install and upgrade](#install-and-upgrade) · [Documentation](#documentation) · [Changelog](CHANGELOG.md) cc-remote brings your local agent's sessions, tool activity, files and controls to remote clients. Start a task on your computer, check progress from your phone, @@ -41,6 +41,9 @@ features or wire protocol across different commits. - **Work from a phone.** Compact engine menus, light/dark themes, image zoom, PWA installation and optional background notifications. Notifications default to generic status; showing a session name requires an explicit opt-in. +- **Continue in a terminal.** The full-screen TUI supports Claude/Codex Code and + Work, sharing Web sessions, running state and queues. Use Vim-style controls, + a searchable session tree, tabs and tool details. [Install and run](#terminal-workspace). ### Code and Work @@ -62,6 +65,7 @@ and extensions still depend on the native installation on the selected device. |---|---|---| | Connection | Daily Claude CLI + Agent SDK | Official app-server, shared daemon | | Code / Work | Both | Both | +| Full-screen terminal TUI | Supported | Supported | | Models and reasoning | Native models and supported levels | Native models, reasoning effort, service tier | | Plan | Native Plan mode | Native Plan collaboration mode | | Goal | Completion condition, check count, latest feedback, token usage and elapsed time | Objective, optional token budget, pause/resume, complete and clear | @@ -147,6 +151,7 @@ preview domain; it does not proxy arbitrary private-network services. ```mermaid flowchart LR browser["Phone / browser"] <-->|HTTPS / WSS| relay["Relay + static Web client"] + terminal["Terminal TUI"] <-->|WSS| relay subgraph device["Your machine"] wrapper["Wrapper"] <--> engine["Claude SDK / Codex app-server"] end @@ -305,19 +310,49 @@ policy is not a replacement for separate OS users, containers or virtual machine | Codex App: [macOS](docs/codex-desktop-launcher.md) / [Linux](docs/codex-desktop-linux.md) | Optional App, daily CLI and Wrapper on one daemon | | [Codex App tools](docs/codex-app-tools.md) | Optional App-control MCP | | [Timed messages](docs/timed-messages.md) | Scheduled queue receipts, message tags and countdown UI | +| [Terminal workspace](docs/tui.md) | TUI setup, connections, Vim controls, session tree, previews and terminal limits | | [Changelog](CHANGELOG.md) | Version changes and migrations | + + ## Terminal workspace (preview) -The built-in Python/Textual TUI shares relay/wrapper sessions with Web. It -provides session tabs, a Space e directory tree with search, Vim-style reading -and input modes, message jumps, and copy/quote actions that retain the reading -position. Thinking/tool/process -details, elapsed time, Goal/Plan, usage, queue editing and shared session -controls are available. Markdown and compatible-terminal image previews are -built in; other graphical content has an explicit Web handoff. -See the [terminal workspace guide](docs/tui.md) for installation, keys and -terminal boundaries. Use `--line-mode` for the original line-oriented client. +The built-in Python/Textual full-screen TUI supports **Claude/Codex Code and +Work** through the same Relay and Wrapper as Web. Sessions, running state and +queues share the control link. Closing the TUI leaves running tasks and accepted +queued messages intact; native clients' session ownership rules still apply. + +After installing the source dependencies above, add the optional terminal packages: + +```bash +.venv/bin/python -m pip install -r requirements-tui.txt +# Offline preview; no server connection or model calls. +./scripts/cc-remote-tui --demo +# Connect to your Relay; replace the example domain and enter the password when prompted. +./scripts/cc-remote-tui --engine codex --url wss://cc.example.com/ws +``` + +Use `--engine claude` for Claude or `--space work` for Work. Append a session ID +to open it directly, or use `--machine ` to select an authorized device. +The TUI, Relay and Wrapper must use the same source/protocol version. See the +[terminal workspace guide](docs/tui.md) for authentication, local automatic +login and a launcher that works from any directory. + +| Action | Default keys | +|---|---| +| Session directory tree and search | `Space e` in Normal mode, then `/` to search | +| Switch open sessions | `H` / `L` in Normal mode; `Space ,` searches tabs | +| Focus transcript / draft | `Ctrl+k` / `Ctrl+j` | +| Type and send | `i` enters Insert; `Esc` returns to Normal; Normal `Enter` sends | +| Queue / stop the current turn | `Ctrl+e` / `Ctrl+x` | +| Current shortcut help | `Space h` | + +Vim selection, copy and quote, on-demand thinking/tool/process details, elapsed +time, Goal / Plan, usage and queue editing are available. Active status text +sweeps from left to right and stops on completion or interruption; set +`TEXTUAL_ANIMATIONS=none` to disable animation. Markdown previews work in the +terminal; images depend on its graphics protocol, and other content offers a +Web handoff. Use `--line-mode` for the original line-oriented client. ## Development diff --git a/cc_remote/claude_service/client.py b/cc_remote/claude_service/client.py index 8ecea662..fcbfc63a 100644 --- a/cc_remote/claude_service/client.py +++ b/cc_remote/claude_service/client.py @@ -214,12 +214,15 @@ async def receive_messages(self): if result["failure"] and not result["events"]: raise RuntimeError("Claude SDK stream ended: " + result["failure"]) - async def steer(self, prompt, *, native_id, metadata, turn_id): + async def steer(self, prompt, *, native_id, metadata, turn_id, background_id=None): if not self.description.get("native_steering"): raise ClaudeSteerRejected("Claude service requires a steering upgrade") + if background_id is not None and not self.description.get("background_steering"): + raise ClaudeSteerRejected("Claude service requires a background steering upgrade") accepted = await self.call("steer", { "prompt": prompt, "native_id": native_id, "metadata": metadata, "turn_id": turn_id, + **({"background_id": background_id} if background_id is not None else {}), }, request_id="steer-" + native_id) if not accepted: raise ClaudeSteerRejected("Claude response has already ended") diff --git a/cc_remote/claude_service/server.py b/cc_remote/claude_service/server.py index e878ca80..ba6c9b79 100644 --- a/cc_remote/claude_service/server.py +++ b/cc_remote/claude_service/server.py @@ -19,7 +19,7 @@ from pathlib import Path from uuid import uuid4 -from cc_remote.claude_steering import PendingSteers, steer_message +from cc_remote.claude_steering import PendingSteers, steer_message, _origin_key from .wire import ( ControllerLeaseConflict, decode_sdk, encode_sdk, private_directory, @@ -201,6 +201,20 @@ async def read_messages(self) -> None: value = self.steers.annotate(value) if "__cc_steer" in value: self.origin_id = value["__cc_steer"]["id"] + if self.turn is not None: + if self.turn.get("awaiting_steer"): + identity = self.turn.get("background_id") + self.steers.handoff_background(identity) + # The exact human echo transfers this continuation + # to the managed turn. Some native versions emit no + # separate autonomous Result at this safe boundary. + self.background_turns = [t for t in self.background_turns + if t.get("identity") != identity] + self.turn.pop("awaiting_steer", None) + if (value.get("__cc_steer_cancelled") and not self.steers.pending + and self.turn and self.turn.get("awaiting_steer")): + self.turn = None + self._cleanup_steer_attachments() # The journal is on disk, like Claude's own transcript. Do not # stop the sole native reader at a per-turn byte cap: an offline # long turn could then never deliver the Result that frees it. @@ -208,10 +222,12 @@ async def read_messages(self) -> None: origin = value.get("origin") kind = origin.get("kind") if isinstance(origin, dict) else None if value.get("type") == "user" and kind not in (None, "human"): - self.background_turns.append({"start_seq": seq - 1, "terminal_seq": None}) + self.background_turns.append({"start_seq": seq - 1, "terminal_seq": None, + "origin": _origin_key(origin), + "identity": self.steers.background_id}) if value.get("type") == "result" and kind not in (None, "human"): for turn in reversed(self.background_turns): - if turn["terminal_seq"] is None: + if turn["terminal_seq"] is None and turn.get("origin") == _origin_key(origin): turn["terminal_seq"] = seq break if value.get("type") == "system": @@ -225,6 +241,7 @@ async def read_messages(self) -> None: if isinstance(sid, str) and sid: self.metadata["session_id"] = sid if (self.turn is not None and _human_result(value) + and not self.turn.get("awaiting_steer") and not value.get("__cc_steer_intermediate")): self.terminal_seq = seq await self.notify() @@ -248,6 +265,8 @@ def description(self) -> dict: "head": self.journal.seq, "controls": self.controls, "pid": os.getpid(), "task_seeds": list(self.task_seeds.values()), "native_steering": True, + "background_steering": True, + "pending_steers": {uid: {"id": data["id"]} for uid, data in self.steers.pending.items()}, } async def events(self, after: int) -> dict: @@ -278,7 +297,9 @@ async def mutate(self, request_id: str, method: str, params: dict): # A controller replacement may stage identical attachment bytes at # another private path. Compare the original browser payload digest, # not those incidental paths; the first mutation keeps its payload. - identity = [method, params["turn_id"], params["native_id"], + # The first accepted operation owns its target. Recovery can see + # that background continuation already adopted as a human turn. + identity = [method, params["native_id"], params["metadata"]["fingerprint"]] fingerprint = hashlib.sha256(json.dumps(identity, sort_keys=True).encode()).hexdigest() previous = self.mutation_fingerprints.get(request_id) @@ -327,10 +348,20 @@ async def _mutate(self, method: str, params: dict): return result async with self.lock: if method == "steer": - if (self.turn is None or self.terminal_seq is not None - or self.failure or params["turn_id"] != self.turn["id"]): + background = self.turn is None and params.get("background_id") is not None + if self.failure or (background and params["background_id"] != self.steers.background_id): + return False + if not background and (self.turn is None or self.terminal_seq is not None + or params["turn_id"] != self.turn["id"]): return False self.steers.add(params["native_id"], params["metadata"]) + if background: + self.turn = {**params["metadata"], "background_steer": True, "awaiting_steer": True, + "background_id": params["background_id"], + "background_origin": self.steers.background_origin_data, + "previous_origin_id": self.origin_id, + "start_seq": self.journal.seq, "started_at": time.time()} + self.terminal_seq = None directory = params["metadata"].get("attachment_dir") if directory: self.steer_attachment_dirs.add(directory) diff --git a/cc_remote/claude_steering.py b/cc_remote/claude_steering.py index 24422fe2..e0ee6b3a 100644 --- a/cc_remote/claude_steering.py +++ b/cc_remote/claude_steering.py @@ -2,6 +2,9 @@ from __future__ import annotations +import hashlib +import json + class ClaudeSteerRejected(RuntimeError): """The instruction was definitely not written to the native input stream.""" @@ -14,6 +17,15 @@ def steer_message(prompt, native_id: str) -> dict: "parent_tool_use_id": None, "uuid": native_id, "priority": "next"} +def _origin_key(origin: dict) -> str: + kind = origin.get("kind") + for field in ("taskId", "task_id", "senderTaskId", "fromSession", "verifiedPeerPid", "from", "server", "name"): + value = origin.get(field) + if isinstance(value, (str, int)) and not isinstance(value, bool) and str(value).strip(): + return json.dumps([kind, "task" if field in {"taskId", "task_id"} else field, str(value).strip()]) + return json.dumps(origin, sort_keys=True) + + class PendingSteers: """Fence accepted inputs against their exact replayed human UUIDs. @@ -25,6 +37,15 @@ class PendingSteers: def __init__(self): self.pending: dict[str, dict] = {} self.capabilities: set[str] = set() + self.background_id: str | None = None + self.background_origin: str | None = None + self.background_origin_data: dict | None = None + + def handoff_background(self, identity: str | None) -> None: + if identity and identity == self.background_id: + self.background_id = None + self.background_origin = None + self.background_origin_data = None def add(self, native_id: str, metadata: dict) -> None: if len(self.pending) >= 32 or native_id in self.pending: @@ -32,6 +53,26 @@ def add(self, native_id: str, metadata: dict) -> None: self.pending[native_id] = metadata def annotate(self, value: dict) -> dict: + origin = value.get("origin") + kind = origin.get("kind") if isinstance(origin, dict) else None + if not value.get("parent_tool_use_id") and kind not in (None, "human"): + message = value.get("message") + content = message.get("content") if isinstance(message, dict) else None + tool_result = isinstance(content, list) and any( + isinstance(part, dict) and part.get("type") == "tool_result" for part in content) + if value.get("type") == "user" and not tool_result: + self.background_origin = _origin_key(origin) + self.background_origin_data = origin + self.background_id = "background-" + hashlib.sha256(json.dumps( + [value.get("uuid"), self.background_origin], sort_keys=True).encode()).hexdigest()[:32] + elif value.get("type") == "result" and _origin_key(origin) == self.background_origin: + self.background_id = None + self.background_origin = None + self.background_origin_data = None + cancelled = value.get("__cc_steer_cancelled") + if cancelled: + self.pending = {uid: data for uid, data in self.pending.items() + if data.get("id") != cancelled.get("id")} if value.get("type") == "system" and value.get("subtype") == "init": self.capabilities = set(value.get("capabilities") or []) if value.get("type") == "command_lifecycle" and value.get("state") == "cancelled": @@ -39,14 +80,12 @@ def annotate(self, value: dict) -> dict: if metadata is not None: return {**value, "type": "system", "subtype": "cc_remote_steer_cancelled", "__cc_steer_cancelled": metadata} - origin = value.get("origin") - kind = origin.get("kind") if isinstance(origin, dict) else None if kind not in (None, "human") or value.get("parent_tool_use_id"): return value if value.get("type") == "user": metadata = self.pending.pop(value.get("uuid"), None) if metadata is not None: - return {**value, "__cc_steer": metadata} + return {**value, "__cc_steer": value.get("__cc_steer", metadata)} elif value.get("type") == "result" and self.pending: return {**value, "__cc_steer_intermediate": True} return value diff --git a/cc_remote/protocol.py b/cc_remote/protocol.py index 4154eb4e..ffb47b54 100644 --- a/cc_remote/protocol.py +++ b/cc_remote/protocol.py @@ -2630,7 +2630,7 @@ class TurnDetail(_Base): class GetAgentDetail(_Command): - """client -> wrapper: fetch one Claude subagent's public process projection.""" + """client -> wrapper: fetch one subagent's public process projection.""" type: Literal["get_agent_detail"] = "get_agent_detail" session_id: WireId run_id: WireId @@ -2643,12 +2643,13 @@ class GetAgentDetail(_Command): class AgentDetail(_Base): - """wrapper -> browser: one read-only page from a Claude subagent run. + """wrapper -> browser: one read-only page from a subagent run. - ``run_id`` is a stable public hash of the spawning Agent tool call. Raw + For Claude, ``run_id`` is a stable public hash of the spawning Agent tool call. Raw Claude agent ids, delegated prompts and output-file paths never cross this boundary. Live batches are unbuffered hints; a normal response is always a - source-backed or resident authoritative snapshot. + source-backed or resident authoritative snapshot. Codex cards resolve a + native child only after verifying account-local ancestry to this session. """ type: Literal["agent_detail"] = "agent_detail" session_id: WireId diff --git a/cc_remote/tui_app.py b/cc_remote/tui_app.py index 0d41c81c..b740e564 100644 --- a/cc_remote/tui_app.py +++ b/cc_remote/tui_app.py @@ -65,6 +65,7 @@ from cc_remote.tui_attachments import read_attachment from cc_remote.tui_clipboard import read_clipboard_image from cc_remote.tui_chrome import ( + activity_sweep, settings_text, status_text as styled_status, ) @@ -1126,7 +1127,18 @@ def offset(text: str, point: tuple[int, int]) -> int: class Transcript(VimArea): """Read-only document with a logical cursor independent of the composer.""" + COMPONENT_CLASSES = VimArea.COMPONENT_CLASSES | { + "transcript--activity", "transcript--activity-highlight", + "transcript--activity-on-user", + } + DEFAULT_CSS = """ + Transcript > .transcript--activity { color: $text-muted; } + Transcript > .transcript--activity-highlight { color: #f5f5f5; } + Transcript > .transcript--activity-on-user { color: #a4afba; } + """ line_styles: dict[int, str] = {} + activity_ranges: dict[int, tuple[int, int, bool]] = {} + activity_frame: int | None = None resize_bookmark = None follow_scroll_sid = None follow_revision = 0 @@ -1261,12 +1273,44 @@ def get_line(self, line_index: int) -> Text: and rich_lines[line_index].plain == line.plain): line = rich_lines[line_index].copy() line.stylize(self.line_styles.get(line_index, "")) + if self.activity_frame is not None and line_index in self.activity_ranges: + start, end, on_user = self.activity_ranges[line_index] + activity_sweep( + line, start, end, self.activity_frame / 23, + self.get_component_rich_style( + "transcript--activity-on-user" if on_user + else "transcript--activity" + ), + self.get_component_rich_style("transcript--activity-highlight"), + ) return line - def style_messages(self, text: str, starts: list) -> None: + def refresh_activity(self) -> None: + if not self.activity_ranges: + self.activity_frame = None + return + enabled = self.app.animation_level != "none" and not self.app.no_color + frame = int(time.monotonic() * 10) % 24 if enabled else None + if frame == self.activity_frame: + return + self.activity_frame = frame + first = int(self.scroll_y) + rows = self.wrapped_document._offset_to_line_info[ + first:first + self.scrollable_content_region.height + ] + visible = [first + y for y, (row, _) in enumerate(rows) + if row in self.activity_ranges] + if visible: + self._line_cache.clear() + for y in visible: + self.refresh_lines(y) + + def style_messages(self, text: str, starts: list, view) -> None: # Inline styles can change while the rendered characters stay equal. self.notify_style_update() self.line_styles = {} + self.activity_ranges = {} + seen_turns = set() row = 0 for index, (start, block) in enumerate(starts): end = starts[index + 1][0] if index + 1 < len(starts) else len(text) @@ -1298,6 +1342,20 @@ def style_messages(self, text: str, starts: list) -> None: if muted else "bold green" ) + turn = view.presentation.turns.get(block.turn) + if turn and turn.status == "running": + header = text[start:end].split("\n", 1)[0] + if block.role == "detail" or ( + block.role == "tool_group" + and block.data.get("status") == "running" + ): + self.activity_ranges[row] = (0, len(header), False) + elif block.turn not in seen_turns and " [" in header: + # The first block owns the turn's clock/status suffix. + self.activity_ranges[row] = ( + header.index(" [") + 2, len(header), block.role == "user", + ) + seen_turns.add(block.turn) if block.role == "detail" and block.expanded: sections = block.data.get("sections", []) projection = self.parent.projection @@ -1759,6 +1817,7 @@ def paint(self) -> None: else: editor.load_text("") reader.load_text("") + reader.activity_ranges = {} reader.parent.reset(None) reader.focus() elif sid: @@ -1858,7 +1917,7 @@ def paint(self) -> None: text, starts = reader.parent.project( text, starts, projection_identity ) - reader.style_messages(text, starts) + reader.style_messages(text, starts, view) reader.follow_layout_pending = True with reader.presentation_update(): if text != reader.text: @@ -1904,6 +1963,7 @@ def paint(self) -> None: self.jump(role, direction=direction) if view: self.refresh_turn_clocks(view, reader) + reader.refresh_activity() if self.app_focus and not view.loading and self.starts: view.read_tab() reader.parent.sync() @@ -2006,6 +2066,9 @@ def refresh_turn_clocks(self, view, reader) -> None: block, show_turn=True, now=now, detail_key=self.client.keys.layer_label("reader", "details"), ) + if row in reader.activity_ranges: + begin, _, on_user = reader.activity_ranges[row] + reader.activity_ranges[row] = (begin, len(new), on_user) # Clock updates only replace a single header line. Never touch # multi-line activity text or replace content from another block. if old != new and "\n" not in new: diff --git a/cc_remote/tui_chrome.py b/cc_remote/tui_chrome.py index 04cc2f46..e4e90751 100644 --- a/cc_remote/tui_chrome.py +++ b/cc_remote/tui_chrome.py @@ -1,5 +1,8 @@ """Quiet status-line hierarchy: neutral text, one accent, semantic warnings.""" +from rich.cells import get_character_cell_size +from rich.color import Color +from rich.style import Style from rich.text import Text from cc_remote.tui import _safe_remote_text @@ -9,6 +12,28 @@ MUTED = "dim" +def activity_sweep(text: Text, start: int, end: int, phase: float, + base: Style, highlight: Style) -> None: + """Sweep foreground color in terminal cells, preserving text and backgrounds.""" + widths = [get_character_cell_size(char) for char in text.plain[start:end]] + width = sum(widths) + if not width or not base.color or not highlight.color: + return + text.stylize(Style(color=base.color, dim=False), start, end) + radius = max(3, min(12, width * 0.18)) + center = -radius + (width + 2 * radius) * phase + low, high = base.color.get_truecolor(), highlight.color.get_truecolor() + cell = 0 + for index, cells in enumerate(widths, start): + strength = max(0, 1 - abs(cell + cells / 2 - center) / radius) + if strength: + color = Color.from_rgb(*( + round(a + (b - a) * strength) for a, b in zip(low, high) + )) + text.stylize(Style(color=color, dim=False), index, index + 1) + cell += cells + + def setting_style(kind, value): if kind == "model": return f"bold {ACCENT}" diff --git a/cc_remote/wrapper/claude_compaction.py b/cc_remote/wrapper/claude_compaction.py index ee9ce39f..a598ac67 100644 --- a/cc_remote/wrapper/claude_compaction.py +++ b/cc_remote/wrapper/claude_compaction.py @@ -2,8 +2,49 @@ from __future__ import annotations from typing import Any +import hashlib +import re -from cc_remote.protocol import MAX_SAFE_WIRE_INTEGER +from cc_remote.protocol import ( + MAX_SAFE_WIRE_INTEGER, AssistantMsgEnd, AssistantMsgStart, Delta, +) + + +def manual_compact_prompt(content: object) -> str | None: + """Recognize only the native /compact command, never other slash commands.""" + if isinstance(content, list): + if any((block.get("type") != "text" if isinstance(block, dict) + else not isinstance(getattr(block, "text", None), str)) for block in content): + return None + texts = ( + block.get("text", "") if isinstance(block, dict) + else getattr(block, "text", "") + for block in content + ) + content = "".join(text for text in texts if isinstance(text, str)) + if not isinstance(content, str): + return None + text = content.strip() + if text == "/compact": + return text + if not text.startswith(("", "")): + return None + name = re.search(r"(.*?)", text, re.S) + if name is None or name[1].strip() != "/compact": + return None + args = re.search(r"(.*?)", text, re.S) + return "/compact" + (" " + args[1].strip() if args and args[1].strip() else "") + + +def compact_completion_events(boundary_id: str, turn_id: str | None) -> list: + """A stable UI receipt for a proven manual compact, shared by live/history.""" + message_id = "compact-result-" + hashlib.sha256(boundary_id.encode()).hexdigest()[:24] + return [ + AssistantMsgStart(message_id=message_id, turn_id=turn_id, channel="final"), + Delta(message_id=message_id, turn_id=turn_id, channel="final", + text="上下文已压缩，可以继续当前会话。"), + AssistantMsgEnd(message_id=message_id, turn_id=turn_id, channel="final"), + ] def compact_metadata(row: object) -> dict[str, Any]: diff --git a/cc_remote/wrapper/claude_service.py b/cc_remote/wrapper/claude_service.py index c4368cae..8bc8f2fc 100644 --- a/cc_remote/wrapper/claude_service.py +++ b/cc_remote/wrapper/claude_service.py @@ -141,7 +141,14 @@ async def activate(machine, ctx) -> None: await client.call("metadata", {"value": {"key": ctx.key}}) if client.description["head"] > client.description["after"]: ctx.claude_service_background_replay = ReplayProjection(client.description["head"]) - if recovery is not None: + if recovery is not None and recovery.get("background_steer"): + await machine._set_state(ctx, "running") + elif recovery is not None: + # Older maintenance controllers left a successful compact uncommitted + # and reused the preceding prompt's metadata. Replay/commit its retained + # native terminal; never resubmit either that prompt or /compact. + if recovery["id"].startswith("compact-"): + recovery = {**recovery, "prompt": "/compact", "images": None, "files": None} ctx.active_msg_id = recovery["id"] ctx.claude_write_active = True ctx.needs_reload = False diff --git a/cc_remote/wrapper/claude_steer.py b/cc_remote/wrapper/claude_steer.py index 2d4588aa..d9249ade 100644 --- a/cc_remote/wrapper/claude_steer.py +++ b/cc_remote/wrapper/claude_steer.py @@ -2,6 +2,7 @@ from __future__ import annotations +import asyncio import hashlib import json import os @@ -15,8 +16,7 @@ async def handle(machine, ctx, cmd, reject): - if (ctx.state != "running" or ctx.write_state != "writable" - or not ctx.active_msg_id or ctx.translator is None): + if ctx.state != "running" or ctx.write_state != "writable": return await reject(ERR_NOT_STEERABLE, "当前没有可引导的 Claude 任务。") if not cmd.prompt and not cmd.images and not cmd.files: return await reject(ERR_NOT_STEERABLE, "消息内容为空。") @@ -89,6 +89,29 @@ async def apply_echo(machine, ctx, message, native_id): files=metadata.get("files")) +def adopt(machine, ctx, metadata): + """Consume an already accepted input only after its exact native echo.""" + previous = ctx.turn_task + + async def run(): + if previous is not None and not previous.done(): + await asyncio.shield(previous) + ctx.turn_task = asyncio.current_task() + origin_key = machine._claude_followup_origin_key(metadata.get("background_origin")) + if origin_key is not None: + ctx.claude_background_followups.pop(origin_key, None) + ctx.active_msg_id = metadata["id"] + ctx.claude_write_active = True + ctx.needs_reload = False + if ctx.state not in {"interrupting", "draining"}: + await machine._set_state(ctx, "running") + await machine._run_turn( + ctx, metadata.get("prompt", ""), _adopt_steer=True, + _recover_service=ctx.sdk.service_recovery is not None) + + ctx.turn_task = asyncio.create_task(run()) + + def cleanup(ctx): for directory in ctx.claude_steer_attachment_dirs: shutil.rmtree(directory, ignore_errors=True) diff --git a/cc_remote/wrapper/codex_agents.py b/cc_remote/wrapper/codex_agents.py new file mode 100644 index 00000000..f52e3c9c --- /dev/null +++ b/cc_remote/wrapper/codex_agents.py @@ -0,0 +1,173 @@ +"""Read-only, account-local Codex subagent details with proven ancestry.""" +from __future__ import annotations + +import hashlib +import json +import re + +from cc_remote.wrapper.codex_sessions import codex_rollout_path, _read_meta +from cc_remote.wrapper.codex_stream import codex_translate_history, _live_id +from cc_remote.wrapper.history_store import HistorySourceFingerprint + +_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9-]{0,127}$") +_MAX_BYTES = 64 * 1024 * 1024 +_MAX_RECORD = 16 * 1024 * 1024 +_PUBLIC_EVENTS = {"assistant_msg_start", "assistant_msg_end", "delta", + "tool_use", "tool_result", "tool_delta", "process"} + + +def _source(home, sid): + if not isinstance(sid, str) or not _ID.fullmatch(sid): + raise ValueError("协作代理标识无效") + path = codex_rollout_path(sid, codex_home=home) + if not path: + raise ValueError("协作代理记录尚未生成，请稍后重试") + meta = _read_meta(path) + if not meta or meta.get("id") != sid: + raise ValueError("协作代理记录身份不匹配") + return path, meta + + +def _parent(meta): + source = meta.get("source") + if not isinstance(source, dict): + return None + subagent = source.get("subagent") + spawn = subagent.get("thread_spawn") if isinstance(subagent, dict) else None + return spawn.get("parent_thread_id") if isinstance(spawn, dict) else None + + +def _records(path): + # Never allocate an unbounded transcript or record for a detail request. + with open(path, "rb") as stream: + while True: + offset = stream.tell() + if offset > _MAX_BYTES: + raise ValueError("协作代理记录过大，暂时无法读取完整过程") + line = stream.readline(_MAX_RECORD + 1) + if not line: + return + if len(line) > _MAX_RECORD: + raise ValueError("协作代理记录过大，暂时无法读取完整过程") + if not line.endswith(b"\n"): + return # The native writer may still be appending this row. + try: + row = json.loads(line) + except ValueError: + continue + if isinstance(row, dict) and isinstance(row.get("payload"), dict): + yield offset, row["payload"], row.get("type") + + +def _legacy_targets(path, run_id): + for _, payload, kind in _records(path): + if kind != "event_msg": + continue + item = payload.get("item") if payload.get("type") == "item_completed" else payload + if not isinstance(item, dict): + continue + typ = str(item.get("type", "")).replace("_", "").lower() + if typ == "subagentactivity": + if _live_id(item.get("id") or item.get("event_id"), "sub-agent") == run_id: + return [item.get("agentThreadId") or item.get("agent_thread_id")] + elif typ == "collabagenttoolcall": + if _live_id(item.get("id"), "collab-agent") == run_id: + return item.get("receiverThreadIds") or item.get("receiver_thread_ids") or [] + return [] + + +def _detail_events(events, root_sid, ancestors): + # Correlate before paging: the tool call and its activity may land on + # different pages. Only this child's translated (bounded) inputs qualify. + messages = {e.tool_use_id: e.input.get("message") for e in events + if e.type == "tool_use" and e.tool.rsplit(".", 1)[-1] + in {"send_message", "send_input", "sendInput"}} + public = [e.model_dump(exclude_none=True) for e in events if e.type in _PUBLIC_EVENTS] + for event in public: + if event["type"] != "process" or event["kind"] != "agent": + continue + inputs = event.get("input") or {} + run_id = inputs.get("agent_run_id") + if not isinstance(run_id, str) or not run_id.startswith("codex-agent:"): + continue + target = run_id.removeprefix("codex-agent:") + if target not in ancestors: + continue + message = messages.get(event["item_id"]) + if not isinstance(message, str) and event.get("tool") == "sendInput": + message = inputs.get("prompt") + role = "主代理" if target == root_sid else "上级代理" + has_message = isinstance(message, str) and bool(message) + event["title"] = f"向{role}汇报" if has_message else f"{role}动态" + if has_message: + event["detail"] = message + # An ancestor reference is an inline activity, not a descendant detail. + # Explicit null keeps legacy cards without this field navigable. + event["input"] = {"agent_run_id": None} + return public + + +def load_detail(home, root_sid, run_id, tool_result_max): + """Resolve a public card against native ancestry and return a bounded snapshot.""" + root_path, _ = _source(home, root_sid) + targets = ([run_id.removeprefix("codex-agent:")] if run_id.startswith("codex-agent:") + else _legacy_targets(root_path, run_id)) + if not targets or len(targets) > 32: + raise ValueError("未找到这个协作代理") + sources = [] + for target in dict.fromkeys(targets): + path, meta = _source(home, target) + parent = _parent(meta) + seen = {target} + while parent != root_sid: + if not parent or parent in seen or len(seen) >= 32: + raise ValueError("协作代理不属于当前会话") + seen.add(parent) + _, ancestor = _source(home, parent) + parent = _parent(ancestor) + sources.append((target, path, meta, (seen - {target}) | {root_sid})) + if len(sources) > 1: + # A native wait/send call can address several agents. Preserve each + # identity as a separate drill-down, rather than mixing their streams. + events = [{"type": "process", "item_id": f"codex-agent:{sid}", + "kind": "agent", "phase": "update", "status": "unknown", + "title": str(meta.get("agent_path") or "协作代理")[:1024]} + for sid, _, meta, _ in sources] + revision = hashlib.sha256(json.dumps(targets).encode()).hexdigest() + return events, "unknown", "协作代理", revision + sid, path, meta, ancestors = sources[0] + # Forked agents inherit parent context. Only their own native tasks belong + # in this panel, never the parent's copied transcript prefix. + offset = None + status = "unknown" + snapshot = HistorySourceFingerprint.capture(path) + native_boundary = next((pos for pos, p, t in _records(path) + if t == "event_msg" and p.get("type") == "thread_settings_applied" + and p.get("thread_id") == sid), None) + parent_turns = set() + if native_boundary is None: + parent_path, _ = _source(home, _parent(meta)) + parent_turns = {p.get("turn_id") for _, p, t in _records(parent_path) + if t == "event_msg" and p.get("type") == "task_started"} + for pos, payload, kind in _records(path): + if kind != "event_msg" or (native_boundary is not None and pos < native_boundary): + continue + typ = payload.get("type") + if typ == "task_started" and payload.get("turn_id") not in parent_turns: + if offset is None: + offset = pos + status = "running" + elif offset is not None and typ in {"task_complete", "task_completed", "turn_aborted"}: + status = ("interrupted" if typ == "turn_aborted" + else "failed" if payload.get("error") else "succeeded") + elif offset is not None and typ in {"task_failed", "turn_failed", "task_error"}: + status = "failed" + title = str(meta.get("agent_path") or "协作代理")[:1024] + if offset is None: + return [], "pending", title, snapshot.token + events, _ = codex_translate_history(path, tool_result_max, start_offset=offset, + end_offset=snapshot.size, snapshot_in_progress=status == "running") + if HistorySourceFingerprint.capture(path) != snapshot: + raise ValueError("协作代理记录已更新，请重试读取") + return (_detail_events(events, root_sid, ancestors), + status, title, snapshot.token) diff --git a/cc_remote/wrapper/codex_controls.py b/cc_remote/wrapper/codex_controls.py index 473f1230..a96f255d 100644 --- a/cc_remote/wrapper/codex_controls.py +++ b/cc_remote/wrapper/codex_controls.py @@ -33,6 +33,7 @@ class CodexControls: context_max_tokens: int | None = None context_window_tokens: int | None = None context_settings_set: bool = False + manual_compactions: tuple[str, ...] = () def as_dict(self) -> dict[str, object]: result = {} @@ -49,6 +50,8 @@ def as_dict(self) -> dict[str, object]: result["context_window_tokens"] = self.context_window_tokens if self.context_settings_set: result["context_settings_set"] = True + if self.manual_compactions: + result["manual_compactions"] = list(self.manual_compactions) return result @@ -96,6 +99,10 @@ def _controls(values: object) -> CodexControls: else raw.get("context_threshold_tokens")), context_window_tokens=_token_count(raw.get("context_window_tokens")), context_settings_set=raw.get("context_settings_set") is True, + manual_compactions=tuple( + value for value in (raw.get("manual_compactions") or [])[-64:] + if isinstance(value, str) and _SESSION_ID.fullmatch(value) + ) if isinstance(raw.get("manual_compactions"), list) else (), ) @@ -250,6 +257,7 @@ def update( context_max_tokens=existing.context_max_tokens, context_window_tokens=existing.context_window_tokens, context_settings_set=existing.context_settings_set, + manual_compactions=existing.manual_compactions, ) payload = controls.as_dict() updated = dict(self._sessions) @@ -326,6 +334,20 @@ def delete(self, session_id: str) -> None: self._persist(updated) self._sessions = updated + def remember_compaction(self, session_id: str, turn_id: str) -> None: + """Keep the explicit command only for its proven native compaction turn.""" + session_id, turn_id = _session_id(session_id), _session_id(turn_id) + with self._lock: + existing = _controls(self._sessions.get(session_id)) + turns = tuple(value for value in existing.manual_compactions if value != turn_id) + controls = replace(existing, manual_compactions=(*turns, turn_id)[-64:]) + updated = dict(self._sessions) + updated[session_id] = controls.as_dict() + while len(updated) > _MAX_ENTRIES: + updated.pop(next(iter(updated))) + self._persist(updated) + self._sessions = updated + def set_cwd_override( self, session_id: str, cwd_override: str | None, ) -> CodexControls: @@ -343,6 +365,7 @@ def set_cwd_override( context_max_tokens=existing.context_max_tokens, context_window_tokens=existing.context_window_tokens, context_settings_set=existing.context_settings_set, + manual_compactions=existing.manual_compactions, ) payload = controls.as_dict() updated = dict(self._sessions) @@ -374,6 +397,7 @@ def clear_cwd_override_if_matches( context_max_tokens=existing.context_max_tokens, context_window_tokens=existing.context_window_tokens, context_settings_set=existing.context_settings_set, + manual_compactions=existing.manual_compactions, ) payload = controls.as_dict() updated = dict(self._sessions) @@ -415,6 +439,7 @@ def restore_cwd_override_after_failed_set( context_max_tokens=existing.context_max_tokens, context_window_tokens=existing.context_window_tokens, context_settings_set=existing.context_settings_set, + manual_compactions=existing.manual_compactions, ) payload = controls.as_dict() updated = dict(durable) diff --git a/cc_remote/wrapper/codex_handle.py b/cc_remote/wrapper/codex_handle.py index 1d2009c8..29692235 100644 --- a/cc_remote/wrapper/codex_handle.py +++ b/cc_remote/wrapper/codex_handle.py @@ -1666,6 +1666,10 @@ def __init__(self, cfg, cwd: Optional[str] = None, self._goal_baseline_loaded = False self._goal_prompt_pending: OrderedDict[str, str] = OrderedDict() self._goal_prompt_candidate: Optional[_GoalPromptCandidate] = None + self._manual_compact_request: asyncio.Future[bool] | None = None + self._manual_compact_candidate: str | None = None + self._manual_compact_generation = -1 + self._explicit_interrupt_turn: tuple[int, str] | None = None # Prompt correlation ends as soon as turn/started claims its candidate. # Keep retired objectives independently across response -> notification # gaps. Authoritative external replacements extend the same fence so @@ -3952,6 +3956,9 @@ async def interrupt(self) -> None: fence.suppression_enabled = False if not (self.proc and self.thread_id and self.turn_id): raise RuntimeError("codex turn is not running") + # Spontaneous turns have no managed continuation fence. Stop still owns + # their real terminal, including a compaction item arriving after Stop. + self._explicit_interrupt_turn = (self._generation, self.turn_id) target_thread_id = self.thread_id target_turn_id = self._review_execution_turn_id or self.turn_id try: @@ -5441,8 +5448,36 @@ async def compact_thread(self) -> None: await self._release_managed_compaction_continuation() if self.turn_active: raise RuntimeError("codex thread is busy") - await self._request( - "thread/compact/start", {"threadId": self.thread_id}) + accepted = asyncio.get_running_loop().create_future() + self._manual_compact_request = accepted + self._manual_compact_candidate = None + self._manual_compact_generation = self._generation + try: + await self._request("thread/compact/start", {"threadId": self.thread_id}) + except BaseException: + accepted.set_result(False) + if self._manual_compact_request is accepted: + self._manual_compact_request = None + raise + else: + accepted.set_result(True) + + async def manual_compact_for_turn(self, turn_id: str) -> bool: + """Called by the consumer only after native compaction output proves it. + + The stdout reader must never await the RPC response it itself delivers. + Keep early notifications queued until the control request is accepted. + """ + request = self._manual_compact_request + if (request is None or self._manual_compact_candidate != turn_id + or self._manual_compact_generation != self._generation): + return False + accepted = await asyncio.shield(request) + if (self._manual_compact_request is not request + or self._manual_compact_generation != self._generation): + return False + self._manual_compact_request = None + return accepted async def rollback_thread(self, num_turns: int) -> dict[str, Any]: assert self.thread_id, "connect() first" @@ -7174,6 +7209,7 @@ async def _dispatch(self, m: dict, raw_size: Optional[int] = None) -> None: and target_turn_id == self._compaction_continuation_turn_id and target_turn_id == self.turn_id and self.turn_active + and self._explicit_interrupt_turn != (self._generation, target_turn_id) ) if compact_continuation_boundary: self._compaction_continuation_turn_id = None @@ -7222,6 +7258,11 @@ async def _dispatch(self, m: dict, raw_size: Optional[int] = None) -> None: was_active = self.turn_active turn = (m.get("params") or {}).get("turn") or {} turn_id = turn.get("id") + if (not was_active and not review_execution_frame + and isinstance(turn_id, str) and turn_id + and self._manual_compact_request is not None + and self._manual_compact_candidate is None): + self._manual_compact_candidate = turn_id if (isinstance(turn_id, str) and turn_id and not review_execution_frame): if self._compaction_continuation_turn_id not in { diff --git a/cc_remote/wrapper/codex_history.py b/cc_remote/wrapper/codex_history.py index f9f47484..a3a9218c 100644 --- a/cc_remote/wrapper/codex_history.py +++ b/cc_remote/wrapper/codex_history.py @@ -17,6 +17,7 @@ from typing import Any, Awaitable, Callable from cc_remote.protocol import TurnEnd, TurnResult, UserMsg +from cc_remote.wrapper.claude_compaction import compact_completion_events from cc_remote.wrapper.codex_history_prefetch import CodexHistoryPrefetch from cc_remote.wrapper.codex_rpc import ( CodexRpcRejected, @@ -389,6 +390,12 @@ def serialized(event, timestamp: float | None) -> dict[str, Any]: final_segment = segment_index == segment_count - 1 status = turn["status"] if final_segment and status != "inProgress": + if (status == "completed" + and any(item.get("type") == "contextCompaction" for item in segment) + and any(item.get("type") == "userMessage" + and _user_message(item, ts=started_ts).prompt == "/compact" for item in segment)): + events.extend(serialized(event, completed_ts) + for event in compact_completion_events(native_turn_id, native_turn_id)) translated = translator.feed({ "method": "turn/completed", "params": { diff --git a/cc_remote/wrapper/codex_sessions.py b/cc_remote/wrapper/codex_sessions.py index 60466a2c..64f70486 100644 --- a/cc_remote/wrapper/codex_sessions.py +++ b/cc_remote/wrapper/codex_sessions.py @@ -1112,7 +1112,11 @@ def _rollout_path( scanned += 1 resolved = os.path.realpath(match) if os.path.commonpath((root, resolved)) == root: - return match + # Filename substrings can match other threads (child vs + # grandchild), in filesystem-dependent directory order. + meta = _read_meta(match) + if meta and meta.get("id") == session_id: + return match return None except Exception: return None diff --git a/cc_remote/wrapper/codex_stream.py b/cc_remote/wrapper/codex_stream.py index 2a1cea44..1e62a632 100644 --- a/cc_remote/wrapper/codex_stream.py +++ b/cc_remote/wrapper/codex_stream.py @@ -3928,6 +3928,9 @@ def _collab_event(item: dict, turn_id: str | None, completed: bool): "receivers": item.get("receiverThreadIds"), "agents": safe_states, }, 64 * 1024) + receivers = item.get("receiverThreadIds") + if isinstance(receivers, list) and len(receivers) == 1 and isinstance(receivers[0], str): + input_value["agent_run_id"] = f"codex-agent:{receivers[0]}" return ProcessEvent( item_id=_live_id(item.get("id"), "collab-agent"), kind="agent", @@ -3959,9 +3962,12 @@ def _subagent_event(item: dict, turn_id: str | None, completed: bool): title={ "started": "协作代理已启动", "interacted": "协作代理有新进展", + "completed": "协作代理已完成", "interrupted": "协作代理已中断", }.get(kind, "协作代理"), summary=path or None, + input={"agent_run_id": f"codex-agent:{item['agentThreadId']}"} + if isinstance(item.get("agentThreadId"), str) else None, ) @@ -4968,7 +4974,25 @@ def close_turn( upsert_tool_result(result) elif t == "event_msg" and payload_type == "item_completed": item = p.get("item") if isinstance(p.get("item"), dict) else {} - if str(item.get("type") or "").lower() == "plan": + item_type = str(item.get("type") or "").replace("_", "").lower() + if item_type == "subagentactivity": + open_assistant_only_turn() + events.append(_subagent_event({ + **item, "agentThreadId": item.get("agentThreadId") or item.get("agent_thread_id"), + "agentPath": item.get("agentPath") or item.get("agent_path"), + }, _history_optional_turn_id(p.get("turn_id") or active_turn_id or pending_turn_id), True)) + turn_visible = True + elif item_type == "collabagenttoolcall": + open_assistant_only_turn() + events.append(_collab_event({ + **item, + "receiverThreadIds": item.get("receiverThreadIds") or item.get("receiver_thread_ids"), + "senderThreadId": item.get("senderThreadId") or item.get("sender_thread_id"), + "agentsStates": item.get("agentsStates") or item.get("agents_states"), + "reasoningEffort": item.get("reasoningEffort") or item.get("reasoning_effort"), + }, _history_optional_turn_id(p.get("turn_id") or active_turn_id or pending_turn_id), True)) + turn_visible = True + elif item_type == "plan": open_assistant_only_turn() item_id = _history_id( item.get("id"), "plan-detail", line_no, raw_ts) diff --git a/cc_remote/wrapper/history_store.py b/cc_remote/wrapper/history_store.py index 63c128f5..aa94e79b 100644 --- a/cc_remote/wrapper/history_store.py +++ b/cc_remote/wrapper/history_store.py @@ -67,7 +67,8 @@ # retaining the old boundary for task bookkeeping without a reply. # v41 keeps native isMeta recovery prompts inside their original human turn. # v42 replaces recovered text prefixes and bounds summary answer block counts. -_SCHEMA_VERSION = 42 +# v43 makes manual /compact a visible turn owning its native boundary. +_SCHEMA_VERSION = 43 _FINGERPRINT_SAMPLE_BYTES = 64 * 1024 _DEFAULT_MAX_ENTRIES = 128 _DEFAULT_MAX_BYTES = 64 * 1024 * 1024 @@ -1294,6 +1295,13 @@ def _select_page_row( def _ensure_schema(self) -> None: with self._connect() as connection: current = int(connection.execute("PRAGMA user_version").fetchone()[0]) + if current in range(10, 43): + # /compact is now a visible native command. Its graph boundary + # and the derived pages must agree; retain source-bound assets. + for table in ("history_pages", "history_turn_details"): + connection.execute(f"DELETE FROM {table} WHERE engine='claude'") + for table in ("claude_compact_sources", "claude_compact_records", "claude_compact_queue"): + connection.execute(f"DROP TABLE IF EXISTS {table}") if current in range(10, 42): # v42 bounds answer summaries and handles recovered prefixes. # Full source events, images and native graph indexes remain @@ -1448,8 +1456,8 @@ def _ensure_schema(self) -> None: for table in ("history_pages", "history_turn_details"): connection.execute( f"DELETE FROM {table} WHERE engine='codex'") - elif current in (21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41): - # The independent v22-v42 invalidations above suffice. + elif current in (21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42): + # The independent v22-v43 invalidations above suffice. pass elif current not in (0, _SCHEMA_VERSION): # v9 changes the invariant of history_turn_details: those rows diff --git a/cc_remote/wrapper/machine.py b/cc_remote/wrapper/machine.py index e91a2392..aa83b0ca 100644 --- a/cc_remote/wrapper/machine.py +++ b/cc_remote/wrapper/machine.py @@ -38,6 +38,7 @@ from __future__ import annotations from cc_remote.wrapper import claude_service +from cc_remote.wrapper.claude_compaction import compact_completion_events import asyncio import base64 @@ -6293,6 +6294,7 @@ def _configure_claude_sdk_callbacks( from cc_remote.wrapper import claude_steer sdk.native_close_callback = lambda: claude_steer.cleanup(ctx) + sdk.steer_adoption_callback = lambda metadata: claude_steer.adopt(self, ctx, metadata) claude_service.configure(self, ctx) @staticmethod @@ -6789,6 +6791,7 @@ def _claude_autonomous_followup_pending(ctx: SessionContext) -> bool: and ( ctx.claude_background_followups or ctx.claude_background_followup_overflow + or getattr(ctx.sdk, "autonomous_steer_pending", False) ) ) @@ -7172,7 +7175,9 @@ async def _compact_managed_claude_context( ) compact_event: ProcessEvent | None = None terminal: ResultMessage | None = None - ctx.sdk.next_turn_id = f"compact-{uuid4().hex}" + compact_turn_id = f"compact-{uuid4().hex}" + ctx.sdk.next_turn_id = compact_turn_id + ctx.sdk.service_turn_metadata = {"id": compact_turn_id, "prompt": "/compact"} ctx.claude_write_active = True try: await ctx.sdk.query("/compact") @@ -7189,6 +7194,9 @@ async def _compact_managed_claude_context( await self._emit(ctx, event) if isinstance(message, ResultMessage): terminal = message + ack = getattr(ctx.sdk, "ack_service_message", None) + if ack is not None: + await ack(message, turn_id=compact_turn_id) break finally: release_background = getattr( @@ -15401,6 +15409,14 @@ def _history_refresh_backoff_seconds(self, scan_elapsed: float) -> float: self.HISTORY_REFRESH_MAX_INTERVAL_SECONDS, ) + def _manual_codex_compact_users(self, sid: str, turn_ids) -> dict[str, UserMsg]: + store = getattr(self, "_codex_controls", None) + if store is None: + return {} + commands = set(store.get(sid).manual_compactions) + return {turn_id: UserMsg(msg_id=turn_id, prompt="/compact") + for turn_id in turn_ids if turn_id in commands} + async def _recover_official_codex_user( self, sid: str, @@ -15411,6 +15427,10 @@ async def _recover_official_codex_user( max_reverse_scan_bytes: int | None = None, ) -> UserMsg | None: """Recover one exact persisted user row, including inline image bytes.""" + if user_index == 0: + command = self._manual_codex_compact_users(sid, (native_turn_id,)).get(native_turn_id) + if command is not None: + return command path = await asyncio.to_thread(self._codex_rollout_for_wire, sid) if not path: return None @@ -15429,13 +15449,17 @@ async def _recover_official_codex_users( native_turn_ids: tuple[str, ...], ) -> dict[str, UserMsg]: """Recover assistant-only Goal prompts with one bounded rollout pass.""" + commands = self._manual_codex_compact_users(sid, native_turn_ids) + native_turn_ids = tuple(turn_id for turn_id in native_turn_ids if turn_id not in commands) + if not native_turn_ids: + return commands path = await asyncio.to_thread(self._codex_rollout_for_wire, sid) if not path: - return {} + return commands try: source_stat = await asyncio.to_thread(os.stat, path) except OSError: - return {} + return commands fingerprint = ( int(source_stat.st_dev), int(source_stat.st_ino), @@ -15461,7 +15485,7 @@ async def _recover_official_codex_users( previous_misses[key] = miss pending.append(native_turn_id) if not pending: - return {} + return commands recovery = await asyncio.to_thread( codex_history_turn_users, path, @@ -15493,7 +15517,7 @@ async def _recover_official_codex_users( > self.CODEX_GOAL_RECOVERY_MISS_CACHE_ENTRIES ): self._codex_goal_recovery_misses.popitem(last=False) - return recovery.users + return {**recovery.users, **commands} async def _build_official_codex_history( self, @@ -16917,7 +16941,7 @@ async def send( ) async def _handle_get_agent_detail(self, cmd) -> AgentDetail: - """Return one read-only Claude Agent process page without resuming it.""" + """Return one read-only Agent process page without resuming it.""" started_at = time.perf_counter() sid = self._resolve_session_alias(cmd.session_id) or cmd.session_id revision = self._history_revision(sid) @@ -16960,7 +16984,7 @@ async def send( ) await self.transport.send(detail) log.info( - "Claude Agent detail sent", + "Agent detail sent", session_id=sid, run_id=cmd.run_id, events=len(detail.events), @@ -16977,11 +17001,39 @@ async def send( self._watch_session(sid) watch = self._watch.get(sid) or {} ctx = self._ctx_by_sid(sid) - if ((ctx is not None and ctx.engine != "claude") - or (ctx is None and watch.get("engine") == "codex")): - return await send(error="当前会话不支持协作代理详情") if ctx is not None and ctx.space != "code": return await send(error="Work 不提供协作代理详情") + if ((ctx is not None and ctx.engine == "codex") + or (ctx is None and watch.get("engine") == "codex")): + from cc_remote.wrapper.codex_agents import load_detail + + try: + profile, native_sid = self._codex_target(sid) + if await asyncio.to_thread( + self._work.for_engine("codex").get_by_session, + native_sid, codex_profile_id=profile.id) is not None: + return await send(error="Work 不提供协作代理详情") + rows, status, title, source_revision = await asyncio.to_thread( + load_detail, profile.home, native_sid, cmd.run_id, + self.cfg.tool_result_max) + if (getattr(cmd, "detail_revision", None) + and cmd.detail_revision != source_revision): + return await send(error="协作代理详情已更新，请重新打开", + detail_revision=source_revision) + page, more, oldest, newer, newest = _turn_detail_page( + rows, before=getattr(cmd, "before", None), + limit=getattr(cmd, "limit", 192), + max_bytes=min(8 * 1024 * 1024, + max(512 * 1024, self.cfg.ws_max_size_bytes // 2))) + return await send(page, title=title, status=status, + detail_revision=source_revision, has_more=more, + oldest_cursor=oldest, has_newer=newer, newer_cursor=newest) + except ValueError as exc: + return await send(error=str(exc)[:4096]) + except Exception as exc: + log.warning("Codex Agent detail read failed", session_id=sid, + error_type=type(exc).__name__) + return await send(error="协作代理详情暂时不可用，请稍后重试") try: claude_profile, native_sid = self._claude_target(sid) except ValueError: @@ -21563,6 +21615,13 @@ async def _on_claude_background_message( the autonomous work extends its origin turn rather than creating a new visible human turn. """ + if getattr(message, "_cc_steer_cancelled", None): + from cc_remote.wrapper import claude_steer + + await self._emit(ctx, await claude_steer.apply_echo(self, ctx, message, None)) + await self._settle_claude_lifecycle_if_quiescent(ctx) + self._schedule_query_queue_drain(ctx) + return if await self._observe_claude_rate_limit_message(ctx, message): return await self._observe_claude_model_fallback(ctx, message) @@ -21915,6 +21974,8 @@ async def _run_codex_spontaneous_turn( repair_history = False seen_user_item_ids: set[str] = set() anchor_recovery_attempted = False + manual_compact = False + compact_completed = False def start_restart_watch() -> None: nonlocal restart_watch_task @@ -22459,6 +22520,29 @@ async def interrupted_handoff_result() -> Optional[str]: await ctx.codex_steer_gate.wait() await self._confirm_uncertain_codex_steer(ctx, raw) + params = raw.get("params") if isinstance(raw.get("params"), dict) else {} + item = params.get("item") if isinstance(params.get("item"), dict) else {} + is_compaction = (raw.get("method") == "thread/compacted" + or item.get("type") == "contextCompaction") + manual = getattr(ctx.sdk, "manual_compact_for_turn", None) + if is_compaction and not manual_compact and manual is not None: + manual_compact = await manual(current_turn_id) + if manual_compact: + user = UserMsg(msg_id=logical_msg_id, prompt="/compact") + sid = self._ctx_wire_sid(ctx) or ctx.key + store = getattr(self, "_codex_controls", None) + if store is not None: + try: + await asyncio.to_thread(store.remember_compaction, sid, current_turn_id) + except CodexControlStoreError: + log.warning("manual compact command could not be persisted", session_id=sid) + self._invalidate_codex_history(sid) + self._codex_history.remember_automatic_user(sid, current_turn_id, user) + ctx.codex_spontaneous_anchor_id = logical_msg_id + ctx.active_msg_id = logical_msg_id + await self._emit(ctx, user) + await self._emit(ctx, StateEvent( + state=ctx.state, phase="waiting", detail="压缩中", msg_id=logical_msg_id)) published_user = await publish_live_user(raw) if ( not published_user @@ -22467,7 +22551,13 @@ async def interrupted_handoff_result() -> Optional[str]: ): await ensure_automatic_anchor() events = translator.feed(raw) + compact_completed = compact_completed or any( + isinstance(event, ProcessEvent) and event.kind == "compaction" + and event.status == "succeeded" for event in events) terminal = is_turn_terminal(raw) + if (manual_compact and compact_completed and terminal + and _codex_terminal_status(raw) == "completed"): + events = [*compact_completion_events(current_turn_id, logical_msg_id), *events] completed_after_overflow = ( overflowed and terminal and _codex_terminal_status(raw) == "completed" @@ -32182,26 +32272,36 @@ async def _handle_compact_session(self, cmd): ctx.claude_compaction_revision != compact_revision ) if not compacted_while_waiting: - try: - await self._compact_managed_claude_context( - ctx, reason="manual compact command") - except Exception as exc: - log.warning( - "Claude compact failed", - session_id=ctx.session_id, - error_type=type(exc).__name__, - ) + if (ctx.state != "idle" + or getattr(ctx.sdk, "is_claude_broker", False) + or self._claude_has_background_work(ctx)): error = Error( - code=ERR_INTERNAL, - message=( - "Claude 原生上下文压缩失败；" - "当前上下文未被标记为已压缩" - ), + code=ERR_BUSY, + message="Claude 会话正在使用中，暂时无法压缩上下文", sid=self._ctx_wire_sid(ctx), to=getattr(cmd, "client_id", None), ) await self.transport.send(error) return error + # A human command has the same ownership, interrupt/drain, + # service commit and reconnect contract as an ordinary turn. + # Background maintenance keeps the internal helper above. + ctx.active_msg_id = f"compact-{uuid4().hex}" + ctx.interrupt_event.clear() + ctx.interrupt_deadline = None + ctx.state = "running" + await self._emit(ctx, StateEvent( + state="running", phase="waiting", detail="压缩中", + msg_id=ctx.active_msg_id)) + ctx.turn_task = asyncio.create_task(self._run_turn(ctx, "/compact")) + notice = Notice( + notice_id=f"compact-{uuid4().hex}", severity="info", + category="runtime", title="上下文压缩已启动", + message="Claude 正在压缩当前会话的上下文。", + thread_id=self._ctx_wire_sid(ctx), sid=self._ctx_wire_sid(ctx), + to=getattr(cmd, "client_id", None)) + await self.transport.send(notice) + return notice if self._claude_auto_compact_event(ctx).pending: ctx.auto_compact_compaction_done = True ctx.auto_compact_phase = "waiting_terminal" @@ -32230,7 +32330,10 @@ async def _handle_compact_session(self, cmd): return ctx sid = self._ctx_wire_sid(ctx) or cmd.session_id try: - await ctx.sdk.compact_thread() + async with ctx.query_lock, ctx.launch_lock: + if ctx.state != "idle": + raise RuntimeError("Codex session is busy") + await ctx.sdk.compact_thread() notice = Notice( notice_id=f"compact-{uuid4().hex}", severity="info", @@ -37208,8 +37311,10 @@ async def _run_turn( *, launch_receipt: asyncio.Future[bool] | None = None, _recover_service: bool = False, + _adopt_steer: bool = False, ) -> None: is_codex = ctx.engine == "codex" + manual_compact = not is_codex and prompt.strip() == "/compact" and not images and not files is_codex_shared = self._codex_shared_affinity(ctx) service_client = getattr(ctx.sdk, "client", None) service_replay = ( @@ -37232,6 +37337,7 @@ async def _run_turn( item_titles=ctx.claude_item_titles, item_meta=ctx.claude_item_meta, item_commands=ctx.claude_item_commands, + manual_compact=manual_compact, )) if not is_codex: ctx.claude_client_alias_bound_msg_id = None @@ -37728,7 +37834,7 @@ async def reconnect_claude(reason: str) -> None: await self._publish_claude_auto_compact(ctx) try: - if not _recover_service: + if not _recover_service and not _adopt_steer: if not is_codex: check_delivery = getattr(ctx.sdk, "check_service_delivery", None) if check_delivery is not None: @@ -37791,7 +37897,7 @@ async def reconnect_claude(reason: str) -> None: sid=ctx.session_id, ) await reconnect_claude("message pump failure") - if (not is_codex + if (not is_codex and not manual_compact and self._claude_auto_compact_event(ctx).pending): async with ctx.query_lock: auto_event, _ = await self._apply_pending_claude_auto_compact( @@ -38074,6 +38180,10 @@ async def msg_stream(): files=file_meta, )) settle_launch(True) + if manual_compact and ctx.state == "running": + await self._emit(ctx, StateEvent( + state="running", phase="waiting", detail="压缩中", + msg_id=ctx.active_msg_id)) # Codex sessions don't emit a Model event like cc's init SystemMessage, # so announce the configured codex model (gpt-*) once — else the header # would keep showing a stale Claude model. @@ -38299,6 +38409,10 @@ async def msg_stream(): ack = getattr(ctx.sdk, "ack_service_message", None) if ack is not None: await ack(msg, turn_id=ctx.active_msg_id) + if manual_compact and ctx.translator.compact_boundary_id and not msg.is_error: + ctx.claude_compaction_revision += 1 + if self._claude_auto_compact_event(ctx).pending: + ctx.auto_compact_compaction_done = True break if not is_codex: diff --git a/cc_remote/wrapper/sdk.py b/cc_remote/wrapper/sdk.py index a6bbe8c4..4475cfaa 100644 --- a/cc_remote/wrapper/sdk.py +++ b/cc_remote/wrapper/sdk.py @@ -226,6 +226,8 @@ def __init__( self.service_defer_events = False self._steers = PendingSteers() self._turn_root_id: str | None = None + self._autonomous_steer_root: dict | None = None + self.steer_adoption_callback: Callable[[dict], None] | None = None self._context_revision = 0 # reasoning effort is a spawn-time flag (--effort), not a runtime setter. # `effort` is the desired level; `applied_effort` is what the live client @@ -704,12 +706,18 @@ async def connect( launch_auto_compact[1]) self._start_message_pump() if self.service_recovery is not None: - self._turn_active = True self._turn_root_id = self.service_recovery["id"] - self._turn_origin_id = self.service_recovery["id"] - self._message_route_owner = "managed" + if self.service_recovery.get("background_steer"): + self._autonomous_steer_root = self.service_recovery + self._turn_origin_id = self.service_recovery.get("previous_origin_id") + else: + self._turn_active = True + self._turn_origin_id = self.service_recovery["id"] + self._message_route_owner = "managed" elif hasattr(self.client, "description"): self._turn_origin_id = self.client.description.get("origin_id") + if hasattr(self.client, "description"): + self._steers.pending = dict(self.client.description.get("pending_steers") or {}) if not self.service_defer_events: self.start_service_events() log.info("sdk connected", resume=bool(resume_id), fork=fork, cwd=opts.cwd, @@ -988,7 +996,7 @@ async def query(self, prompt) -> None: if self._message_pump_task is not None: if self._message_pump_task.done(): raise RuntimeError("Claude SDK message pump is not running") from self._message_pump_error - if self._turn_active or self._turn_consumer_active: + if self._turn_active or self._turn_consumer_active or self._autonomous_steer_root: raise RuntimeError("Claude SDK already has an active response") # Do not let a Query overtake a task notification that the sole # reader already routed but whose machine callback has not yet @@ -1052,30 +1060,47 @@ async def query(self, prompt) -> None: # going through connect(). Real SDK connections always use the sole pump. await client.query(prompt) + @property + def autonomous_steer_pending(self) -> bool: + return bool(self._autonomous_steer_root and not self._autonomous_steer_root.get("adopted")) + async def steer(self, prompt, *, native_id: str, metadata: dict) -> None: """Write `priority=next` without interrupting or creating another reader.""" async with self._control_request_lock: async with self._message_route_lock: self.check_service_delivery() client = self.client + background_id = self._steers.background_id if not ( + self._turn_active or self._autonomous_steer_root or self._turn_consumer_active) else None if (client is None or self.control_plane_failed - or self.message_pump_failed or not self._turn_active + or self.message_pump_failed + or not (self._turn_active or self.autonomous_steer_pending or background_id) or self._message_pump_task is None): raise ClaudeSteerRejected("Claude has no active response") - if hasattr(client, "steer"): - # The persistent owner checks its live boundary too; its - # terminal may already be ahead of this controller's poll. - await client.steer(prompt, native_id=native_id, - metadata=metadata, turn_id=self._turn_root_id) - else: + if not hasattr(client, "steer") or native_id not in self._steers.pending: self._steers.add(native_id, metadata) + if background_id: + self._autonomous_steer_root = {**metadata, "background_id": background_id, + "background_origin": self._steers.background_origin_data, + "previous_origin_id": self._turn_origin_id} + self._turn_root_id = metadata["id"] + try: + if hasattr(client, "steer"): + await client.steer(prompt, native_id=native_id, + metadata=metadata, turn_id=self._turn_root_id, + **({"background_id": background_id} if background_id else {})) + else: - async def stream(): - yield steer_message(prompt, native_id) + async def stream(): + yield steer_message(prompt, native_id) - # Keep registration on uncertain writes: a late exact echo - # can still confirm acceptance. Never retry this as Query. - await client.query(stream()) + # Retain ownership on an uncertain write; never retry. + await client.query(stream()) + except ClaudeSteerRejected: + self._steers.pending.pop(native_id, None) + if background_id: + self._autonomous_steer_root = None + raise async def interrupt(self) -> None: assert self.client is not None @@ -1468,6 +1493,7 @@ def _start_message_pump(self) -> None: self._turn_consumer_active = False self._steers = PendingSteers() self._turn_root_id = None + self._autonomous_steer_root = None self._message_route_owner = None self._message_pump_error = None self._service_delivery_error = None @@ -1538,6 +1564,22 @@ async def _message_pump(self, client: ClaudeSDKClient) -> None: and not bool(getattr(message, "is_error", False))): self.context_probe_suppressed = False async with self._message_route_lock: + if steer is not None and self.autonomous_steer_pending: + # Drain only earlier autonomous frames before activating + # the new human route. Tools and Results before the exact + # echo must never acquire this user's identity. + await self._background_callbacks_drained.wait() + self._turn_active = True + self._autonomous_steer_root["adopted"] = True + self._pending_turn_background_release = asyncio.Event() + self._pending_turn_origin_id = steer["id"] + self._steers.handoff_background(self._autonomous_steer_root.get("background_id")) + if self.steer_adoption_callback: + self.steer_adoption_callback({**steer, "background_origin": + self._autonomous_steer_root.get("background_origin")}) + if (getattr(message, "_cc_steer_cancelled", None) + and not self._steers.pending and not self._turn_active): + self._autonomous_steer_root = None origin_kind = _message_origin_kind(message) top_level_user = bool( isinstance(message, UserMessage) @@ -1636,8 +1678,17 @@ async def _message_pump(self, client: ClaudeSDKClient) -> None: self._background_callbacks_pending += 1 self._background_callbacks_drained.clear() try: + background_owner = service_origin if service_seed else self._turn_origin_id + if not service_seed and self._autonomous_steer_root: + from cc_remote.claude_steering import _origin_key + + old_origin = self._autonomous_steer_root.get("background_origin") + origin = getattr(message, "origin", None) + if (isinstance(origin, dict) and isinstance(old_origin, dict) + and _origin_key(origin) == _origin_key(old_origin)): + background_owner = self._autonomous_steer_root.get("previous_origin_id") await self._background_messages.put( - (message, release, service_origin if service_seed else self._turn_origin_id)) + (message, release, background_owner)) except BaseException: self._background_callback_completed() raise @@ -1738,6 +1789,7 @@ async def _receive_response_pumped(self): raise message.error yield message if isinstance(message, ResultMessage): + self._autonomous_steer_root = None return finally: self._turn_consumer_active = False @@ -1839,6 +1891,7 @@ async def _stop_message_pump(self) -> None: self._turn_active = False self._turn_consumer_active = False self._message_route_owner = None + self._autonomous_steer_root = None self._turn_background_release = None self._pending_turn_background_release = None self._pending_turn_origin_id = None diff --git a/cc_remote/wrapper/stream.py b/cc_remote/wrapper/stream.py index 341a71ae..9eeba33b 100644 --- a/cc_remote/wrapper/stream.py +++ b/cc_remote/wrapper/stream.py @@ -39,7 +39,9 @@ TurnEnd, TurnResult, UserMsg, ) from cc_remote.wrapper.sanitize import bounded_text, bounded_tool_input -from cc_remote.wrapper.claude_compaction import compact_metadata +from cc_remote.wrapper.claude_compaction import ( + compact_completion_events, compact_metadata, manual_compact_prompt, +) from cc_remote.wrapper.claude_model_fallback import FALLBACK_TOOL, model_fallback_event from cc_remote.wrapper.turn_changes import native_claude_diff from cc_remote.wrapper.token_usage import UsageLedger, native_usage @@ -283,6 +285,10 @@ def replayed_user_message_id(message: Any) -> str | None: # can start before that late marker reaches the SDK reader; binding it # would give the next optimistic row the preceding turn's UUID. return None + raw_text = message.content if isinstance(message.content, str) else "".join( + block.text for block in message.content if isinstance(block, TextBlock)) + if raw_text and _is_meta_user_text(raw_text): + return None content = message.content if isinstance(message.content, list) else [] if any(isinstance(block, (ToolResultBlock, ServerToolResultBlock)) for block in content): @@ -664,8 +670,11 @@ def __init__(self, tool_result_max: int, turn_id: str | None = None, item_turns: dict[str, str] | None = None, item_titles: dict[str, str] | None = None, item_meta: dict[str, tuple[str, str | None]] | None = None, - item_commands: dict[str, str] | None = None): + item_commands: dict[str, str] | None = None, + manual_compact: bool = False): self.tool_result_max = tool_result_max + self.manual_compact = manual_compact + self.compact_boundary_id: str | None = None self.turn_id = _wire_id(turn_id, "turn") if turn_id else None # These maps are optionally shared by every translator for one resident # session. Claude's queue is continuous across ResultMessage boundaries; @@ -1475,6 +1484,7 @@ def _feed_compaction(self, msg: SystemMessage) -> list[ProcessEvent]: msg.data if isinstance(msg.data, dict) else {}) if event is None: return [] + self.compact_boundary_id = event.item_id event.turn_id = self.turn_id if self._compaction_id is not None: # Status UUIDs and persisted boundary UUIDs differ. Bind the live @@ -1599,10 +1609,13 @@ def feed(self, msg) -> list: turn_id=self.turn_id, channel="final")) for tool_id in sorted({key[0] for key in self._tool_pending}): events.extend(self._flush_tool_deltas(tool_id)) + missing_boundary = self.manual_compact and self.compact_boundary_id is None + if self.manual_compact and self.compact_boundary_id and not msg.is_error: + events.extend(compact_completion_events(self.compact_boundary_id, self.turn_id)) terminal = TurnEnd(result=TurnResult( - subtype=msg.subtype, + subtype="error" if missing_boundary and not msg.is_error else msg.subtype, duration_ms=msg.duration_ms, - is_error=msg.is_error, + is_error=msg.is_error or missing_boundary, total_cost_usd=msg.total_cost_usd, num_turns=msg.num_turns, ), turn_id=self._last_assistant_uuid, @@ -1990,6 +2003,31 @@ def _compact_chain_index( for uid in chain_ids ): return None + # Claude persists the completed boundary before replaying the command that + # caused it. Put that exact /compact row before its boundary for pagination + # and narrative ownership; ordinary post-compaction prompts stay in place. + commands: dict[str, str] = {} + boundary = None + try: + with open(source_path, "rb") as source: + for uid in chain_ids: + if rows[uid][:2] == ("system", "compact_boundary"): + boundary = uid + elif boundary is not None and rows[uid][6]: + row = _indexed_transcript_row(source, uid, rows[uid], + max_record_bytes=max_record_bytes) + compact = _indexed_transcript_row(source, boundary, rows[boundary], + max_record_bytes=max_record_bytes) + if (row and compact and compact_metadata(compact).get("trigger") != "auto" + and manual_compact_prompt((row.get("message") or {}).get("content"))): + commands[boundary] = uid + boundary = None + except OSError: + return None + if commands: + moved = set(commands.values()) + chain_ids = [part for uid in chain_ids if uid not in moved + for part in ([commands[uid], uid] if uid in commands else [uid])] return chain_ids, rows, queued @@ -2482,6 +2520,7 @@ def _load_compact_chain_messages( "role": "system", "content": "", }, + internal_event=internal, parent_tool_use_id=None, )) timestamp = row.get("timestamp") @@ -2860,7 +2899,7 @@ def translate_history( but cannot move an already-settled answer's terminal clock. Rich assistant blocks retain the same thinking/commentary/final and semantic tool structure as the live stream. Non-conversational user turns (compact summaries, - slash-command envelopes, local-command stdout) remain hidden. + slash-command envelopes other than /compact, local-command stdout) remain hidden. """ events: list = [] turn_open = False @@ -2877,6 +2916,7 @@ def translate_history( settled_answer_seen = False background_followup = False turn_failed = False + compact_requested = False # Older wrappers could bind a replacement query's browser id to Claude's # late ``[Request interrupted by user]`` record. The marker terminates the # preceding turn; transfer that proven-but-misplaced alias only when the @@ -2983,6 +3023,7 @@ def close_turn( nonlocal ambiguous_final_mid, ambiguous_final_start nonlocal turn_start_ts, settled_answer_seen, turn_failed nonlocal background_followup + nonlocal compact_requested if turn_open: # SessionMessage rows can omit stop_reason. Live must conservatively # treat such text as commentary, but history has the next user/EOF as @@ -3029,6 +3070,7 @@ def close_turn( settled_answer_seen = False background_followup = False turn_failed = False + compact_requested = False for message_index, m in enumerate(messages): advance_terminal_clock = True @@ -3052,6 +3094,9 @@ def close_turn( # human turn. Match the SDK's isMeta filter even when reading # raw compact ancestry, without guessing from prompt text. continue + compact_prompt = manual_compact_prompt(content) + if compact_prompt is not None: + content = compact_prompt if isinstance(content, str): internal_event = (internal_user_events or {}).get(source_uid) if internal_event is not None: @@ -3086,6 +3131,9 @@ def close_turn( turn_open = True current_turn_id = message_uid background_followup = False + compact_requested = compact_prompt is not None + # A command row alone does not prove successful compaction. + turn_failed = compact_requested elif isinstance(content, list): if content and all( isinstance(block, dict) and block.get("type") == "text" @@ -3197,7 +3245,8 @@ def close_turn( current_turn_id = message_uid background_followup = False elif role == "system": - internal_event = (internal_user_events or {}).get(source_uid) + internal_event = ((internal_user_events or {}).get(source_uid) + or getattr(m, "internal_event", None)) if internal_event is not None: # Manual compaction can finish minutes after an answer. Keep # its process timestamp without retiming the settled response. @@ -3210,6 +3259,12 @@ def close_turn( event.ts = timestamp events.append(event) turn_open = True + if compact_requested and event.kind == "compaction" and event.status == "succeeded": + receipt = compact_completion_events(event.item_id, current_turn_id) + for item in receipt: + item.ts = event.ts + events.extend(receipt) + turn_failed = False elif role == "assistant": if not isinstance(content, list): continue @@ -3655,6 +3710,8 @@ def _is_meta_user_text(text: str) -> bool: """Skip non-conversational user turns that would just clutter the history: compact summaries, slash-command envelopes, and local-command stdout/stderr.""" t = text.lstrip() + if manual_compact_prompt(t) is not None: + return False return ( t.startswith("This session is being continued from a previous conversation") or t.startswith("") diff --git a/docs/claude-session-service.md b/docs/claude-session-service.md index b5fad195..ee4a33cf 100644 --- a/docs/claude-session-service.md +++ b/docs/claude-session-service.md @@ -47,6 +47,15 @@ releases them. The service also retains attachment ownership across controller replacement, so closing before replaying a steering echo still removes its files. An unconfirmed close does not authorize deleting live-task attachments. +Steering during an autonomous background continuation requires the owner's +`background_steering` capability. The input still uses native `priority=next`; +only its exact human UUID echo transfers the continuation to a managed turn. +An earlier background Result cannot complete that input. Pending inputs and +their original commit identity survive controller replacement, including when +the first queued input is cancelled. Older service owners reject this operation +before writing to the native stream. Upgrade them using the drain procedure +below; a Wrapper-only upgrade does not add this capability to an existing owner. + This is a cc-remote SDK service, not Claude Code's terminal background mode or the experimental PTY broker. The daily native Claude TUI keeps its existing external-ownership rules; this service does not give it shared input ownership. diff --git a/docs/tui.md b/docs/tui.md index 067d1282..4500ef1f 100644 --- a/docs/tui.md +++ b/docs/tui.md @@ -20,6 +20,11 @@ inside progress prose it closes the outer section. The final answer remains outside the fold, and Enter on it toggles its own turn's section. New output and completion preserve manually chosen folds. History details load on demand. +Active turn status and tool summaries have a left-to-right text sweep that +stops on completion, failure or interruption. Only the foreground color +changes; text, selection and reading position stay intact. Set +`TEXTUAL_ANIMATIONS=none` or `NO_COLOR=1` to disable it. + The compact bottom status keeps mode, execution state and elapsed time visible; long activity/notice text is truncated to the terminal width. Shortcut lists live in `Space h` rather than permanently occupying the bottom of the screen. diff --git a/docs/tui_zh.md b/docs/tui_zh.md index b962086f..6744983b 100644 --- a/docs/tui_zh.md +++ b/docs/tui_zh.md @@ -17,6 +17,10 @@ Enter 切换当前层；在工具正文任意行按 Enter/Esc 只收起内层， 正文按则收起外层。最终回答始终留在外面，在它上面按 Enter 切换对应外层。 新输出和回合完成不会重置手动折叠状态；旧历史的重型详情仍按需读取。 +运行中的回合状态和工具活动摘要有从左向右的文字扫光，完成、失败或中断后停止。 +动画只改变文字前景色，不改变正文、选区或阅读位置；设置 +`TEXTUAL_ANIMATIONS=none` 或 `NO_COLOR=1` 可关闭。 + 底栏只保留简短模式、运行状态和耗时，过长的活动与通知按窗口宽度截断。 完整快捷键列表统一放在 `Space h`，不再常驻占用底部空间。 diff --git a/tests/test_claude_agents.py b/tests/test_claude_agents.py index a817d985..1a6eab45 100644 --- a/tests/test_claude_agents.py +++ b/tests/test_claude_agents.py @@ -185,7 +185,7 @@ def test_local_background_bash_never_creates_clickable_agent_run(): public_agent_run_id("unseen-agent-tool"),) -def test_agent_detail_fails_closed_for_codex_and_work(): +def test_agent_detail_fails_closed_for_missing_codex_and_work(): async def run(): machine, _transport = _mk_machine() command = SimpleNamespace( @@ -199,7 +199,7 @@ async def run(): machine.sessions["session"] = codex rejected_codex = await machine._handle_get_agent_detail(command) assert rejected_codex.authoritative is False - assert "不支持" in rejected_codex.error + assert "尚未生成" in rejected_codex.error work = _mk_ctx("session", "session") work.engine = "claude" diff --git a/tests/test_claude_compaction_history.py b/tests/test_claude_compaction_history.py index 4fc01ae5..0b1c95fb 100644 --- a/tests/test_claude_compaction_history.py +++ b/tests/test_claude_compaction_history.py @@ -1,4 +1,4 @@ -"""Native manual compaction must not manufacture a human conversation turn.""" +"""Manual /compact owns its native command row; automatic summaries stay hidden.""" import json import sqlite3 @@ -121,18 +121,31 @@ def row(uid, role, content, timestamp, parent=None, **extra): messages, timestamps, internal = transcript_compact_snapshot( SID, path=str(source), index_store=store) events = translate_history(messages, 4096, timestamps, internal) - assert [e.prompt for e in events if isinstance(e, UserMsg)] == ["inspect the code"] + assert [e.prompt for e in events if isinstance(e, UserMsg)] == ["inspect the code", "/compact"] ends = [e for e in events if isinstance(e, TurnEnd)] - assert len(ends) == 1 + assert len(ends) == 2 assert ends[0].ts == timestamps["answer"] assert ends[0].result.duration_ms == 5_000 + assert ends[1].checkpoint_id == "command" + assert ends[1].ts == timestamps["boundary"] + assert ends[1].result.duration_ms == 151_000 + assert not ends[1].result.is_error compact = [e for e in events if isinstance(e, ProcessEvent) and e.kind == "compaction"] assert len(compact) == 1 assert compact[0].ts == timestamps["boundary"] + assert compact[0].turn_id == "command" + assert any(isinstance(e, Delta) and e.text == "上下文已压缩，可以继续当前会话。" for e in events) page = transcript_compact_history_page(SID, path=str(source), index_store=store, limit=1) assert page is not None - assert not page.has_more - assert page.oldest_cursor == "human" + assert page.has_more + assert page.oldest_cursor == "command" + newest = translate_history(page.messages, 4096, page.timestamps, page.internal_events) + assert [e.prompt for e in newest if isinstance(e, UserMsg)] == ["/compact"] + assert any(isinstance(e, ProcessEvent) and e.item_id == "boundary" for e in newest) + older = transcript_compact_history_page( + SID, path=str(source), index_store=store, before="command", limit=1) + assert older.oldest_cursor == "human" + assert not older.has_more if iteration == 1: fingerprint = HistorySourceFingerprint.capture(source) for engine in ("claude", "codex"): diff --git a/tests/test_claude_steering.py b/tests/test_claude_steering.py index 2bb9ea12..d19d7f56 100644 --- a/tests/test_claude_steering.py +++ b/tests/test_claude_steering.py @@ -2,6 +2,7 @@ import asyncio from types import SimpleNamespace +from unittest.mock import AsyncMock import pytest from claude_agent_sdk.types import ResultMessage @@ -13,6 +14,8 @@ from cc_remote.wrapper import claude_steer from tests.test_claude_service import environment, released from tests.test_multisession import _mk_ctx, _mk_machine +from tests.test_claude_autocompact import _machine_with_sdk +from cc_remote.config import WrapperConfig def user(uid, text="guide"): @@ -56,6 +59,225 @@ async def interrupt(self): self.interrupts += 1 +ORIGIN = {"kind": "task-notification", "taskId": "background-task"} + + +async def until(predicate): + async with asyncio.timeout(3): + while not predicate(): + await asyncio.sleep(0.001) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("uncertain", [False, True]) +@pytest.mark.parametrize("background_result", ["before", "after", "none"]) +async def test_autonomous_steer_waits_for_exact_echo_then_uses_normal_turn(uncertain, background_result): + sdk = SdkHandle(WrapperConfig()) + machine, transport, ctx = _machine_with_sdk(sdk) + machine._configure_claude_sdk_callbacks(ctx, sdk) + sdk.refresh_goal = AsyncMock(return_value=None) + sdk.client = native = NativeClient() + sdk._start_message_pump() + try: + await native.queue.put({**user("injected", "background task finished"), "origin": ORIGIN}) + await until(lambda: ctx.state == "running") + assert ctx.active_msg_id is None and ctx.translator is None + native.fail_write = uncertain + command = Steer(sid=ctx.key, cmd_id="steer", client_id="browser", msg_id="guide-ui", prompt="guide") + response = await machine._handle_steer(command) + assert response is None if not uncertain else response.code == "steer_outcome_unknown" + assert native.inputs[0][0]["priority"] == "next" + uid = native.inputs[0][0]["uuid"] + await native.queue.put(user("old-tool", [{"type": "tool_result", "tool_use_id": "old", "content": "done"}])) + if background_result == "before": + await native.queue.put({**result(), "origin": ORIGIN}) + await until(lambda: sdk._background_callbacks_pending == 0 and not sdk._steers.background_id) + assert ctx.state == "running" and ctx.turn_task is None + assert not any(e.type in {"turn_end", "turn_steered"} for e in transport.sent) + with pytest.raises(RuntimeError, match="active response"): + await sdk.query("must not overtake accepted guidance") + await native.queue.put(user(uid)) + if background_result == "after": + await native.queue.put({**result(), "origin": ORIGIN}) + await native.queue.put(assistant("guided-reply", [{"type": "text", "text": "guided answer"}])) + await native.queue.put(result()) + await until(lambda: any(e.type == "turn_end" for e in transport.sent)) + await until(lambda: ctx.turn_task is None) + assert [e.msg_id for e in transport.sent if e.type == "turn_steered"] == ["guide-ui"] + assert sum(e.type == "turn_end" for e in transport.sent) == 1 + assert ctx.state == "idle" and not sdk._autonomous_steer_root + assert native.consumers == 1 and native.interrupts == 0 and len(native.inputs) == 1 + finally: + if ctx.turn_task: + ctx.turn_task.cancel() + await asyncio.gather(ctx.turn_task, return_exceptions=True) + await sdk._stop_message_pump() + + +@pytest.mark.asyncio +async def test_background_steering_cancellation_retires_only_pending_input(): + sdk = SdkHandle(WrapperConfig()) + machine, transport, ctx = _machine_with_sdk(sdk) + machine._configure_claude_sdk_callbacks(ctx, sdk) + sdk.client = native = NativeClient() + sdk._start_message_pump() + sdk._steers.capabilities.add("interrupt_cancel_queued_v1") + try: + await native.queue.put({**user("injected"), "origin": ORIGIN}) + await until(lambda: ctx.state == "running") + await machine._handle_steer(Steer(sid=ctx.key, cmd_id="c", client_id="browser", msg_id="guide", prompt="guide")) + uid = native.inputs[0][0]["uuid"] + + async def cancel(request): + assert request == {"subtype": "interrupt", "cancel_queued": True} + await native.queue.put({"type": "command_lifecycle", "state": "cancelled", "command_uuid": uid}) + await native.queue.put({**result(), "origin": ORIGIN}) + + native._send_control_request = cancel + await machine._handle_interrupt(SimpleNamespace(sid=ctx.key)) + await until(lambda: ctx.state == "idle") + assert any(e.type == "error" and e.msg_id == "guide" for e in transport.sent) + assert not any(e.type in {"turn_steered", "turn_end"} for e in transport.sent) + assert not sdk._steers.pending and not sdk._autonomous_steer_root + finally: + if ctx.claude_autonomous_interrupt_task: + ctx.claude_autonomous_interrupt_task.cancel() + await asyncio.gather(ctx.claude_autonomous_interrupt_task, return_exceptions=True) + await sdk._stop_message_pump() + + +@pytest.mark.asyncio +async def test_background_steer_rejects_after_human_terminal_before_consumer_drain(): + sdk = SdkHandle(WrapperConfig()) + sdk.client = native = NativeClient() + sdk._start_message_pump() + try: + await native.queue.put({**user("injected"), "origin": ORIGIN}) + await until(lambda: sdk._steers.background_id) + await sdk.steer("guide", native_id="guide", metadata={"id": "guide"}) + await native.queue.put(user("guide")) + await native.queue.put(result()) + await until(lambda: sdk._autonomous_steer_root.get("adopted") and not sdk._turn_active) + with pytest.raises(ClaudeSteerRejected): + await sdk.steer("too late", native_id="late", metadata={"id": "late"}) + assert len(native.inputs) == 1 + finally: + await sdk._stop_message_pump() + + +@pytest.mark.asyncio +async def test_service_multiple_background_inputs_cancel_one_and_keep_root_commit(tmp_path): + async with environment() as (service, attach): + client = await attach() + worker = service.sessions[client.id] + await worker.client.queue.put({**user("injected"), "origin": ORIGIN}) + await until(lambda: worker.steers.background_id) + attachment = tmp_path / "staged" + attachment.mkdir() + await client.steer("first", native_id="first", metadata={"id": "first-ui", "attachment_dir": str(attachment)}, + turn_id="first-ui", background_id=worker.steers.background_id) + await client.steer("second", native_id="second", metadata={"id": "second-ui", "prompt": "second"}, + turn_id="first-ui") + await worker.client.queue.put({"type": "command_lifecycle", "state": "cancelled", "command_uuid": "first"}) + await until(lambda: "first" not in worker.steers.pending) + assert worker.turn["id"] == "first-ui" and attachment.exists() + await worker.client.queue.put(user("second")) + await worker.client.queue.put(result()) + await until(lambda: worker.terminal_seq is not None) + assert worker.turn["id"] == "first-ui" and not worker.background_turns + await client.call("commit", {"turn_id": "first-ui", "seq": worker.terminal_seq}) + assert worker.turn is None and not attachment.exists() + assert len(worker.client.prompts) == 2 + + +def test_background_target_matches_exact_origin_across_native_field_spellings(): + from cc_remote.claude_steering import PendingSteers + pending = PendingSteers() + pending.annotate({**user("injected"), "origin": ORIGIN}) + identity = pending.background_id + pending.annotate({**result(), "origin": {"kind": "task-notification", "taskId": "unrelated"}}) + assert pending.background_id == identity + pending.annotate({**result(), "origin": {"kind": "task-notification", "task_id": "background-task"}}) + assert pending.background_id is None + + +@pytest.mark.asyncio +@pytest.mark.parametrize("echo_before_detach", [False, True]) +@pytest.mark.parametrize("background_result", [False, True]) +async def test_service_background_steer_recovery_does_not_resubmit(echo_before_detach, background_result): + guide_id = "22222222-2222-4222-8222-222222222222" + metadata = {"id": "guide-ui", "prompt": "guide", "fingerprint": "a" * 64} + async with environment() as (service, attach): + first = await attach() + worker = service.sessions[first.id] + native = worker.client + await native.queue.put({**user("injected"), "origin": ORIGIN}) + await until(lambda: worker.steers.background_id) + target = worker.steers.background_id + await first.steer("guide", native_id=guide_id, metadata=metadata, + turn_id="guide-ui", background_id=target) + assert worker.turn["awaiting_steer"] + if background_result: + await native.queue.put({**result(), "origin": ORIGIN}) + await until(lambda: worker.steers.background_id is None) + assert worker.terminal_seq is None + if echo_before_detach: + await native.queue.put(user(guide_id)) + await until(lambda: not worker.turn.get("awaiting_steer")) + await first.detach() + await released(worker) + sdk = SdkHandle(WrapperConfig(claude_service_socket=first.connection.socket_path)) + sdk.service_metadata = worker.metadata.copy() + sdk.service_defer_events = True + machine, transport, ctx = _machine_with_sdk(sdk) + machine._configure_claude_sdk_callbacks(ctx, sdk) + sdk.refresh_goal = AsyncMock(return_value=None) + # _configure reads this test machine's empty service setting; preserve + # the existing exact owner identity from the original controller. + sdk.service_metadata = worker.metadata.copy() + try: + await sdk.connect(resume_id="native-session", cwd="/tmp") + from cc_remote.wrapper.claude_service import activate + await activate(machine, ctx) + if not echo_before_detach: + assert not any(e.type in {"user_msg", "turn_steered"} for e in transport.sent) + await sdk.steer("guide", native_id=guide_id, metadata=metadata) + await native.queue.put(user(guide_id)) + await native.queue.put(assistant("answer", [{"type": "text", "text": "recovered"}])) + await native.queue.put(result()) + await until(lambda: worker.turn is None) + await until(lambda: ctx.turn_task is None) + assert len(native.prompts) == 1 and native.interrupts == 0 + assert sum(e.type == "turn_steered" for e in transport.sent) == 1 + assert sum(e.type == "turn_end" for e in transport.sent) == 1 + assert worker.background_start is None + assert ctx.state == "idle" + finally: + if ctx.turn_task: + ctx.turn_task.cancel() + await asyncio.gather(ctx.turn_task, return_exceptions=True) + await sdk.detach_for_shutdown() + + +@pytest.mark.asyncio +async def test_service_rejects_stale_background_identity_and_old_capability(): + async with environment() as (service, attach): + client = await attach() + worker = service.sessions[client.id] + await worker.client.queue.put({**user("injected"), "origin": ORIGIN}) + await until(lambda: worker.steers.background_id) + target = worker.steers.background_id + client.description.pop("background_steering") + with pytest.raises(ClaudeSteerRejected): + await client.steer("guide", native_id="g", metadata={"id": "g"}, turn_id="g", background_id=target) + client.description["background_steering"] = True + await worker.client.queue.put({**result(), "origin": ORIGIN}) + await until(lambda: worker.steers.background_id is None) + with pytest.raises(ClaudeSteerRejected): + await client.steer("guide", native_id="g", metadata={"id": "g"}, turn_id="g", background_id=target) + assert not worker.client.prompts and worker.turn is None + + @pytest.mark.asyncio @pytest.mark.parametrize("racing_terminal", [False, True]) @pytest.mark.parametrize("uncertain_write", [False, True]) diff --git a/tests/test_codex_agents.py b/tests/test_codex_agents.py new file mode 100644 index 00000000..7dcd2bdd --- /dev/null +++ b/tests/test_codex_agents.py @@ -0,0 +1,199 @@ +"""Agent details never start a model, cross accounts, or show forked context.""" +import json +from types import SimpleNamespace + +import pytest + +from cc_remote.wrapper.codex_agents import load_detail +from cc_remote.wrapper.codex_stream import CodexStreamTranslator, codex_translate_history +from tests.test_multisession import _mk_ctx, _mk_machine + + +def rollout(home, sid, parent=None, rows=(), archived=False): + root = home / ("archived_sessions" if archived else "sessions") + root.mkdir(parents=True, exist_ok=True) + path = root / f"rollout-{sid}.jsonl" + meta = {"id": sid, "agent_path": f"/root/{sid}"} + if parent: + meta["source"] = {"subagent": {"thread_spawn": {"parent_thread_id": parent}}} + records = [("session_meta", meta), *rows] + path.write_text("".join(json.dumps({"timestamp": "2026-09-18T00:00:00Z", "type": t, "payload": p}) + "\n" + for t, p in records)) + return path + + +def task(turn): + return "event_msg", {"type": "task_started", "turn_id": turn} + + +def answer(text): + return "event_msg", {"type": "agent_message", "message": text, "phase": "final_answer"} + + +def activity(child, call="activity"): + return "event_msg", {"type": "item_completed", "turn_id": "parent-turn", "item": { + "type": "SubAgentActivity", "id": call, "agent_thread_id": child, + "agent_path": "/root/worker", "kind": "interacted"}} + + +def test_child_detail_excludes_inherited_context_and_prompt_and_supports_old_cards(tmp_path): + rollout(tmp_path, "parent", rows=[task("parent-turn"), activity("child")]) + child = rollout(tmp_path, "child", "parent", rows=[ + task("parent-turn"), answer("INHERITED_PARENT"), task("child-turn"), + ("event_msg", {"type": "user_message", "message": "PRIVATE_DELEGATION"}), + ("response_item", {"type": "function_call", "name": "exec_command", "call_id": "tool", + "arguments": '{"cmd":"pwd"}'}), + ("response_item", {"type": "function_call_output", "call_id": "tool", "output": "exit code 0"}), + answer("child answer"), ("event_msg", {"type": "task_complete", "turn_id": "child-turn"}), + ], archived=True) + rows, status, title, revision = load_detail(tmp_path, "parent", "activity", 8000) + direct = load_detail(tmp_path, "parent", "codex-agent:child", 8000) + assert direct[1:] == (status, title, revision) + assert [{k: v for k, v in r.items() if k != "ts"} for r in direct[0]] == [ + {k: v for k, v in r.items() if k != "ts"} for r in rows] + text = json.dumps(rows) + assert "child answer" in text and "tool_use" in text and "tool_result" in text + assert "INHERITED_PARENT" not in text and "PRIVATE_DELEGATION" not in text + assert status == "succeeded" and title == "/root/child" + child.write_text(child.read_text() + json.dumps({"type": "event_msg", "payload": { + "type": "task_started", "turn_id": "next"}}) + "\n") + newer = load_detail(tmp_path, "parent", "codex-agent:child", 8000) + assert newer[1] == "running" and newer[3] != revision + + +def test_only_proven_descendants_in_same_account_can_be_read(tmp_path): + home = tmp_path / "account" + rollout(home, "parent") + rollout(home, "child", "parent") + rollout(home, "nested", "child") + rollout(home, "unrelated") + rollout(tmp_path / "other", "foreign", "parent") + assert load_detail(home, "parent", "codex-agent:nested", 8000)[0] == [] + for target in ("parent", "unrelated", "foreign", "../../other", "missing"): + with pytest.raises(ValueError): + load_detail(home, "parent", f"codex-agent:{target}", 8000) + (home / "sessions/rollout-foreign.jsonl").symlink_to(tmp_path / "other/sessions/rollout-foreign.jsonl") + with pytest.raises(ValueError): + load_detail(home, "parent", "codex-agent:foreign", 8000) + + +def test_live_and_replayed_cards_reference_same_child(tmp_path): + native = {"method": "item/completed", "params": {"threadId": "parent", "turnId": "turn", + "item": {"type": "subAgentActivity", "id": "activity", "kind": "interacted", + "agentThreadId": "child", "agentPath": "/root/worker"}}} + live = CodexStreamTranslator(8000).feed(native)[0] + path = rollout(tmp_path, "parent", rows=[task("turn"), activity("child")]) + replayed = next(e for e in codex_translate_history(str(path), 8000)[0] + if e.type == "process" and e.kind == "agent") + assert live.input == replayed.input == {"agent_run_id": "codex-agent:child"} + + +@pytest.mark.parametrize(("sender", "recipient", "title"), [ + ("child", "parent", "向主代理汇报"), + ("nested", "parent", "向主代理汇报"), + ("nested", "child", "向上级代理汇报"), +]) +def test_ancestor_reports_expand_the_exact_message_without_child_navigation( + tmp_path, sender, recipient, title): + rollout(tmp_path, "parent", rows=[task("parent-turn")]) + rollout(tmp_path, "child", "parent", rows=[task("child-turn")]) + rollout(tmp_path, "sibling", "parent", rows=[task("sibling-turn"), answer("sibling answer"), + ("event_msg", {"type": "task_complete", "turn_id": "sibling-turn"})]) + rows = [ + task("parent-turn"), + ("response_item", {"type": "function_call", "name": "send_message", "call_id": "missing", + "arguments": json.dumps({"message": "INHERITED_REPORT"})}), + ("event_msg", {"type": "thread_settings_applied", "thread_id": sender}), + task("own-turn"), + ("response_item", {"type": "function_call", "name": "send_message", "call_id": "report", + "arguments": json.dumps({"target": "/root", "message": "已核对\n汇报正文"})}), + activity(recipient, "report"), + ("response_item", {"type": "function_call", "name": "send_message", "call_id": "other-report", + "arguments": json.dumps({"message": "另一条消息"})}), + activity(recipient, "other-report"), activity(recipient, "missing"), + activity("grandchild", "spawned"), activity("sibling", "sibling-call"), + ] + rollout(tmp_path, sender, "parent" if sender == "child" else "child", rows=rows) + rollout(tmp_path, "grandchild", sender, rows=[task("grandchild-turn"), answer("nested answer"), + ("event_msg", {"type": "task_complete", "turn_id": "grandchild-turn"})]) + events = load_detail(tmp_path, "parent", f"codex-agent:{sender}", 8000)[0] + cards = {e["item_id"]: e for e in events if e["type"] == "process" and e["kind"] == "agent"} + assert cards["report"]["title"] == title + assert cards["report"]["detail"] == "已核对\n汇报正文" + assert cards["report"]["input"] == {"agent_run_id": None} + assert cards["report"]["status"] == "succeeded" + assert cards["other-report"]["detail"] == "另一条消息" + assert cards["missing"]["input"] == {"agent_run_id": None} + assert cards["missing"]["title"] == ("主代理动态" if recipient == "parent" else "上级代理动态") + assert "detail" not in cards["missing"] + assert "INHERITED_REPORT" not in json.dumps(events) + for call, target, text in [("spawned", "grandchild", "nested answer"), + ("sibling-call", "sibling", "sibling answer")]: + run_id = cards[call]["input"]["agent_run_id"] + assert run_id == f"codex-agent:{target}" + assert text in json.dumps(load_detail(tmp_path, "parent", run_id, 8000)[0]) + + +def test_native_send_input_to_ancestor_displays_its_prompt(tmp_path): + rollout(tmp_path, "parent") + rollout(tmp_path, "child", "parent", rows=[task("own"), + ("event_msg", {"type": "item_completed", "item": { + "type": "CollabAgentToolCall", "id": "native-report", "tool": "sendInput", + "receiver_thread_ids": ["parent"], "sender_thread_id": "child", + "prompt": "原生汇报正文", "status": "completed"}})]) + events = load_detail(tmp_path, "parent", "codex-agent:child", 8000)[0] + card = next(e for e in events if e["type"] == "process" and e["kind"] == "agent") + assert card["title"] == "向主代理汇报" + assert card["detail"] == "原生汇报正文" + assert card["input"] == {"agent_run_id": None} + + +def test_native_child_boundary_survives_parent_rollback_and_multi_agent_cards(tmp_path): + # The current parent need not retain the context originally forked into a + # child. A native child thread-settings boundary excludes that old prefix. + rollout(tmp_path, "parent", rows=[("event_msg", {"type": "item_completed", "item": { + "type": "CollabAgentToolCall", "id": "wait-both", "tool": "wait", + "receiver_thread_ids": ["child", "other"], "status": "completed"}})]) + rollout(tmp_path, "child", "parent", rows=[task("inherited"), answer("old parent"), + ("event_msg", {"type": "thread_settings_applied", "thread_id": "child"}), + task("own"), answer("own answer"), ("event_msg", {"type": "task_complete", "turn_id": "own"})]) + rollout(tmp_path, "other", "parent") + events = load_detail(tmp_path, "parent", "codex-agent:child", 8000)[0] + assert "own answer" in json.dumps(events) and "old parent" not in json.dumps(events) + group = load_detail(tmp_path, "parent", "wait-both", 8000)[0] + assert [e["item_id"] for e in group] == ["codex-agent:child", "codex-agent:other"] + parent_path = tmp_path / "sessions/rollout-parent.jsonl" + cards = [e for e in codex_translate_history(str(parent_path), 8000)[0] + if e.type == "process" and e.kind == "agent"] + assert cards[0].input["receivers"] == ["child", "other"] + + +@pytest.mark.asyncio +async def test_machine_pages_and_invalidates_details_without_engine_access(tmp_path, monkeypatch): + machine, _ = _mk_machine() + monkeypatch.setattr(machine, "_codex_target", lambda sid: (SimpleNamespace(home=tmp_path, id="primary"), sid)) + ctx = _mk_ctx("parent", "parent") + ctx.engine = "codex" + machine.sessions[ctx.key] = ctx + rollout(tmp_path, "parent") + child = rollout(tmp_path, "child", "parent", rows=[task("t"), answer("answer"), + ("response_item", {"type": "function_call", "name": "send_message", "call_id": "report", + "arguments": '{"target":"/root","message":"paged report"}'}), + activity("parent", "report"), + ("event_msg", {"type": "token_count"})]) + cmd = SimpleNamespace(session_id="parent", run_id="codex-agent:child", request_id="r", + client_id="browser", revision=None, detail_revision=None, before=None, limit=1) + detail = await machine._handle_get_agent_detail(cmd) + assert detail.authoritative and detail.has_more + assert detail.to == "browser" + assert len(detail.events) == 1 and detail.events[0]["type"] == "process" + assert detail.events[0]["detail"] == "paged report" + assert detail.events[0]["input"] == {"agent_run_id": None} + cmd.before, cmd.detail_revision = detail.oldest_cursor, detail.detail_revision + page = await machine._handle_get_agent_detail(cmd) + assert page.authoritative and page.events != detail.events + assert len(page.events) == 1 and page.events[0]["type"] == "tool_use" + child.write_text(child.read_text() + json.dumps({"type": "event_msg", "payload": { + "type": "task_complete", "turn_id": "t"}}) + "\n") + stale = await machine._handle_get_agent_detail(cmd) + assert not stale.authoritative and "已更新" in stale.error diff --git a/tests/test_codex_archived_rollout.py b/tests/test_codex_archived_rollout.py index 04663c62..db35700e 100644 --- a/tests/test_codex_archived_rollout.py +++ b/tests/test_codex_archived_rollout.py @@ -3,6 +3,8 @@ import json import sqlite3 +import pytest + from cc_remote.wrapper import codex_sessions @@ -69,6 +71,35 @@ def test_active_rollout_wins_if_both_stores_contain_same_id(tmp_path, monkeypatc assert codex_sessions.codex_session_cwd(session_id) == "/repo/active" +@pytest.mark.parametrize("archived", [False, True]) +@pytest.mark.parametrize("exact_first", [False, True]) +def test_rollout_lookup_requires_exact_thread_identity_regardless_of_order( + tmp_path, monkeypatch, archived, exact_first, +): + exact = _write_catalog_rollout(tmp_path, "child", archived=archived) + _write_catalog_rollout(tmp_path, "grandchild") + # A copied session_id is not the native thread id of a fork. + _write_catalog_rollout(tmp_path, "child-copy", session_id="child") + (tmp_path / "sessions/rollout-empty-child.jsonl").write_text("") + (tmp_path / "sessions/rollout-broken-child.jsonl").write_text("{invalid}\n") + iglob = codex_sessions.glob.iglob + + def ordered_matches(pattern, *, recursive): + # Directory order differs between filesystems. Exercise both orders + # without relying on this machine's native directory enumeration. + return iter(sorted( + iglob(pattern, recursive=recursive), + key=lambda path: path == str(exact), + reverse=exact_first, + )) + + monkeypatch.setattr(codex_sessions.glob, "iglob", ordered_matches) + + assert codex_sessions.codex_rollout_path("child", codex_home=tmp_path) == str(exact) + exact.unlink() + assert codex_sessions.codex_rollout_path("child", codex_home=tmp_path) is None + + def test_codex_session_presence_uses_exact_state_db_and_preserves_uncertainty( tmp_path, ): diff --git a/tests/test_codex_profiles.py b/tests/test_codex_profiles.py index ab1e5769..5b2295d5 100644 --- a/tests/test_codex_profiles.py +++ b/tests/test_codex_profiles.py @@ -682,7 +682,9 @@ def test_profile_rollouts_and_watches_remain_in_their_own_home( ) for path in (primary_rollout, stack_rollout): path.parent.mkdir(parents=True) - path.write_text("", encoding="utf-8") + path.write_text(json.dumps({ + "type": "session_meta", "payload": {"id": "same-native-id"}, + }) + "\n", encoding="utf-8") machine._watch_session("primary@same-native-id") machine._watch_session("stack@same-native-id") diff --git a/tests/test_history.py b/tests/test_history.py index 780e4c0f..b48fda3f 100644 --- a/tests/test_history.py +++ b/tests/test_history.py @@ -5893,13 +5893,14 @@ def test_compacted_claude_main_chain_recovers_precompact_history( assert recovered is not None messages, timestamps = recovered assert [message.uuid for message in messages] == [ - "user-before", "assistant-before", "compact-boundary", "compact-summary", - "compact-command", "user-after", "assistant-after", + "user-before", "assistant-before", "compact-command", "compact-boundary", + "compact-summary", "user-after", "assistant-after", ] events = translate_history(messages, 10_000, timestamps=timestamps) assert [event.prompt for event in events if isinstance(event, UserMsg)] == [ - "before compact", "after compact", + "before compact", "/compact", "after compact", ] + assert not any(event.result.is_error for event in events if isinstance(event, TurnEnd)) assert timestamps["user-before"] < timestamps["user-after"] @@ -6237,12 +6238,20 @@ def test_compacted_claude_page_loads_only_requested_main_chain_turns( "claude-compact", path=str(transcript), before="user-4", limit=2) assert older is not None assert [message.uuid for message in older.messages] == [ - "user-2", "answer-2", "compact-boundary", - "compact-summary", "compact-command", + "compact-command", "compact-boundary", "compact-summary", "user-3", "answer-3", ] assert older.has_more is True - assert older.oldest_cursor == "user-2" + assert older.oldest_cursor == "compact-command" + + earliest = transcript_compact_history_page( + "claude-compact", path=str(transcript), before="compact-command", limit=2) + assert earliest is not None + assert [message.uuid for message in earliest.messages] == [ + "user-1", "answer-1", "user-2", "answer-2", + ] + assert earliest.has_more is False + assert earliest.oldest_cursor == "user-1" def test_compacted_claude_page_uses_only_visible_human_boundaries(tmp_path): diff --git a/tests/test_history_store.py b/tests/test_history_store.py index 2941e162..7ea1ef3f 100644 --- a/tests/test_history_store.py +++ b/tests/test_history_store.py @@ -461,7 +461,7 @@ def test_paged_file_migration_rebuilds_summaries_once_without_removing_assets(tm connection.execute(f"PRAGMA user_version={old_version}") migrated = HistoryIndexStore(tmp_path / "state") with sqlite3.connect(migrated.path) as connection: - assert connection.execute("PRAGMA user_version").fetchone()[0] == 42 + assert connection.execute("PRAGMA user_version").fetchone()[0] == 43 for table in ("history_pages", "history_turn_details"): assert connection.execute( f"SELECT COUNT(*) FROM {table} WHERE engine='codex'" @@ -476,7 +476,8 @@ def test_paged_file_migration_rebuilds_summaries_once_without_removing_assets(tm assert reopened.get_page("codex", "codex", source, before=None, limit=4) is not None -def test_v42_summary_migration_preserves_source_details_and_assets(tmp_path): +@pytest.mark.parametrize("old_version", [41, 42]) +def test_compact_migration_rebuilds_claude_details_and_preserves_assets(tmp_path, old_version): path = tmp_path / "source.jsonl" path.write_text("{}\n") source = HistorySourceFingerprint.capture(path) @@ -486,12 +487,14 @@ def test_v42_summary_migration_preserves_source_details_and_assets(tmp_path): store.put_image_asset(engine, engine, source, engine, "image", "thumbnail", "image/png", 1, 1, engine.encode()) with sqlite3.connect(store.path) as connection: - connection.execute("PRAGMA user_version=41") + connection.execute(f"PRAGMA user_version={old_version}") migrated = HistoryIndexStore(tmp_path / "state") with sqlite3.connect(migrated.path) as connection: - assert connection.execute("PRAGMA user_version").fetchone()[0] == 42 - assert connection.execute("SELECT COUNT(*) FROM history_pages").fetchone()[0] == 0 - assert connection.execute("SELECT COUNT(*) FROM history_turn_details").fetchone()[0] == 2 + assert connection.execute("PRAGMA user_version").fetchone()[0] == 43 + assert connection.execute( + "SELECT engine FROM history_pages").fetchall() == ([] if old_version == 41 else [("codex",)]) + assert connection.execute( + "SELECT engine FROM history_turn_details").fetchall() == [("codex",)] assert connection.execute("SELECT COUNT(*) FROM history_image_assets").fetchone()[0] == 2 assert migrated.put_page("codex", "codex", source, before=None, limit=4, page=_page("rebuilt")) reopened = HistoryIndexStore(tmp_path / "state") @@ -562,7 +565,7 @@ def test_v19_migration_rebuilds_history_and_adds_agent_details(tmp_path): assert migrated.get_page( "session-1", "claude", source, before=None, limit=4) is None with sqlite3.connect(migrated.path) as connection: - assert connection.execute("PRAGMA user_version").fetchone()[0] == 42 + assert connection.execute("PRAGMA user_version").fetchone()[0] == 43 assert connection.execute( "SELECT COUNT(*) FROM history_agent_details").fetchone()[0] == 0 @@ -590,7 +593,7 @@ def test_v20_migration_rebuilds_codex_and_claude_identity_projections(tmp_path): migrated = HistoryIndexStore(state_dir) with sqlite3.connect(migrated.path) as connection: - assert connection.execute("PRAGMA user_version").fetchone()[0] == 42 + assert connection.execute("PRAGMA user_version").fetchone()[0] == 43 for table in ("history_pages", "history_turn_details"): assert connection.execute( f"SELECT COUNT(*) FROM {table} WHERE engine='claude'" @@ -651,7 +654,7 @@ def test_v21_migration_rebuilds_claude_alias_and_codex_process_projections( migrated = HistoryIndexStore(state_dir) with sqlite3.connect(migrated.path) as connection: - assert connection.execute("PRAGMA user_version").fetchone()[0] == 42 + assert connection.execute("PRAGMA user_version").fetchone()[0] == 43 for table in ("history_pages", "history_turn_details"): assert connection.execute( f"SELECT COUNT(*) FROM {table} WHERE engine='claude'" @@ -708,7 +711,7 @@ def test_v22_migration_applies_codex_and_claude_projection_repairs( migrated = HistoryIndexStore(state_dir) with sqlite3.connect(migrated.path) as connection: - assert connection.execute("PRAGMA user_version").fetchone()[0] == 42 + assert connection.execute("PRAGMA user_version").fetchone()[0] == 43 for table in ("history_pages", "history_turn_details"): assert connection.execute( f"SELECT COUNT(*) FROM {table} WHERE engine='claude'" @@ -770,7 +773,7 @@ def test_recent_migration_applies_codex_and_claude_projection_repairs( migrated = HistoryIndexStore(state_dir) with sqlite3.connect(migrated.path) as connection: - assert connection.execute("PRAGMA user_version").fetchone()[0] == 42 + assert connection.execute("PRAGMA user_version").fetchone()[0] == 43 for table in ("history_pages", "history_turn_details"): assert connection.execute( f"SELECT COUNT(*) FROM {table} WHERE engine='claude'" @@ -841,7 +844,7 @@ def test_async_question_migration_preserves_assets( migrated = HistoryIndexStore(state_dir) with sqlite3.connect(migrated.path) as connection: - assert connection.execute("PRAGMA user_version").fetchone()[0] == 42 + assert connection.execute("PRAGMA user_version").fetchone()[0] == 43 for table in ("history_pages", "history_turn_details"): assert connection.execute( f"SELECT COUNT(*) FROM {table} WHERE engine='claude'" @@ -950,7 +953,7 @@ def test_legacy_migration_rebuilds_all_derived_history_rows( migrated = HistoryIndexStore(state_dir) with sqlite3.connect(migrated.path) as connection: - assert connection.execute("PRAGMA user_version").fetchone()[0] == 42 + assert connection.execute("PRAGMA user_version").fetchone()[0] == 43 for table in ( "history_pages", "history_turn_details", @@ -997,7 +1000,7 @@ def test_v10_migration_invalidates_changed_projection_rows(tmp_path): migrated = HistoryIndexStore(state_dir) with sqlite3.connect(migrated.path) as connection: - assert connection.execute("PRAGMA user_version").fetchone()[0] == 42 + assert connection.execute("PRAGMA user_version").fetchone()[0] == 43 for table in ( "history_pages", "history_turn_details", "history_image_assets", ): @@ -1048,7 +1051,7 @@ def test_v11_migration_invalidates_claude_pages_and_adds_compact_index( "claude-session", "claude", source, before=None, limit=4, ) is None with sqlite3.connect(migrated.path) as connection: - assert connection.execute("PRAGMA user_version").fetchone()[0] == 42 + assert connection.execute("PRAGMA user_version").fetchone()[0] == 43 tables = { row[0] for row in connection.execute( "SELECT name FROM sqlite_master WHERE type='table'" @@ -1094,7 +1097,7 @@ def test_recent_migration_invalidates_changed_projection_rows( migrated = HistoryIndexStore(state_dir) with sqlite3.connect(migrated.path) as connection: - assert connection.execute("PRAGMA user_version").fetchone()[0] == 42 + assert connection.execute("PRAGMA user_version").fetchone()[0] == 43 for table in ( "history_pages", "history_turn_details", "history_image_assets", ): @@ -1136,7 +1139,7 @@ def test_owner_and_interrupt_alias_migration_invalidates_both_projections( migrated = HistoryIndexStore(state_dir) with sqlite3.connect(migrated.path) as connection: - assert connection.execute("PRAGMA user_version").fetchone()[0] == 42 + assert connection.execute("PRAGMA user_version").fetchone()[0] == 43 for table in ("history_pages", "history_turn_details"): assert connection.execute( f"SELECT COUNT(*) FROM {table} WHERE engine='claude'" @@ -1200,7 +1203,7 @@ def test_recent_summary_migration_rebuilds_pages_but_preserves_source_assets( migrated = HistoryIndexStore(state_dir) with sqlite3.connect(migrated.path) as connection: - assert connection.execute("PRAGMA user_version").fetchone()[0] == 42 + assert connection.execute("PRAGMA user_version").fetchone()[0] == 43 assert connection.execute( "SELECT COUNT(*) FROM history_pages" ).fetchone()[0] == 0 diff --git a/tests/test_manual_compaction.py b/tests/test_manual_compaction.py new file mode 100644 index 00000000..5a7594fd --- /dev/null +++ b/tests/test_manual_compaction.py @@ -0,0 +1,344 @@ +"""Manual compact lifecycle, service commits and native-turn attribution.""" +from __future__ import annotations + +import asyncio +from types import SimpleNamespace + +import pytest + +from cc_remote.protocol import CompactSession, Delta, Error, Interrupt, ProcessEvent, Query, TurnEnd, UserMsg +from cc_remote.wrapper import claude_service +from cc_remote.wrapper.claude_compaction import manual_compact_prompt +from cc_remote.wrapper.codex_controls import CodexControlStore +from cc_remote.wrapper.codex_handle import CodexHandle +from cc_remote.wrapper.sdk import SdkHandle +from cc_remote.wrapper.stream import replayed_user_message_id, translate_history +from tests.test_claude_autocompact import SESSION_ID, _machine_with_sdk +from tests.test_claude_compaction_flow import boundary, result, status +from tests.test_claude_service import environment, released +from tests.test_codex_spontaneous_stream import _notification +from tests.test_multisession import _mk_ctx, _mk_machine + + +def test_only_compact_command_is_visible_and_stdout_cannot_claim_its_identity(): + from claude_agent_sdk.types import TextBlock, UserMessage + + messages = [SimpleNamespace(type="user", uuid=str(index), message={ + "role": "user", "content": content, + }) for index, content in enumerate([ + "/context", + "local command", + "/compactkeep the plan", + "Compacted", + "/model/compact", + ])] + events = translate_history(messages, 4096) + assert [e.prompt for e in events if isinstance(e, UserMsg)] == ["/compact keep the plan"] + assert all(e.result.is_error for e in events if isinstance(e, TurnEnd)) + stdout = UserMessage(content=[TextBlock(text=messages[3].message["content"])], uuid=SESSION_ID) + assert replayed_user_message_id(stdout) is None + command = UserMessage(content=[TextBlock(text=messages[2].message["content"])], uuid=SESSION_ID) + assert replayed_user_message_id(command) == SESSION_ID + assert manual_compact_prompt([ + {"type": "text", "text": "/compact"}, {"type": "image", "source": {}}, + ]) is None + + +@pytest.mark.parametrize("outcome", ["success", "error", "interrupted", "missing-boundary"]) +def test_manual_claude_compact_commits_service_and_next_message_is_accepted(outcome): + async def run(): + async with environment() as (service, attach): + remote = await attach(session_id=SESSION_ID) + worker = service.sessions[remote.id] + handle = SdkHandle(SimpleNamespace(turn_reader_queue_cap=4)) + handle.client = remote + handle.effort = handle.applied_effort = "max" + handle.applied_auto_compact_mode = handle.auto_compact_mode + handle.applied_auto_compact_threshold_tokens = handle.auto_compact_threshold_tokens + handle._start_message_pump() + machine, transport, ctx = _machine_with_sdk(handle) + + async def context(_cmd, _action): + return ctx + + machine._claude_code_context = context + try: + await machine._handle_compact_session(CompactSession( + session_id=SESSION_ID, engine="claude")) + task = ctx.turn_task + async with asyncio.timeout(2): + while not any(isinstance(e, UserMsg) for e in transport.sent): + await asyncio.sleep(0) + assert worker.client.prompts == ["/compact"] + assert ctx.state == "running" + assert [e.prompt for e in transport.sent if isinstance(e, UserMsg)] == ["/compact"] + rejection = await machine._handle_query(Query(sid=SESSION_ID, prompt="too early", msg_id="early")) + assert isinstance(rejection, Error) + await worker.client.queue.put(status()) + if outcome == "success": + await worker.client.queue.put(boundary()) + if outcome == "interrupted": + async def interrupt(): + worker.client.interrupts += 1 + + worker.client.interrupt = interrupt + await machine._handle_interrupt(Interrupt(sid=SESSION_ID)) + assert worker.client.interrupts == 1 + assert worker.turn is not None + assert not task.done() + assert isinstance(await machine._handle_query(Query( + sid=SESSION_ID, prompt="during drain", msg_id="draining")), Error) + failed = outcome in {"error", "interrupted"} + terminal = {**result(), "is_error": failed, + "subtype": "error_during_execution" if failed else "success"} + await worker.client.queue.put(terminal) + await asyncio.wait_for(task, 2) + assert ctx.state == "idle" + assert worker.turn is None + ends = [e for e in transport.sent if isinstance(e, TurnEnd)] + assert len(ends) == 1 and ends[0].result.is_error == (outcome != "success") + receipt = [e for e in transport.sent if isinstance(e, Delta) + and e.text == "上下文已压缩，可以继续当前会话。"] + assert bool(receipt) == (outcome == "success") + handle.next_turn_id = "next-message" + await handle.query("continue") + assert worker.client.prompts == ["/compact", "continue"] + await worker.client.queue.put(result()) + async for message in handle.receive_response(): + await handle.ack_service_message(message, turn_id="next-message") + assert worker.turn is None + finally: + await handle._stop_message_pump() + + asyncio.run(run()) + + +def test_legacy_uncommitted_compact_recovers_without_resubmitting_stale_prompt(): + async def run(): + async with environment() as (service, attach): + original = await attach(session_id=SESSION_ID) + original.next_turn = {"id": "compact-legacy", "prompt": "previous unrelated question"} + await original.query("/compact") + worker = service.sessions[original.id] + await worker.client.queue.put(status()) + await worker.client.queue.put(boundary()) + await worker.client.queue.put(result()) + async with asyncio.timeout(2): + while worker.terminal_seq is None: + await asyncio.sleep(0) + await original.detach() + await released(worker) + remote = await attach(session_id=SESSION_ID) + remote.ready.clear() + assert remote.id == original.id + handle = SdkHandle(SimpleNamespace(turn_reader_queue_cap=4)) + handle.client = remote + handle.effort = handle.applied_effort = "max" + handle.applied_auto_compact_mode = handle.auto_compact_mode + handle.applied_auto_compact_threshold_tokens = handle.auto_compact_threshold_tokens + handle.service_recovery = remote.recovery + handle._start_message_pump() + handle._turn_root_id = handle._turn_origin_id = remote.recovery["id"] + handle._turn_active = True + handle._message_route_owner = "managed" + machine, transport, ctx = _machine_with_sdk(handle) + try: + await claude_service.activate(machine, ctx) + await asyncio.wait_for(ctx.turn_task, 2) + assert ctx.state == "idle" and worker.turn is None + assert worker.client.prompts == ["/compact"] + assert [e.prompt for e in transport.sent if isinstance(e, UserMsg)] == ["/compact"] + assert any(isinstance(e, Delta) and "上下文已压缩" in e.text for e in transport.sent) + handle.next_turn_id = "next-after-recovery" + await handle.query("continue") + assert worker.client.prompts == ["/compact", "continue"] + await worker.client.queue.put(result()) + async for message in handle.receive_response(): + await handle.ack_service_message(message, turn_id="next-after-recovery") + assert worker.turn is None + finally: + await handle._stop_message_pump() + + asyncio.run(run()) + + +@pytest.mark.parametrize("failed", [False, True]) +def test_internal_claude_compact_commits_success_and_error_without_human_turn(failed): + async def run(): + async with environment() as (service, attach): + remote = await attach(session_id=SESSION_ID) + worker = service.sessions[remote.id] + handle = SdkHandle(SimpleNamespace(turn_reader_queue_cap=4)) + handle.client = remote + handle._start_message_pump() + machine, transport, ctx = _machine_with_sdk(handle) + try: + task = asyncio.create_task(machine._compact_managed_claude_context(ctx, reason="test")) + async with asyncio.timeout(2): + while not worker.client.prompts: + await asyncio.sleep(0) + await worker.client.queue.put(status()) + if not failed: + await worker.client.queue.put(boundary()) + await worker.client.queue.put({**result(), "is_error": failed}) + if failed: + with pytest.raises(RuntimeError, match="compact failed"): + await asyncio.wait_for(task, 2) + else: + await asyncio.wait_for(task, 2) + assert worker.turn is None + assert not any(isinstance(e, (UserMsg, TurnEnd)) for e in transport.sent) + assert not ctx.claude_write_active + finally: + await handle._stop_message_pump() + + asyncio.run(run()) + + +@pytest.mark.parametrize("early", [False, True]) +@pytest.mark.parametrize("outcome", ["completed", "interrupted", "failed"]) +def test_codex_manual_compact_keeps_native_lifecycle_and_command_after_reload(early, outcome): + async def run(): + machine, transport = _mk_machine() + ctx = _mk_ctx("thread-spontaneous", "thread-spontaneous") + ctx.engine = "codex" + handle = CodexHandle(machine.cfg) + handle.thread_id = ctx.session_id + handle.proc = SimpleNamespace(returncode=None) + ctx.sdk = handle + machine.sessions[ctx.key] = ctx + handle.turn_lifecycle_callback = lambda phase, tid: machine._on_codex_turn_lifecycle(ctx, phase, tid) + turn = "manual-compact" + + async def start(): + await handle._dispatch(_notification("turn/started", turn, turn={"id": turn})) + await handle._dispatch(_notification("item/started", turn, + item={"id": "compact-item", "type": "contextCompaction"})) + + async def request(method, params): + assert method == "thread/compact/start" + assert params == {"threadId": ctx.session_id} + if early: + await start() + await asyncio.sleep(0) + return {} + + handle._request = request + await handle.compact_thread() + if not early: + await start() + async with asyncio.timeout(2): + while not any(isinstance(e, UserMsg) and e.prompt == "/compact" for e in transport.sent): + await asyncio.sleep(0) + assert ctx.state == "running" + if outcome == "interrupted": + async def interrupt_request(method, params): + assert method == "turn/interrupt" + assert params["turnId"] == turn + return {} + + handle._request = interrupt_request + await handle.interrupt() + await handle._dispatch(_notification("item/completed", turn, + item={"id": "compact-item", "type": "contextCompaction"})) + assert ctx.state == "running" + task = ctx.codex_spontaneous_task + await handle._dispatch(_notification("turn/completed", turn, + turn={"id": turn, "status": outcome, "durationMs": 20})) + await asyncio.wait_for(task, 2) + assert ctx.state == "idle" and not handle.turn_active + ends = [e for e in transport.sent if isinstance(e, TurnEnd)] + assert len(ends) == 1 and ends[0].result.is_error == (outcome != "completed") + assert any(isinstance(e, Delta) and "上下文已压缩" in e.text for e in transport.sent) == (outcome == "completed") + machine._codex_controls = CodexControlStore(machine.cfg.state_dir) + recovered = await machine._recover_official_codex_users(ctx.key, (turn,)) + assert recovered[turn].prompt == "/compact" + # A later automatic compaction must never inherit the manual command. + assert not await handle.manual_compact_for_turn("later-turn") + assert any(isinstance(e, ProcessEvent) and e.phase == "start" for e in transport.sent) + + asyncio.run(run()) + + +@pytest.mark.parametrize("early", [False, True]) +def test_rejected_codex_compact_does_not_claim_an_automatic_turn(early): + async def run(): + machine, _ = _mk_machine() + handle = CodexHandle(machine.cfg) + handle.thread_id = "thread-spontaneous" + handle.proc = SimpleNamespace(returncode=None) + + async def request(_method, _params): + if early: + await handle._dispatch(_notification( + "turn/started", "automatic", turn={"id": "automatic"})) + raise RuntimeError("rejected") + + handle._request = request + with pytest.raises(RuntimeError, match="rejected"): + await handle.compact_thread() + if not early: + await handle._dispatch(_notification( + "turn/started", "automatic", turn={"id": "automatic"})) + assert not await handle.manual_compact_for_turn("automatic") + + asyncio.run(run()) + + +def test_codex_compact_attribution_does_not_cross_connection_generation(): + async def run(): + machine, _ = _mk_machine() + handle = CodexHandle(machine.cfg) + handle.thread_id = "thread-spontaneous" + handle.proc = SimpleNamespace(returncode=None) + started = asyncio.Event() + response = asyncio.Event() + + async def request(_method, _params): + await handle._dispatch(_notification("turn/started", "compact", turn={"id": "compact"})) + started.set() + await response.wait() + return {} + + handle._request = request + pending = asyncio.create_task(handle.compact_thread()) + await started.wait() + claim = asyncio.create_task(handle.manual_compact_for_turn("compact")) + await asyncio.sleep(0) + handle._generation += 1 + response.set() + await pending + assert not await claim + + asyncio.run(run()) + + +def test_codex_compact_history_is_account_scoped_and_survives_control_updates(tmp_path): + from cc_remote.wrapper.codex_history import CodexOfficialHistory + from tests.test_codex_history import _turn + + async def run(): + machine, _ = _mk_machine() + store = machine._codex_controls = CodexControlStore(tmp_path) + sid = "primary@thread-compact" + store.remember_compaction(sid, "native-compact") + store.update(sid, approval_policy="never", permission_profile=None, web_search="disabled") + store.set_cwd_override(sid, "/tmp/project") + store.set_context(sid, 200_000, 300_000) + machine._codex_controls = CodexControlStore(tmp_path) + assert machine._manual_codex_compact_users("other@thread-compact", ("native-compact",)) == {} + + async def rpc(method, _params, cwd=None): + assert method == "thread/turns/list" + return {"data": [_turn("native-compact", [ + {"type": "contextCompaction", "id": "native-boundary"}, + ])], "nextCursor": None} + + history = CodexOfficialHistory(4096, rpc=rpc, recover_users=machine._recover_official_codex_users) + page = await history.summary_page(sid, before=None, limit=1) + assert len(page.turns) == 1 + turn = page.turns[0] + assert turn["prompt"] == "/compact" and turn["done"] and not turn.get("error") + assert any(block.get("text") == "上下文已压缩，可以继续当前会话。" for block in turn["blocks"]) + + asyncio.run(run()) diff --git a/tests/test_tui_shimmer.py b/tests/test_tui_shimmer.py new file mode 100644 index 00000000..13d6a046 --- /dev/null +++ b/tests/test_tui_shimmer.py @@ -0,0 +1,149 @@ +"""Activity animation must remain paint-only and follow real turn state.""" + +from types import SimpleNamespace + +import pytest +from rich.console import Console +from rich.style import Style +from rich.text import Text +from textual.widgets.text_area import Selection + +from cc_remote.tui_app import Composer, Transcript, WorkspaceApp, WorkspaceClient +from cc_remote.tui_chrome import activity_sweep +from cc_remote.tui_state import Block + + +def test_sweep_moves_across_cjk_cells_without_changing_text_or_background(): + value = "[正在处理 1m 24s · Read ×2]" + console = Console(color_system="truecolor") + peaks = [] + for phase in (0.25, 0.5, 0.75): + text = Text(value, style="on #123456") + activity_sweep(text, 1, len(value) - 1, phase, + Style(color="#707070", bgcolor="#000000"), + Style(color="#f0f0f0", bgcolor="#ffffff")) + assert text.plain == value + styles = [text.get_style_at_offset(console, i) for i in range(len(text))] + assert all(s.bgcolor.name == "#123456" for s in styles) + assert styles[0].color is None and styles[-1].color is None + levels = [s.color.get_truecolor().red for s in styles[1:-1]] + peaks.append(levels.index(max(levels))) + assert peaks[0] < peaks[1] < peaks[2] + + +def active_client(): + client = WorkspaceClient("ws://localhost:8766/ws", "", "", "codex", "s") + view = client.workspace.view("s") + view.event(dict(type="user_msg", msg_id="t", prompt="Check the project", ts=1000)) + view.event(dict(type="delta", message_id="progress", turn_id="t", + channel="commentary", text="Reading project files.", ts=1001)) + view.event(dict(type="tool_use", tool_use_id="read", turn_id="t", + tool="Read", input={"file_path": "README.md"}, ts=1001)) + app = WorkspaceApp(client, connect=False) + app.animation_level = "full" + app.no_color = False + return client, view, app + + +def visible_activity(reader): + first = int(reader.scroll_y) + rows = reader.wrapped_document._offset_to_line_info[ + first:first + reader.scrollable_content_region.height + ] + return tuple( + tuple(reader.render_line(y)) for y, (row, _) in enumerate(rows) + if row in reader.activity_ranges + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("width", [44, 100]) +@pytest.mark.parametrize("theme", ["textual-dark", "textual-light"]) +async def test_timer_sweeps_visible_rows_without_replacing_or_moving_text(monkeypatch, width, theme): + clock = [0.3] + monkeypatch.setattr("cc_remote.tui_app.time", SimpleNamespace( + time=lambda: 1001, monotonic=lambda: clock[0], + )) + client, view, app = active_client() + app.theme = theme + async with app.run_test(size=(width, 35)) as pilot: + reader = app.query_one(Transcript) + app.query_one(Composer).focus() + app.stop_following() + row = reader.text[:reader.text.index("Reading project files.")].count("\n") + reader.selection = Selection((row, 0), (row, 7)) + reader.scroll_home(animate=False, immediate=True) + app.remember() + await pilot.pause() + initial = visible_activity(reader) + assert initial + user_style = reader.get_line(0).get_style_at_offset(app.console, 14) + assert user_style.bgcolor.name == "#202a36" + assert min(user_style.color.get_truecolor()) >= 160 + retained = (reader.text, reader.selection, reader.selected_text, reader.scroll_offset) + mutations = [] + monkeypatch.setattr(reader, "load_text", lambda *_: mutations.append("load")) + monkeypatch.setattr(reader, "replace", lambda *_, **__: mutations.append("replace")) + clock[0] = 1.4 + await pilot.pause(0.2) + assert visible_activity(reader) != initial + assert (reader.text, reader.selection, reader.selected_text, reader.scroll_offset) == retained + assert not mutations and not client._outbox and not reader.history.undo_stack + assert view.presentation.turns["t"].status == "running" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("subtype,is_error", [ + ("success", False), ("error_during_execution", True), ("error", True), +]) +async def test_terminal_stops_sweep_even_with_stale_running_tool(subtype, is_error): + client, view, app = active_client() + async with app.run_test() as pilot: + reader = app.query_one(Transcript) + assert reader.activity_ranges + view.event(dict(type="turn_end", turn_id="t", ts=1002, + result={"subtype": subtype, "is_error": is_error})) + app.paint() + await pilot.pause() + assert not reader.activity_ranges + assert reader.activity_frame is None + assert view.presentation.turns["t"].status != "running" + assert not client._outbox + + +@pytest.mark.asyncio +@pytest.mark.parametrize("setting", ["animation_level", "no_color"]) +async def test_disabled_animation_clears_paint_and_does_not_refresh(monkeypatch, setting): + client, _, app = active_client() + async with app.run_test() as pilot: + reader = app.query_one(Transcript) + assert reader.activity_frame is not None + setattr(app, setting, "none" if setting == "animation_level" else True) + reader.refresh_activity() + assert reader.activity_frame is None + await pilot.pause() + refreshes = [] + monkeypatch.setattr(reader, "refresh_lines", lambda *args: refreshes.append(args)) + reader.refresh_activity() + assert not refreshes + + +@pytest.mark.asyncio +async def test_offscreen_activity_and_session_switch_do_not_animate_unrelated_text(monkeypatch): + client, view, app = active_client() + view.put(Block("answer", "assistant", "Processing is a word.\n" * 80, "t", "final")) + async with app.run_test() as pilot: + reader = app.query_one(Transcript) + reader.scroll_end(animate=False, immediate=True) + await pilot.pause() + assert not visible_activity(reader) + refreshes = [] + monkeypatch.setattr(reader, "refresh_lines", lambda *args: refreshes.append(args)) + reader.activity_frame = -1 + reader.refresh_activity() + assert not refreshes + client.workspace.view("other").put(Block("other", "assistant", "Processing is not a status.")) + client.attached_sid = "other" + app.paint() + assert not reader.activity_ranges + assert not client._outbox diff --git a/web/src/App.tsx b/web/src/App.tsx index 0a3dbd79..9e0211c7 100644 --- a/web/src/App.tsx +++ b/web/src/App.tsx @@ -5200,14 +5200,14 @@ export default function App() { previewFileForSid(activeBtwSid, file, line); const openAgentDetail = (runId: string, title?: string) => { - if (!focusedSid || focusedEngine !== "claude" || space !== "code" + if (!focusedSid || space !== "code" || !rt.historyRevision) return; closeViewer(); const agent = rt.backgroundProcesses.find((block) => block.item_id === runId) ?? rt.turns.flatMap((turn) => turn.blocks).find((block) => block.kind === "process" && block.item_id === runId); setAgentPanel({ sid: focusedSid, revision: rt.historyRevision, - runId, title: title || "协作代理", + runId, title: title || "协作代理", engine: focusedEngine, status: agent?.kind === "process" ? agent.status : undefined }); }; const previewAgentFile = (file: string, line?: number) => { @@ -5822,7 +5822,7 @@ export default function App() { turnUsage={rt.turnUsage} activeTurnId={activeTurnId} ambiguousActiveTurnIds={ambiguousActiveTurnIds} - onOpenAgent={focusedEngine === "claude" && space === "code" + onOpenAgent={space === "code" ? openAgentDetail : undefined} onFork={!historyView.recovering && !archivedBrowse && space === "code" diff --git a/web/src/components/AgentDetailController.tsx b/web/src/components/AgentDetailController.tsx index d983c86f..fde64386 100644 --- a/web/src/components/AgentDetailController.tsx +++ b/web/src/components/AgentDetailController.tsx @@ -5,7 +5,7 @@ import { emptyAgentRun, type AgentDetailPanelState, } from "../agent-detail"; -import type { AgentDetail, ProcessStatus } from "../protocol"; +import type { AgentDetail, Engine, ProcessStatus } from "../protocol"; import type { RelayWs } from "../ws"; import { uuid } from "../util"; import { AgentDetailPanel } from "./AgentDetailPanel"; @@ -17,6 +17,7 @@ export interface AgentDetailSelection { runId: string; title: string; status?: ProcessStatus; + engine?: Engine; } export function AgentDetailController({ selection, ws, onListen, onClose, @@ -134,7 +135,7 @@ export function AgentDetailController({ selection, ws, onListen, onClose, return () => clearTimeout(timer); }, [request, run]); if (!run) return null; - return 1} + return 1} onBack={() => setPanel((value) => ({ ...value, stack: value.stack.slice(0, -1), }))} diff --git a/web/src/components/AgentDetailPanel.tsx b/web/src/components/AgentDetailPanel.tsx index 8e89d2cb..f5781736 100644 --- a/web/src/components/AgentDetailPanel.tsx +++ b/web/src/components/AgentDetailPanel.tsx @@ -1,6 +1,7 @@ import type { AgentDetailRun } from "../agent-detail"; +import type { Engine } from "../protocol"; import { finalTextBlocks, presentableProcessBlocks } from "../process-blocks"; -import { ClaudeWorking, Icon } from "../icons"; +import { ClaudeWorking, EngineIcon, Icon } from "../icons"; import { MessageBlock } from "./MessageBlock"; import { ProcessTimeline } from "./ProcessTimeline"; import { PanelResizer } from "./PanelResizer"; @@ -13,9 +14,10 @@ function statusLabel(run: AgentDetailRun): string { return "状态未知"; } -export function AgentDetailPanel({ run, canGoBack, onBack, onClose, onRetry, +export function AgentDetailPanel({ run, engine = "claude", canGoBack, onBack, onClose, onRetry, onLoadEarlier, onOpenAgent, onOpenFile }: { run: AgentDetailRun; + engine?: Engine; canGoBack: boolean; onBack: () => void; onClose: () => void; @@ -24,7 +26,7 @@ export function AgentDetailPanel({ run, canGoBack, onBack, onClose, onRetry, onOpenAgent: (runId: string, title?: string) => void; onOpenFile?: (path: string, line?: number) => void; }) { - const process = presentableProcessBlocks(run.blocks, "claude"); + const process = presentableProcessBlocks(run.blocks, engine); const final = finalTextBlocks(run.blocks); const done = !["running", "pending"].includes(run.status); return ( @@ -66,7 +68,7 @@ export function AgentDetailPanel({ run, canGoBack, onBack, onClose, onRetry, )} {process.length > 0 && ( )} {final.map((block) => ( @@ -74,7 +76,7 @@ export function AgentDetailPanel({ run, canGoBack, onBack, onClose, onRetry, done={block.done} onOpenFile={onOpenFile} /> ))} {!done &&
- + {engine === "claude" ? : } 子代理处理中
} {!run.loading && !run.error && run.blocks.length === 0 && ( diff --git a/web/src/components/Composer.tsx b/web/src/components/Composer.tsx index dfb2debe..96941d19 100644 --- a/web/src/components/Composer.tsx +++ b/web/src/components/Composer.tsx @@ -574,7 +574,6 @@ export function Composer(p: Props) { case "compact": if (args.trim()) { flash("/compact 不接受参数"); return; } p.onCompact?.(); - flash("正在启动原生上下文压缩…"); break; case "rollback": flash("Codex Rollback 暂未开放"); break; // /btw: open an ephemeral side-fork panel (both engines). diff --git a/web/src/components/ProcessTimeline.tsx b/web/src/components/ProcessTimeline.tsx index 61550f05..d6df7cb3 100644 --- a/web/src/components/ProcessTimeline.tsx +++ b/web/src/components/ProcessTimeline.tsx @@ -71,6 +71,53 @@ function statusIcon(status: ProcessBlock["status"], done: boolean) { return ; } +function useProcessInteraction( + onStart?: () => number, + onEnd?: (token: number, followOutput?: boolean) => void, +) { + const active = useRef(new Map()); + const pending = useRef(new Map()); + const frame = useRef(null); + useEffect(() => () => { + if (frame.current !== null) window.cancelAnimationFrame(frame.current); + frame.current = null; + for (const token of active.current.values()) onEnd?.(token, false); + for (const token of pending.current.values()) onEnd?.(token, false); + active.current.clear(); + pending.current.clear(); + }, [onEnd]); + const cancel = (pointerId: number) => { + const token = active.current.get(pointerId) ?? pending.current.get(pointerId); + if (token == null) return; + active.current.delete(pointerId); + pending.current.delete(pointerId); + onEnd?.(token, false); + }; + return { + cancel, + begin(pointerId: number) { + cancel(pointerId); + const token = onStart?.(); + if (token != null) active.current.set(pointerId, token); + }, + release(pointerId: number) { + const token = active.current.get(pointerId); + if (token == null) return; + active.current.delete(pointerId); + pending.current.set(pointerId, token); + if (frame.current !== null) window.cancelAnimationFrame(frame.current); + // Keep the viewport frozen through native click and ResizeObserver. + frame.current = window.requestAnimationFrame(() => { + frame.current = window.requestAnimationFrame(() => { + frame.current = null; + for (const token of pending.current.values()) onEnd?.(token); + pending.current.clear(); + }); + }); + }, + }; +} + function ProcessDisclosure({ className, summary, children, openOverride, onOpenChange, onInteractionStart, onInteractionEnd }: { className: string; @@ -84,52 +131,11 @@ function ProcessDisclosure({ className, summary, children, openOverride, const [uncontrolledOpen, setUncontrolledOpen] = useState(false); const open = openOverride ?? uncontrolledOpen; const tapGuard = useRef(new PointerTapGuard()); - const interactionTokens = useRef(new Map()); - const pendingInteractionTokens = useRef(new Map()); - const releaseInteractionFrame = useRef(null); - useEffect(() => () => { - if (releaseInteractionFrame.current !== null) { - window.cancelAnimationFrame(releaseInteractionFrame.current); - } - for (const token of interactionTokens.current.values()) { - onInteractionEnd?.(token, false); - } - for (const token of pendingInteractionTokens.current.values()) { - onInteractionEnd?.(token, false); - } - interactionTokens.current.clear(); - pendingInteractionTokens.current.clear(); - }, [onInteractionEnd]); + const interaction = useProcessInteraction(onInteractionStart, onInteractionEnd); const setOpen = (next: boolean) => { setUncontrolledOpen(next); onOpenChange?.(next); }; - const releaseInteraction = (pointerId: number) => { - const token = interactionTokens.current.get(pointerId); - if (token == null) return; - interactionTokens.current.delete(pointerId); - pendingInteractionTokens.current.set(pointerId, token); - if (releaseInteractionFrame.current !== null) { - window.cancelAnimationFrame(releaseInteractionFrame.current); - } - releaseInteractionFrame.current = window.requestAnimationFrame(() => { - releaseInteractionFrame.current = window.requestAnimationFrame(() => { - releaseInteractionFrame.current = null; - for (const token of pendingInteractionTokens.current.values()) { - onInteractionEnd?.(token); - } - pendingInteractionTokens.current.clear(); - }); - }); - }; - const cancelInteraction = (pointerId: number) => { - const token = interactionTokens.current.get(pointerId) - ?? pendingInteractionTokens.current.get(pointerId); - if (token == null) return; - interactionTokens.current.delete(pointerId); - pendingInteractionTokens.current.delete(pointerId); - onInteractionEnd?.(token, false); - }; return (
{ if (tapGuard.current.pointerMove( event.pointerId, event.clientX, event.clientY, )) { - cancelInteraction(event.pointerId); + interaction.cancel(event.pointerId); releaseDraggedPointer( event.currentTarget, event.pointerId, event.pointerType); } }} onPointerUp={(event) => { tapGuard.current.pointerUp(event.pointerId); - releaseInteraction(event.pointerId); + interaction.release(event.pointerId); }} onPointerCancel={(event) => { - cancelInteraction(event.pointerId); + interaction.cancel(event.pointerId); cancelDraggedPointer( tapGuard.current, event.currentTarget, event.pointerId, event.pointerType); @@ -415,11 +419,13 @@ export function ProcessActivity({ block, onOpenFile, imageAssets, onLoadImage, ); } - if (block.processKind === "agent" && onOpenAgent) { + const inlineAgent = block.processKind === "agent" && block.input?.agent_run_id === null; + if (block.processKind === "agent" && !inlineAgent && onOpenAgent) { return ( ); } + const displayInput = { ...block.input }; + if (block.processKind === "agent") delete displayInput.agent_run_id; const imageView = block.tool?.toLowerCase().replaceAll("_", "") === "viewimage"; const imagePath = imageView ? filePathsFromInput(block.input)[0] ?? "" @@ -459,7 +467,7 @@ export function ProcessActivity({ block, onOpenFile, imageAssets, onLoadImage, const hasBody = !!(block.summary || block.detail || block.output || block.diff || block.progress || block.explanation || block.command || block.cwd || block.plan?.length || block.exit_code != null || block.duration_ms != null - || (block.input && Object.keys(block.input).length)); + || Object.keys(displayInput).length); const body = ( <> {block.progress &&
{block.progress}
} @@ -478,7 +486,7 @@ export function ProcessActivity({ block, onOpenFile, imageAssets, onLoadImage, {block.cwd &&
{block.cwd}
} {block.summary && !imageView &&
{block.summary}
} - {block.detail && } + {block.detail && } {onOpenFile && filePaths.map((filePath) => (