From 0aff1ea783fa4b489298a2832d87a855861d5514 Mon Sep 17 00:00:00 2001 From: Ace Data Cloud Dev Date: Sun, 4 Oct 2026 23:06:48 +0800 Subject: [PATCH 1/5] fix: make local coding sessions stop, stream, and resume reliably --- README.md | 23 ++++- coding_bridge/capabilities.py | 47 ++++----- coding_bridge/connection.py | 18 +++- coding_bridge/history.py | 29 +++++- coding_bridge/providers/claude.py | 126 ++++++++++++++++------- coding_bridge/providers/codex.py | 160 +++++++++++++++++++----------- coding_bridge/service_cli.py | 16 ++- coding_bridge/session.py | 6 ++ pyproject.toml | 1 + tests/conftest.py | 8 -- tests/test_capabilities.py | 14 ++- tests/test_claude_entrypoint.py | 33 ++++++ tests/test_codex_process.py | 116 ++++++++++++++++++++++ tests/test_connection.py | 22 ++++ tests/test_edit_fork.py | 30 ++++++ tests/test_history.py | 18 ++++ tests/test_service.py | 12 +++ tests/test_session.py | 15 +++ tests/test_streaming.py | 79 ++++++++------- 19 files changed, 600 insertions(+), 173 deletions(-) create mode 100644 tests/test_codex_process.py diff --git a/README.md b/README.md index a2958a7..41a68f6 100644 --- a/README.md +++ b/README.md @@ -12,7 +12,8 @@ relay only carries JSON messages. ## Install Requires **Python 3.10+** and a signed-in -[Claude Code](https://docs.claude.com/en/docs/claude-code) CLI. To use the +[Claude Code](https://docs.claude.com/en/docs/claude-code) CLI. The Claude Agent SDK +bundled CLI is detected too; a separate global installation is optional. To use the **Codex** provider instead, install the [Codex CLI](https://github.com/openai/codex) and run `codex login`. @@ -147,9 +148,8 @@ That shape is the whole security model: - **Execution is local.** Your repo, shell, and MCP servers only ever touch this machine. - **Outbound only.** No listening ports, so nothing to expose or firewall. -- **Per-tool approval — in the modes that ask.** In `default` mode every tool - the agent wants to run is relayed to you as a permission request; nothing runs - until you allow it, and a timeout denies by default. The looser modes are +- **Per-tool approval — in the modes that ask.** In `default` mode Claude follows the local approval rules; tools that need + approval are relayed to you. A timeout denies by default. The looser modes are looser on purpose: `acceptEdits` auto-accepts edits, and `bypassPermissions` runs everything unattended. Codex never asks interactively at all — `codex exec` maps the permission mode to a sandbox policy instead (plan → @@ -245,3 +245,18 @@ decision to a remote approver — was inspired by [VibeBridge](https://github.com/Swayyyyy/VibeBridge) (GPL-3.0). This is an independent implementation built on the public `claude-agent-sdk`; **no VibeBridge source code is included.** See [NOTICE](https://github.com/AceDataCloud/CodingBridge/blob/main/NOTICE). + +## Session reliability + +- **Stop** cancels the active Codex process and its child commands and returns the + session to idle. You can continue the same conversation afterward. Claude uses + its native interrupt control, and pending Bridge approvals are denied on stop. +- Codex output is displayed as soon as a complete message arrives, without an + artificial typewriter delay. Individual JSON events up to 8 MiB are supported; + larger events fail clearly and terminate the subprocess instead of hanging. +- Claude model and permission changes must succeed before they are applied. + Changing reasoning effort resumes the same transcript with the new setting. +- History and capability discovery run independently so loading a large local + history cannot block stop or approval commands. +- New Homebrew service installations use a stable `opt` interpreter path when + available. Reinstall an existing service once to replace an old Cellar path. diff --git a/coding_bridge/capabilities.py b/coding_bridge/capabilities.py index b99a346..c534fcb 100644 --- a/coding_bridge/capabilities.py +++ b/coding_bridge/capabilities.py @@ -16,6 +16,7 @@ import asyncio import contextlib +import importlib.util import json import logging import os @@ -58,8 +59,9 @@ # Codex host paths. `models_cache.json` is codex's own API-fetched, per-account # model catalog (authoritative); config.toml carries the user's chosen default # model / effort, used only as a fallback when the cache can't be read. -CODEX_CONFIG = Path.home() / ".codex" / "config.toml" -CODEX_MODELS_CACHE = Path.home() / ".codex" / "models_cache.json" +CODEX_HOME = Path(os.environ.get("CODEX_HOME") or str(Path.home() / ".codex")) +CODEX_CONFIG = CODEX_HOME / "config.toml" +CODEX_MODELS_CACHE = CODEX_HOME / "models_cache.json" _CODEX_FALLBACK_MODELS: list[dict[str, str]] = [ {"value": "gpt-5.5", "label": "GPT-5.5"}, {"value": "gpt-5", "label": "GPT-5"}, @@ -170,9 +172,26 @@ def _candidate_cli_paths(cli: str) -> list[str]: if cli == "claude": fixed.insert(0, home / ".claude" / "local" / "claude") # native installer out += [str(p) for p in fixed] + if cli == "claude": + bundled = _bundled_claude() + if bundled: + out.append(bundled) return out +def _bundled_claude() -> str | None: + try: + spec = importlib.util.find_spec("claude_agent_sdk") + except (ImportError, ValueError): + return None + if spec and spec.origin: + name = "claude.exe" if os.name == "nt" else "claude" + path = Path(spec.origin).parent / "_bundled" / name + if path.is_file(): + return str(path) + return None + + def resolve_cli(cli: str, settings: Any | None = None) -> str | None: """Absolute path to a provider CLI, or None if genuinely absent. @@ -365,6 +384,7 @@ async def _probe_claude_commands(settings: Any) -> list[dict[str, Any]]: cwd=(getattr(settings, "default_cwd", "") or None), system_prompt={"type": "preset", "preset": "claude_code"}, setting_sources=["user", "project", "local"], + cli_path=resolve_cli("claude", settings), ) client = ClaudeSDKClient(options=options) await asyncio.wait_for(client.connect(), timeout=45) @@ -380,27 +400,8 @@ async def _probe_claude_commands(settings: Any) -> list[dict[str, Any]]: def _codex_commands() -> list[dict[str, Any]]: - """Codex custom prompts (`$CODEX_HOME/prompts/*.md`) surfaced as slash commands. - - `codex exec` is non-interactive and has no built-in slash processor, so only - user-defined prompt files are advertised; the rest of Codex's interactive - slash commands cannot run remotely. - """ - home = os.environ.get("CODEX_HOME") or os.path.join(os.path.expanduser("~"), ".codex") - prompts_dir = Path(home) / "prompts" - if not prompts_dir.is_dir(): - return [] - commands: list[dict[str, Any]] = [] - try: - entries = sorted(prompts_dir.glob("*.md")) - except OSError: - return [] - for path in entries: - name = path.stem - if not name: - continue - commands.append({"name": name, "description": "", "argument_hint": "", "aliases": []}) - return commands + """exec cannot execute slash commands; do not advertise unusable entries.""" + return [] def _copilot_commands() -> list[dict[str, Any]]: diff --git a/coding_bridge/connection.py b/coding_bridge/connection.py index 47df3ac..717b30e 100644 --- a/coding_bridge/connection.py +++ b/coding_bridge/connection.py @@ -60,6 +60,7 @@ def __init__( self._truncated_sessions: set[str] = set() # Dedup browser commands the relay may redeliver (a command queued across # our reconnect, then flushed). Bounded ring of recently seen cmd_ids. + self._read_tasks: dict[str, asyncio.Task] = {} self._seen_cmds: set[str] = set() self._seen_cmd_order: deque[str] = deque() @@ -259,7 +260,17 @@ async def _on_raw(self, raw: str | bytes) -> None: # re-spawn a turn or re-run a prompt. if self._is_duplicate_command(message.get("cmd_id")): return - await self._dispatch(message.get("payload") or {}) + payload = message.get("payload") or {} + action = payload.get("action") + if action in (Action.HISTORY_LIST, Action.CAPABILITIES_GET): + # Large local histories and CLI initialization must not delay + # stop/approval commands or acknowledgement processing. + existing = self._read_tasks.get(action) + if existing is None or existing.done(): + task = asyncio.create_task(self._dispatch(payload)) + self._read_tasks[action] = task + else: + await self._dispatch(payload) elif msg_type == protocol.NODE_ACK: up_to = (message.get("payload") or {}).get("up_to_node_seq") if isinstance(up_to, int): @@ -579,6 +590,11 @@ async def _send_capabilities(self) -> None: await self.send_payload(event_payload(Event.CAPABILITIES, **descriptor)) async def aclose(self) -> None: + tasks = list(self._read_tasks.values()) + for task in tasks: + task.cancel() + await asyncio.gather(*tasks, return_exceptions=True) + self._read_tasks.clear() for session in list(self.sessions.values()): await session.close() self.sessions.clear() diff --git a/coding_bridge/history.py b/coding_bridge/history.py index c33505b..67912e9 100644 --- a/coding_bridge/history.py +++ b/coding_bridge/history.py @@ -19,6 +19,7 @@ from collections.abc import Callable, Iterator from concurrent.futures import ThreadPoolExecutor from datetime import datetime +from functools import lru_cache from pathlib import Path from typing import Any @@ -30,8 +31,9 @@ CLAUDE_ROOT = ( Path(os.environ.get("CLAUDE_CONFIG_DIR") or str(Path.home() / ".claude")) / "projects" ) -CODEX_ROOT = Path.home() / ".codex" / "sessions" -CODEX_INDEX = Path.home() / ".codex" / "session_index.jsonl" +CODEX_HOME = Path(os.environ.get("CODEX_HOME") or str(Path.home() / ".codex")) +CODEX_ROOT = CODEX_HOME / "sessions" +CODEX_INDEX = CODEX_HOME / "session_index.jsonl" COPILOT_ROOT = ( Path(os.environ.get("COPILOT_HOME") or str(Path.home() / ".copilot")) / "session-state" ) @@ -179,12 +181,26 @@ def _one(path: Path) -> dict[str, Any] | None: return [r for r in results if r is not None] +def _cached_summary(path: Path, provider: str) -> dict[str, Any]: + stat = path.stat() + # Only unchanged files reuse a summary. Copies keep caller annotations from + # leaking into the cache; the bounded cache holds metadata, never transcripts. + return dict(_summary_at_version(str(path), provider, stat.st_mtime_ns, stat.st_size)) + + +@lru_cache(maxsize=1000) +def _summary_at_version(path: str, provider: str, mtime_ns: int, size: int) -> dict[str, Any]: + if provider == "claude": + return _claude_summary(Path(path)) + return _codex_summary(Path(path), {}) + + # --- Claude Code ----------------------------------------------------------- def _list_claude(limit: int) -> list[dict[str, Any]]: if not CLAUDE_ROOT.exists(): return [] files = sorted(CLAUDE_ROOT.glob("*/*.jsonl"), key=_safe_mtime, reverse=True)[:limit] - return _summaries(files, _claude_summary) + return _summaries(files, lambda path: _cached_summary(path, "claude")) def _claude_summary(path: Path) -> dict[str, Any]: @@ -337,7 +353,12 @@ def _list_codex(limit: int) -> list[dict[str, Any]]: return [] files = sorted(CODEX_ROOT.glob("**/rollout-*.jsonl"), key=_safe_mtime, reverse=True)[:limit] index = _codex_index() - return _summaries(files, lambda path: _codex_summary(path, index)) + summaries = _summaries(files, lambda path: _cached_summary(path, "codex")) + for summary in summaries: + meta = index.get(summary.get("session_id") or "", {}) + summary["title"] = _provider_title(meta.get("thread_name")) or summary["title"] + summary["updated_at"] = _iso_ms(meta.get("updated_at")) or summary["updated_at"] + return summaries def _codex_summary(path: Path, index: dict[str, dict[str, Any]]) -> dict[str, Any]: diff --git a/coding_bridge/providers/claude.py b/coding_bridge/providers/claude.py index fe1429d..531d5f2 100644 --- a/coding_bridge/providers/claude.py +++ b/coding_bridge/providers/claude.py @@ -51,6 +51,7 @@ def __init__( self._cwd = settings.default_cwd self._model: str | None = settings.default_model self._permission_mode = "default" + self._effort: str | None = None self._client: Any = None self._connected = False self._server_info: dict[str, Any] | None = None @@ -59,7 +60,7 @@ def __init__( self._turn_seq = 0 self._stream_text_ord = 0 self._open_text: dict[Any, dict[str, Any]] = {} - self._saw_text_stream = False + self._streamed_blocks: list[dict[str, str]] = [] # The SDK/CLI's own session id (the resume target) and the uuid of the # last transcript message seen. Reported on each turn's result so the # browser can fork the conversation here when a prompt is edited. @@ -118,9 +119,7 @@ def _arm_replay_guard(self, session_id: str | None) -> None: with contextlib.suppress(Exception): watermark = history.claude_watermark(session_id) - self._gate_uuids, self._gate_msg_ids = history.claude_ids_before( - session_id, watermark - ) + self._gate_uuids, self._gate_msg_ids = history.claude_ids_before(session_id, watermark) self._gate_active = bool(self._gate_uuids or self._gate_msg_ids) async def send( @@ -133,6 +132,10 @@ async def send( effort: str | None = None, permission_mode: str | None = None, ) -> None: + if self._connected and effort != self._effort: + # Effort has no runtime setter. Resume the same transcript with the + # requested effort instead of silently retaining the previous value. + await self.aclose() if not self._connected: # The subprocess died mid-conversation. Resume its transcript or the # follow-up loses all context — and the CLI would replay it anyway. @@ -173,14 +176,19 @@ async def edit( """ resume_id = self._sdk_session_id # Optionally roll the working tree back to the edited turn before forking. - if restore_code and resume_id and self._client and hasattr(self._client, "rewind_files"): + if restore_code: + if not resume_id or not self._client or not hasattr(self._client, "rewind_files"): + raise RuntimeError("Code restore is unavailable for this session") from .. import history target = history.claude_user_uuid_after(resume_id, cut_uuid) - if target: - with contextlib.suppress(Exception): - await self._client.rewind_files(target) + if not target: + raise RuntimeError("No file checkpoint exists for the selected turn") + # Never claim a rollback succeeded when the SDK rejected it. + await self._client.rewind_files(target) await self.aclose() + self._announced_identity = False + self._last_msg_uuid = None self._cwd = self._cwd or self._settings.default_cwd fork = bool(cut_uuid and resume_id) resume = resume_id if fork else None @@ -198,25 +206,21 @@ async def edit( return await self._turn(self._with_attachments(prompt, images, attachments), resume_of=resume) - async def _apply_runtime_changes( - self, model: str | None, permission_mode: str | None - ) -> None: + async def _apply_runtime_changes(self, model: str | None, permission_mode: str | None) -> None: """Apply the mid-session changes the streaming SDK supports live. - Reasoning effort has no live SDK setter, so it only takes effect on the - next fresh session; model and permission mode switch in place. + Model and permission mode switch in place. send() reconnects the same + transcript when reasoning effort changes because the SDK has no setter. """ if model != self._model and hasattr(self._client, "set_model"): - with contextlib.suppress(Exception): - await self._client.set_model(model or None) + await self._client.set_model(model or None) self._model = model if ( permission_mode and permission_mode != self._permission_mode and hasattr(self._client, "set_permission_mode") ): - with contextlib.suppress(Exception): - await self._client.set_permission_mode(permission_mode) + await self._client.set_permission_mode(permission_mode) self._permission_mode = permission_mode async def _ensure_client( @@ -256,6 +260,7 @@ async def _ensure_client( system_prompt={"type": "preset", "preset": "claude_code"}, setting_sources=["user", "project", "local"], resume=resume_id, + cli_path=capabilities.resolve_cli("claude", self._settings), ) # The SDK stamps CLAUDE_CODE_ENTRYPOINT=sdk-py, which both the VSCode # extension and `claude --resume` treat as "programmatic" and hide from @@ -286,6 +291,7 @@ async def _ensure_client( self._connected = True self._model = model self._permission_mode = permission_mode or "default" + self._effort = effort await self._load_server_info() async def _load_server_info(self) -> None: @@ -369,6 +375,8 @@ async def _gated_handle(self, message: Any) -> bool: mid = msg.get("id") self._gate_stream_replay = bool(mid and mid in self._gate_msg_ids) self._gate_saw_replay = self._gate_saw_replay or self._gate_stream_replay + if not self._gate_stream_replay: + await self._handle_stream_event(stream_event) return False if self._gate_stream_replay: return False # drop deltas/stop of a replayed streaming message @@ -417,9 +425,7 @@ def _note_system(self, message: Any) -> None: resolved_model, ) - def _with_attachments( - self, prompt: str, images: list | None, attachments: list | None - ) -> str: + def _with_attachments(self, prompt: str, images: list | None, attachments: list | None) -> str: image_paths = image_store.save_images(images, self._cwd, session_id=self._session_id) files = attachment_store.save_attachments( attachments, self._cwd, session_id=self._session_id @@ -451,7 +457,12 @@ async def _handle_message(self, message: Any) -> None: event_payload( Event.SESSION_RESULT, self._session_id, - subtype=getattr(message, "subtype", None), + subtype=( + "interrupted" + if getattr(message, "terminal_reason", None) + in {"aborted_streaming", "aborted_tools"} + else getattr(message, "subtype", None) + ), is_error=bool(getattr(message, "is_error", False)), result=result, cost_usd=getattr(message, "total_cost_usd", None), @@ -515,9 +526,7 @@ async def _emit_status(self) -> None: lines.append(f"- **Model:** {self._model}") lines.append(f"- **Permission mode:** {self._permission_mode}") lines.append(f"- **Working directory:** {self._cwd or self._settings.default_cwd}") - await self._emit( - event_payload(Event.SESSION_TEXT, self._session_id, text="\n".join(lines)) - ) + await self._emit(event_payload(Event.SESSION_TEXT, self._session_id, text="\n".join(lines))) async def _emit_slash_notice(self, name: str) -> None: await self._emit( @@ -550,17 +559,18 @@ def _begin_stream_turn(self) -> None: self._turn_seq += 1 self._stream_text_ord = 0 self._open_text = {} - self._saw_text_stream = False + self._streamed_blocks = [] async def _flush_open_text(self) -> None: """Commit any streamed text block that never saw a stop event.""" if not self._open_text: return for blk in list(self._open_text.values()): + if blk.get("committed"): + continue + self._streamed_blocks.append(blk) await self._emit( - event_payload( - Event.SESSION_TEXT, self._session_id, text=blk["text"], id=blk["id"] - ) + event_payload(Event.SESSION_TEXT, self._session_id, text=blk["text"], id=blk["id"]) ) self._open_text = {} @@ -571,20 +581,31 @@ async def _handle_stream_event(self, raw: dict[str, Any]) -> None: still emitted whole from the assembled AssistantMessage. """ etype = raw.get("type") - if etype == "content_block_start": + if etype == "message_start": + self._streamed_blocks = [] + elif etype == "content_block_start": block = raw.get("content_block") or {} if block.get("type") == "text": stream_id = f"{self._session_id}:{self._turn_seq}:{self._stream_text_ord}" self._stream_text_ord += 1 - self._open_text[raw.get("index")] = {"id": stream_id, "text": ""} + initial = block.get("text") or "" + self._open_text[raw.get("index")] = {"id": stream_id, "text": initial} + if initial: + await self._emit( + event_payload( + Event.SESSION_TEXT_DELTA, + self._session_id, + text=initial, + id=stream_id, + ) + ) elif etype == "content_block_delta": delta = raw.get("delta") or {} if delta.get("type") == "text_delta": blk = self._open_text.get(raw.get("index")) - if blk is not None: + if blk is not None and not blk.get("committed"): chunk = delta.get("text") or "" blk["text"] += chunk - self._saw_text_stream = True await self._emit( event_payload( Event.SESSION_TEXT_DELTA, @@ -595,7 +616,8 @@ async def _handle_stream_event(self, raw: dict[str, Any]) -> None: ) elif etype == "content_block_stop": blk = self._open_text.pop(raw.get("index"), None) - if blk is not None: + if blk is not None and not blk.get("committed"): + self._streamed_blocks.append(blk) await self._emit( event_payload( Event.SESSION_TEXT, self._session_id, text=blk["text"], id=blk["id"] @@ -608,10 +630,40 @@ async def _handle_block(self, block: Any) -> None: event_payload(Event.SESSION_THINKING, self._session_id, text=block.thinking) ) elif hasattr(block, "text"): - # Already streamed + committed via stream events this turn. - if self._saw_text_stream: - return - await self._emit(event_payload(Event.SESSION_TEXT, self._session_id, text=block.text)) + # Match this complete block, not a turn-wide "saw any delta" flag. + # Some providers stream an empty/partial block then deliver the real + # answer only in AssistantMessage; it must repair the same bubble. + open_block = next( + (item for item in self._open_text.values() if not item.get("committed")), None + ) + if open_block is not None: + # The SDK can emit AssistantMessage BEFORE content_block_stop. + # Commit its authoritative text to the same bubble exactly once. + open_block.update(text=block.text, committed=True) + await self._emit( + event_payload( + Event.SESSION_TEXT, + self._session_id, + text=block.text, + id=open_block["id"], + ) + ) + elif self._streamed_blocks: + streamed = self._streamed_blocks.pop(0) + if streamed["text"] == block.text: + return + await self._emit( + event_payload( + Event.SESSION_TEXT, + self._session_id, + text=block.text, + id=streamed["id"], + ) + ) + else: + await self._emit( + event_payload(Event.SESSION_TEXT, self._session_id, text=block.text) + ) elif hasattr(block, "name") and hasattr(block, "input"): await self._emit( event_payload( diff --git a/coding_bridge/providers/codex.py b/coding_bridge/providers/codex.py index 217a4c9..0f63a79 100644 --- a/coding_bridge/providers/codex.py +++ b/coding_bridge/providers/codex.py @@ -11,12 +11,15 @@ import contextlib import json import logging -import math -import shutil +import os +import signal import uuid from typing import TYPE_CHECKING, Any +import psutil + from .. import attachments as attachment_store +from .. import capabilities from .. import images as image_store from ..protocol import Event, event_payload from .base import slash_name @@ -38,12 +41,24 @@ _EFFORT_ALIASES = {"max": "high", "ultra-high": "high", "ultrahigh": "high"} _EFFORT_VALUES = {"minimal", "low", "medium", "high", "xhigh"} -# Typewriter cadence for codex agent messages. ``codex exec`` delivers the final -# message whole, so we replay it as text deltas for visual streaming parity with -# Claude. Tests patch STREAM_DELAY to 0 for determinism. -STREAM_CHUNK_TARGET = 80 -STREAM_MIN_CHUNK = 3 -STREAM_DELAY = 0.012 +# Bound malformed/unbounded provider output without asyncio's 64 KiB line limit. +_MAX_EVENT_BYTES = 8 * 1024 * 1024 +_STOP_GRACE_SECONDS = 2.0 + + +async def _json_lines(stream: asyncio.StreamReader): + pending = bytearray() + while chunk := await stream.read(64 * 1024): + pending.extend(chunk) + while (end := pending.find(b"\n")) >= 0: + if end > _MAX_EVENT_BYTES: + raise RuntimeError("Codex output event exceeds the 8 MiB limit") + yield bytes(pending[:end]) + del pending[: end + 1] + if len(pending) > _MAX_EVENT_BYTES: + raise RuntimeError("Codex output event exceeds the 8 MiB limit") + if pending: + yield bytes(pending) def _codex_effort(effort: str | None) -> str | None: @@ -76,6 +91,8 @@ def __init__( self._proc: asyncio.subprocess.Process | None = None # Announced the real (thread) id to the browser yet? See claude.py. self._announced_identity = False + self._interrupted = False + self._stop_lock = asyncio.Lock() async def start( self, @@ -229,7 +246,8 @@ async def _run_turn( images: list | None = None, attachments: list | None = None, ) -> None: - if shutil.which("codex") is None: + cli = capabilities.resolve_cli("codex", self._settings) + if cli is None: raise RuntimeError( "codex CLI is not installed; install it from https://github.com/openai/codex" ) @@ -240,24 +258,24 @@ async def _run_turn( image_paths = legacy_image_paths + attachment_store.image_paths(files) prompt = attachment_store.attachment_note(prompt, files, legacy_image_paths) argv = self._build_argv(prompt, resume=resume, image_paths=image_paths) + argv[0] = cli + self._interrupted = False self._proc = await asyncio.create_subprocess_exec( *argv, cwd=self._cwd or None, stdin=asyncio.subprocess.DEVNULL, # never block reading an inherited stdin stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, + start_new_session=os.name == "posix", ) proc = self._proc - stderr_tail: list[str] = [] + stderr_tail = bytearray() drain = asyncio.create_task(self._drain_stderr(proc, stderr_tail)) self._last_error = None saw_result = False try: assert proc.stdout is not None - while True: - raw = await proc.stdout.readline() - if not raw: - break + async for raw in _json_lines(proc.stdout): line = raw.decode("utf-8", "replace").strip() if not line: continue @@ -267,30 +285,46 @@ async def _run_turn( continue if await self._handle_event(obj): saw_result = True + except BaseException: + # Kill before waiting: unread pipes can otherwise deadlock proc.wait(). + await self._stop_process_tree(proc) + if proc.stdout is not None: + await proc.stdout.read() + raise finally: await proc.wait() drain.cancel() with contextlib.suppress(asyncio.CancelledError): await drain - self._proc = None - if proc.returncode not in (0, None) and not saw_result: - detail = ( - self._last_error - or " ".join(stderr_tail[-5:]).strip() - or f"codex exited with {proc.returncode}" - ) - await self._emit(event_payload(Event.SESSION_ERROR, self._session_id, message=detail)) + if self._proc is proc: + self._proc = None + if not saw_result: + if self._interrupted: + await self._emit( + event_payload( + Event.SESSION_RESULT, + self._session_id, + subtype="interrupted", + is_error=False, + sdk_session_id=self._thread_id, + ) + ) + else: + detail = ( + self._last_error + or stderr_tail.decode("utf-8", "replace").strip() + or f"Codex exited without a turn result (exit {proc.returncode})" + ) + await self._emit( + event_payload(Event.SESSION_ERROR, self._session_id, message=detail) + ) - async def _drain_stderr(self, proc: asyncio.subprocess.Process, tail: list[str]) -> None: + async def _drain_stderr(self, proc: asyncio.subprocess.Process, tail: bytearray) -> None: if proc.stderr is None: return - with contextlib.suppress(asyncio.CancelledError): - while True: - raw = await proc.stderr.readline() - if not raw: - return - tail.append(raw.decode("utf-8", "replace").rstrip()) - del tail[:-20] + while chunk := await proc.stderr.read(16 * 1024): + tail.extend(chunk) + del tail[: -16 * 1024] async def _maybe_announce_identity(self) -> None: """Adopt the codex thread id as this session's canonical id, once known. @@ -400,42 +434,52 @@ async def _handle_item(self, phase: str, item: dict[str, Any]) -> None: await self._emit_text_stream(item["text"]) async def _emit_text_stream(self, text: str) -> None: - """Replay a whole agent message as incremental text deltas + a commit. - - ``codex exec`` has no native token streaming, so we chunk the final - message into deltas (capped chunk count) to mirror Claude's streaming - UX, then emit an authoritative ``session.text`` carrying the same id. - """ - stream_id = f"{self._session_id}:{uuid.uuid4().hex[:8]}" - size = max(STREAM_MIN_CHUNK, math.ceil(len(text) / STREAM_CHUNK_TARGET)) - for start in range(0, len(text), size): - chunk = text[start : start + size] - await self._emit( - event_payload( - Event.SESSION_TEXT_DELTA, self._session_id, text=chunk, id=stream_id - ) - ) - if STREAM_DELAY: - await asyncio.sleep(STREAM_DELAY) + # exec supplies whole messages. Forward immediately instead of delaying + # the next event behind dozens of artificial typewriter updates. await self._emit( - event_payload(Event.SESSION_TEXT, self._session_id, text=text, id=stream_id) + event_payload( + Event.SESSION_TEXT, + self._session_id, + text=text, + id=f"{self._session_id}:{uuid.uuid4().hex[:8]}", + ) ) + async def _stop_process_tree(self, proc: asyncio.subprocess.Process) -> None: + async with self._stop_lock: + + def stop() -> None: + try: + parent = psutil.Process(proc.pid) + children = parent.children(recursive=True) + except psutil.NoSuchProcess: + return + # Codex gives shell jobs their own process groups. Killing only + # Codex's group misses them; capture descendants before reparenting. + targets = [*reversed(children), parent] + for target in targets: + with contextlib.suppress(psutil.NoSuchProcess): + target.terminate() + _, alive = psutil.wait_procs(targets, timeout=_STOP_GRACE_SECONDS) + for target in alive: + with contextlib.suppress(psutil.NoSuchProcess): + target.kill() + if os.name == "posix": + with contextlib.suppress(ProcessLookupError): + os.killpg(proc.pid, signal.SIGKILL) + psutil.wait_procs(alive, timeout=_STOP_GRACE_SECONDS) + + await asyncio.to_thread(stop) + async def interrupt(self) -> None: proc = self._proc - if proc is None or proc.returncode is not None: + if proc is None: return - with contextlib.suppress(ProcessLookupError): - proc.terminate() + self._interrupted = True + await self._stop_process_tree(proc) async def aclose(self) -> None: - proc = self._proc - if proc is not None and proc.returncode is None: - with contextlib.suppress(ProcessLookupError): - proc.kill() - with contextlib.suppress(Exception): - await proc.wait() - self._proc = None + await self.interrupt() def _error_message(obj: dict[str, Any]) -> str | None: diff --git a/coding_bridge/service_cli.py b/coding_bridge/service_cli.py index ecd1fad..712bc70 100644 --- a/coding_bridge/service_cli.py +++ b/coding_bridge/service_cli.py @@ -33,11 +33,25 @@ def _system() -> str: def _plan(settings: Settings) -> ServicePlan | None: try: - return build_service_plan(_system(), sys.executable, settings.config_dir, Path.home()) + return build_service_plan(_system(), _service_python(), settings.config_dir, Path.home()) except ValueError: return None +def _service_python() -> str: + """Keep a Homebrew service on the stable opt link across version upgrades.""" + path = Path(sys.executable) + parts = path.parts + if "Cellar" in parts: + index = parts.index("Cellar") + if len(parts) > index + 3: + stable = Path(*parts[:index]) / "opt" / parts[index + 1] + stable = stable.joinpath(*parts[index + 3:]) + if stable.is_file() and stable.resolve() == path.resolve(): + return str(stable) + return str(path) + + def _run(cmds: list[list[str]]) -> int: """Run manager commands in order; stop at the first failure and return it. diff --git a/coding_bridge/session.py b/coding_bridge/session.py index a72a62b..723f762 100644 --- a/coding_bridge/session.py +++ b/coding_bridge/session.py @@ -338,6 +338,12 @@ async def _reset_provider(self) -> None: await self._provider.aclose() async def interrupt(self) -> None: + pending = self._broker.pending_ids() + self._broker.cancel_all("deny") + for request_id in pending: + await self._emit(event_payload( + Event.PERMISSION_RESOLVED, self.session_id, request_id=request_id, decision="deny", + )) await self._provider.interrupt() async def close(self) -> None: diff --git a/pyproject.toml b/pyproject.toml index e44d6d2..8119914 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -23,6 +23,7 @@ dependencies = [ "websockets>=12,<14", "httpx>=0.27,<0.29", "claude-agent-sdk>=0.1.0", + "psutil>=5.9", "tomli>=2.0;python_version<'3.11'", ] diff --git a/tests/conftest.py b/tests/conftest.py index 842ff9b..f23c672 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -6,14 +6,6 @@ sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) -@pytest.fixture(autouse=True) -def _no_codex_stream_delay(monkeypatch): - """Keep the codex typewriter instant so tests stay fast and deterministic.""" - from coding_bridge.providers import codex - - monkeypatch.setattr(codex, "STREAM_DELAY", 0, raising=False) - - @pytest.fixture(autouse=True) def _isolate_home(tmp_path_factory, monkeypatch): """Redirect ``~`` so sidecars never touch the developer's real ~/.ace-bridge. diff --git a/tests/test_capabilities.py b/tests/test_capabilities.py index cccd893..8d8139f 100644 --- a/tests/test_capabilities.py +++ b/tests/test_capabilities.py @@ -186,7 +186,7 @@ def test_command_name_set_lowercases_names_and_aliases(): assert capabilities.command_name_set(cmds) == {"usage", "cost", "stats", "context"} -def test_codex_commands_reads_prompts_dir(tmp_path, monkeypatch): +def test_codex_does_not_advertise_unexecutable_custom_prompts(tmp_path, monkeypatch): prompts = tmp_path / "prompts" prompts.mkdir() (prompts / "deploy.md").write_text("do deploy") @@ -194,7 +194,7 @@ def test_codex_commands_reads_prompts_dir(tmp_path, monkeypatch): (prompts / "ignore.txt").write_text("not markdown") monkeypatch.setenv("CODEX_HOME", str(tmp_path)) names = sorted(c["name"] for c in capabilities._codex_commands()) - assert names == ["deploy", "review"] + assert names == [] def test_codex_commands_empty_without_dir(tmp_path, monkeypatch): @@ -222,3 +222,13 @@ def test_claude_offers_the_1m_context_variants(): assert {"opus[1m]", "sonnet[1m]"} <= values # "" means "send no --model", so the CLI applies the user's settings.json. assert "" in values + + +def test_claude_sdk_bundled_cli_is_available_without_global_install(monkeypatch, tmp_path): + binary = tmp_path / 'claude' + binary.write_text('#!/bin/sh\n') + binary.chmod(0o755) + monkeypatch.setattr(capabilities.shutil, 'which', lambda cli: None) + monkeypatch.setattr(capabilities, '_bundled_claude', lambda: str(binary)) + assert capabilities.resolve_cli('claude') == str(binary) + assert capabilities.describe()['providers'][0]['available'] diff --git a/tests/test_claude_entrypoint.py b/tests/test_claude_entrypoint.py index 9588536..7d432d5 100644 --- a/tests/test_claude_entrypoint.py +++ b/tests/test_claude_entrypoint.py @@ -118,3 +118,36 @@ def test_from_env_reads_override(monkeypatch): def test_from_env_defaults_to_visible(monkeypatch): monkeypatch.delenv("CODING_BRIDGE_CLAUDE_ENTRYPOINT", raising=False) assert Settings.from_env().claude_entrypoint == DEFAULT_CLAUDE_ENTRYPOINT + + +async def test_explicit_cli_path_is_passed_to_sdk(fake_sdk, monkeypatch): + monkeypatch.setattr( + 'coding_bridge.providers.claude.capabilities.resolve_cli', + lambda cli, settings: '/custom/claude', + ) + options = await _connect(Settings(claude_path='/custom/claude'), fake_sdk) + assert options.cli_path == '/custom/claude' + + +async def test_effort_change_resumes_same_session(monkeypatch): + provider = ClaudeProvider('s1', _noop_emit, None, Settings()) + provider._connected = True + provider._effort = 'low' + provider._sdk_session_id = 'sdk-existing' + calls = {} + async def close(): + provider._connected = False + async def connect(**kwargs): + calls.update(kwargs) + async def turn(*args, **kwargs): + pass + monkeypatch.setattr(provider, 'aclose', close) + monkeypatch.setattr(provider, '_ensure_client', connect) + monkeypatch.setattr(provider, '_turn', turn) + await provider.send('continue', effort='high') + assert calls['effort'] == 'high' + assert calls['resume'] == 'sdk-existing' + + +async def _noop_emit(_event): + pass diff --git a/tests/test_codex_process.py b/tests/test_codex_process.py new file mode 100644 index 0000000..097a2d3 --- /dev/null +++ b/tests/test_codex_process.py @@ -0,0 +1,116 @@ +"""Real subprocess regressions: long JSONL, cancellation and detached shell jobs.""" + +import asyncio +import os +import sys + +import pytest + +from coding_bridge.config import Settings +from coding_bridge.providers import codex +from coding_bridge.providers.codex import CodexProvider + + +def provider(tmp_path): + events = [] + + async def emit(event): + events.append(event) + + return CodexProvider("probe", emit, None, Settings(default_cwd=str(tmp_path))), events + + +def executable(monkeypatch, source): + spawn = asyncio.create_subprocess_exec + + async def start(*args, **kwargs): + return await spawn(sys.executable, "-u", "-c", source, **kwargs) + + monkeypatch.setattr(codex.asyncio, "create_subprocess_exec", start) + monkeypatch.setattr(codex.capabilities, "resolve_cli", lambda *a: sys.executable) + + +async def run(p, tmp_path): + await p.start("test", cwd=str(tmp_path), model=None, permission_mode="default") + + +@pytest.mark.parametrize("size", [70_000, 250_000, 2_000_000]) +async def test_large_stdout_and_stderr_do_not_block(tmp_path, monkeypatch, size): + executable( + monkeypatch, + f"""import json,sys +sys.stderr.write('e' * {size});sys.stderr.flush() +print(json.dumps({{"type":"item.completed","item":{{"type":"agent_message","text":"x"*{size}}}}})) +print(json.dumps({{"type":"turn.completed"}}))""", + ) + p, events = provider(tmp_path) + await asyncio.wait_for(run(p, tmp_path), 10) + assert next(e["text"] for e in events if e["event"] == "session.text") == "x" * size + assert events[-1]["event"] == "session.result" + + +async def test_oversize_event_stops_process_before_wait(tmp_path, monkeypatch): + monkeypatch.setattr(codex, "_MAX_EVENT_BYTES", 100_000) + executable( + monkeypatch, + "import sys,time;sys.stdout.write('x'*500000);sys.stdout.flush();time.sleep(60)", + ) + p, _ = provider(tmp_path) + with pytest.raises(RuntimeError, match="8 MiB"): + await asyncio.wait_for(run(p, tmp_path), 8) + assert p._proc is None + + +async def test_clean_exit_without_result_is_an_error(tmp_path, monkeypatch): + executable(monkeypatch, "pass") + p, events = provider(tmp_path) + await run(p, tmp_path) + assert events[-1]["event"] == "session.error" + assert "without a turn result" in events[-1]["message"] + + +@pytest.mark.skipif(os.name != "posix", reason="POSIX detached process-group regression") +async def test_interrupt_stops_detached_descendant_and_emits_terminal(tmp_path, monkeypatch): + marker = tmp_path / "late.txt" + child = ( + f"import time,pathlib; time.sleep(1); pathlib.Path({str(marker)!r})" + ".write_text('bad');time.sleep(60)" + ) + executable( + monkeypatch, + f"""import subprocess,sys,json,time +subprocess.Popen([sys.executable,'-c',{child!r}],start_new_session=True) +print(json.dumps({{"type":"thread.started","thread_id":"test"}})) +time.sleep(60)""", + ) + p, events = provider(tmp_path) + task = asyncio.create_task(run(p, tmp_path)) + try: + + async def identified(): + while not p._announced_identity: + await asyncio.sleep(0.01) + + await asyncio.wait_for(identified(), 5) + await p.interrupt() + await asyncio.wait_for(task, 8) + await asyncio.sleep(1.1) + assert not marker.exists() + terminal = [e for e in events if e["event"] in ("session.error", "session.result")] + assert len(terminal) == 1 + assert terminal[0]["subtype"] == "interrupted" + finally: + await p.aclose() + + +async def test_cancelled_reader_reaps_process(tmp_path, monkeypatch): + executable(monkeypatch, "import time;time.sleep(60)") + p, _ = provider(tmp_path) + task = asyncio.create_task(run(p, tmp_path)) + while p._proc is None: + await asyncio.sleep(0.01) + proc = p._proc + task.cancel() + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(task, 8) + assert proc.returncode is not None diff --git a/tests/test_connection.py b/tests/test_connection.py index 9ac8ac9..f1a163d 100644 --- a/tests/test_connection.py +++ b/tests/test_connection.py @@ -710,3 +710,25 @@ async def test_history_detail_roundtrips_explicit_bare_selector(tmp_path, monkey ) assert detail["model"] == "opus" assert detail["resolved_model"] == "claude-opus-5" + + +async def test_slow_discovery_does_not_block_ping_or_duplicate_reads(monkeypatch): + conn = _new_conn() + release = asyncio.Event() + started = asyncio.Event() + calls = [] + async def slow(_payload): + calls.append('history') + started.set() + await release.wait() + monkeypatch.setattr(conn, '_send_history_list', slow) + raw = json.dumps({'type': protocol.BROWSER_TO_NODE, 'payload': {'action': Action.HISTORY_LIST}}) + await conn._on_raw(raw) + await started.wait() + await conn._on_raw(raw) + await conn._on_raw(json.dumps({ + 'type': protocol.BROWSER_TO_NODE, 'payload': {'action': Action.PING}, + })) + assert Event.PONG in _events(conn) + assert calls == ['history'] + await conn.aclose() diff --git a/tests/test_edit_fork.py b/tests/test_edit_fork.py index dd062c8..31f53d6 100644 --- a/tests/test_edit_fork.py +++ b/tests/test_edit_fork.py @@ -207,3 +207,33 @@ async def test_codex_edit_reports_unsupported(): notices = [e for e in events if e["event"] == "session.notice"] assert notices and notices[0]["code"] == "edit_unsupported" assert [e for e in events if e["event"] == "session.result"] + + +async def test_edit_announces_the_new_fork_identity(monkeypatch): + provider, events = _capturing() + provider._sdk_session_id = 'old-sdk' + provider._session_id = 'old-sdk' + provider._announced_identity = True + _stub_run(provider, monkeypatch) + await provider.edit('replacement', cut_uuid='keep') + await provider._handle_message(_Result(session_id='new-sdk')) + identities = [e for e in events if e['event'] == 'session.identified'] + assert identities == [{ + 'event': 'session.identified', 'session_id': 'old-sdk', 'sdk_session_id': 'new-sdk', + }] + assert events[-1]['session_id'] == 'new-sdk' + + +async def test_restore_failure_does_not_fork(monkeypatch): + import pytest + provider, _ = _capturing() + provider._sdk_session_id = 'existing' + class Client: + async def rewind_files(self, _target): + raise RuntimeError('checkpoint unavailable') + provider._client = Client() + monkeypatch.setattr(history, 'claude_user_uuid_after', lambda *a: 'checkpoint') + calls = _stub_run(provider, monkeypatch) + with pytest.raises(RuntimeError, match='checkpoint unavailable'): + await provider.edit('rewrite', cut_uuid='keep', restore_code=True) + assert not calls diff --git a/tests/test_history.py b/tests/test_history.py index 1e850a2..838c1e1 100644 --- a/tests/test_history.py +++ b/tests/test_history.py @@ -828,3 +828,21 @@ async def test_dispatch_history_get_requires_params(): await conn._dispatch({"action": Action.HISTORY_GET}) errors = _payloads(conn, Event.SESSION_ERROR) assert errors and "required" in errors[0]["message"] + + +def test_summary_cache_invalidates_when_transcript_changes(tmp_path, monkeypatch): + from coding_bridge import history + path = tmp_path / 'session.jsonl' + path.write_text('one') + calls = [] + def parse(p): + calls.append(p) + return {'title': p.read_text()} + monkeypatch.setattr(history, '_claude_summary', parse) + first = history._cached_summary(path, 'claude') + first['title'] = 'mutated' + assert history._cached_summary(path, 'claude')['title'] == 'one' + assert len(calls) == 1 + path.write_text('changed') + assert history._cached_summary(path, 'claude')['title'] == 'changed' + assert len(calls) == 2 diff --git a/tests/test_service.py b/tests/test_service.py index c5ccf1d..2e4818c 100644 --- a/tests/test_service.py +++ b/tests/test_service.py @@ -196,3 +196,15 @@ def _fake_run(cmd, check=False, capture_output=False): # noqa: ARG001 def _boom(*a, **k): # pragma: no cover - must never run raise AssertionError("subprocess.run should not be called here") + + +def test_homebrew_service_uses_stable_opt_path(monkeypatch, tmp_path): + from coding_bridge import service_cli + binary = tmp_path / 'Cellar/coding-bridge/2026.1/libexec/bin/python' + binary.parent.mkdir(parents=True) + binary.write_text('python') + opt = tmp_path / 'opt/coding-bridge' + opt.parent.mkdir() + opt.symlink_to(binary.parents[2], target_is_directory=True) + monkeypatch.setattr(service_cli.sys, 'executable', str(binary)) + assert service_cli._service_python() == str(opt / 'libexec/bin/python') diff --git a/tests/test_session.py b/tests/test_session.py index 83966f3..7c9d408 100644 --- a/tests/test_session.py +++ b/tests/test_session.py @@ -198,3 +198,18 @@ async def test_retry_limit_is_respected(): prov = holder["provider"] assert prov.calls == 2 # original + 1 retry, then give up assert len(_errors(events)) == 1 + + +async def test_interrupt_denies_pending_approval_and_clears_browser_prompt(): + import asyncio + sess, events, _ = _make(['ok']) + pending = asyncio.create_task(sess._ask_permission('Bash', {'command': 'echo test'}, {})) + await asyncio.sleep(0) + request_id = sess.pending_permissions()[0]['request_id'] + await sess.interrupt() + assert (await pending).decision == 'deny' + assert sess.pending_permissions() == [] + assert any( + e['event'] == Event.PERMISSION_RESOLVED and e['request_id'] == request_id + for e in events + ) diff --git a/tests/test_streaming.py b/tests/test_streaming.py index 7247eff..0992697 100644 --- a/tests/test_streaming.py +++ b/tests/test_streaming.py @@ -1,7 +1,6 @@ -"""Token-level streaming: Claude partial messages and codex typewriter parity.""" +"""Claude partial messages and immediate Codex complete-message delivery.""" from coding_bridge.config import Settings -from coding_bridge.providers import codex from coding_bridge.providers.claude import ClaudeProvider from coding_bridge.providers.codex import CodexProvider @@ -144,46 +143,56 @@ async def test_claude_two_text_blocks_get_distinct_ids(): assert len(ids) == 2 -# --- Codex typewriter parity ------------------------------------------------ +# --- Codex complete messages ------------------------------------------------ -async def test_codex_agent_message_streams_then_commits(): +async def test_codex_forwards_complete_text_once_without_artificial_deltas(): provider, events = _capturing(CodexProvider) + text = "Hello from Codex" * 1000 await provider._handle_event( - { - "type": "item.completed", - "item": {"type": "agent_message", "text": "Hello, world! Codex here."}, - } + {"type": "item.completed", "item": {"type": "agent_message", "text": text}} ) - deltas = [e for e in events if e["event"] == "session.text_delta"] - assert deltas - assert "".join(d["text"] for d in deltas) == "Hello, world! Codex here." - assert len({d["id"] for d in deltas}) == 1 + assert len(events) == 1 + assert events[0]["event"] == "session.text" + assert events[0]["text"] == text - texts = [e for e in events if e["event"] == "session.text"] - assert len(texts) == 1 - assert texts[0]["text"] == "Hello, world! Codex here." - assert texts[0]["id"] == deltas[0]["id"] - assert events[-1]["event"] == "session.text" # commit is last +async def test_claude_partial_block_repaired_by_complete_message(): + provider, events = _capturing(ClaudeProvider) + provider._begin_stream_turn() + await provider._handle_message(_start(0)) + await provider._handle_message(_delta(0, "partial")) + await provider._handle_message(_stop(0)) + await provider._handle_message(_Assistant([_TextBlock("complete answer")])) + texts = [e for e in events if e['event'] == 'session.text'] + assert texts[-1]['text'] == 'complete answer' + assert texts[0]['id'] == texts[-1]['id'] + # A later unstreamed answer is not swallowed because an earlier block streamed. + await provider._handle_message(_Assistant([_TextBlock("final answer")])) + assert events[-1]['text'] == 'final answer' -async def test_codex_stream_chunk_count_is_capped(): - provider, events = _capturing(CodexProvider) - text = "x" * 5000 - await provider._handle_event( - {"type": "item.completed", "item": {"type": "agent_message", "text": text}} - ) - deltas = [e for e in events if e["event"] == "session.text_delta"] - assert len(deltas) <= codex.STREAM_CHUNK_TARGET - assert "".join(d["text"] for d in deltas) == text +async def test_claude_stream_start_with_initial_text_is_preserved(): + provider, events = _capturing(ClaudeProvider) + provider._begin_stream_turn() + await provider._handle_message(_Stream({ + 'type': 'content_block_start', 'index': 0, + 'content_block': {'type': 'text', 'text': 'initial'}, + })) + await provider._handle_message(_stop(0)) + await provider._handle_message(_Assistant([_TextBlock('initial')])) + assert [e['text'] for e in events if e['event'] == 'session.text'] == ['initial'] -async def test_codex_short_message_still_streams(): - provider, events = _capturing(CodexProvider) - await provider._handle_event( - {"type": "item.completed", "item": {"type": "agent_message", "text": "ok"}} - ) - deltas = [e for e in events if e["event"] == "session.text_delta"] - assert "".join(d["text"] for d in deltas) == "ok" - assert events[-1]["event"] == "session.text" - assert events[-1]["text"] == "ok" + +async def test_claude_complete_before_stop_keeps_order_without_duplicates(): + provider, events = _capturing(ClaudeProvider) + provider._begin_stream_turn() + for text in ['first answer', 'second answer', 'final answer']: + await provider._handle_message(_Stream({'type': 'message_start', 'message': {'id': text}})) + await provider._handle_message(_start(0)) + await provider._handle_message(_delta(0, text[:3])) + await provider._handle_message(_Assistant([_TextBlock(text)])) + await provider._handle_message(_stop(0)) + texts = [e for e in events if e['event'] == 'session.text'] + assert [e['text'] for e in texts] == ['first answer', 'second answer', 'final answer'] + assert len({e['id'] for e in texts}) == 3 From 066f0c10f4ec547cdf55c4110319aa814c675bb7 Mon Sep 17 00:00:00 2001 From: Ace Data Cloud Dev Date: Sun, 4 Oct 2026 23:23:29 +0800 Subject: [PATCH 2/5] fix: omit binary media from relayed tool output --- coding_bridge/history.py | 6 ++++-- coding_bridge/protocol.py | 14 ++++++++++++++ coding_bridge/providers/claude.py | 4 ++-- tests/test_streaming.py | 15 +++++++++++++++ 4 files changed, 35 insertions(+), 4 deletions(-) diff --git a/coding_bridge/history.py b/coding_bridge/history.py index 67912e9..3b952e1 100644 --- a/coding_bridge/history.py +++ b/coding_bridge/history.py @@ -23,6 +23,8 @@ from pathlib import Path from typing import Any +from .protocol import tool_result_content + _log = logging.getLogger("coding-bridge.history") # Module-level roots so tests can point them at fixtures via monkeypatch. @@ -296,7 +298,7 @@ def _claude_user_events( { "kind": "tool_result", "tool_use_id": block.get("tool_use_id"), - "content": _stringify(block.get("content")), + "content": _stringify(tool_result_content(block.get("content"))), "is_error": bool(block.get("is_error")), "ts": ts, } @@ -329,7 +331,7 @@ def _claude_assistant_events(content: Any, ts: int | None, events: list[dict[str { "kind": "tool_result", "tool_use_id": block.get("tool_use_id"), - "content": _stringify(block.get("content")), + "content": _stringify(tool_result_content(block.get("content"))), "is_error": bool(block.get("is_error")), "ts": ts, } diff --git a/coding_bridge/protocol.py b/coding_bridge/protocol.py index 4ad0dba..3e4356a 100644 --- a/coding_bridge/protocol.py +++ b/coding_bridge/protocol.py @@ -153,3 +153,17 @@ def event_payload( payload["trace_id"] = trace_id payload.update(fields) return payload + + +def tool_result_content(content: Any) -> Any: + """Keep displayable tool output while excluding embedded binary payloads.""" + if not isinstance(content, list): + return content + visible = [] + for block in content: + if isinstance(block, dict) and block.get("type") in {"image", "audio", "document"}: + source = block.get("source") or {} + visible.append({"type": block["type"], "media_type": source.get("media_type")}) + else: + visible.append(block) + return visible diff --git a/coding_bridge/providers/claude.py b/coding_bridge/providers/claude.py index 531d5f2..7d32b68 100644 --- a/coding_bridge/providers/claude.py +++ b/coding_bridge/providers/claude.py @@ -16,7 +16,7 @@ from .. import attachments as attachment_store from .. import capabilities, claude_transcript from .. import images as image_store -from ..protocol import Event, event_payload +from ..protocol import Event, event_payload, tool_result_content from .base import slash_name if TYPE_CHECKING: @@ -680,7 +680,7 @@ async def _handle_block(self, block: Any) -> None: Event.SESSION_TOOL_RESULT, self._session_id, tool_use_id=block.tool_use_id, - content=_stringify(getattr(block, "content", None)), + content=_stringify(tool_result_content(getattr(block, "content", None))), is_error=bool(getattr(block, "is_error", False)), ) ) diff --git a/tests/test_streaming.py b/tests/test_streaming.py index 0992697..793f99b 100644 --- a/tests/test_streaming.py +++ b/tests/test_streaming.py @@ -196,3 +196,18 @@ async def test_claude_complete_before_stop_keeps_order_without_duplicates(): texts = [e for e in events if e['event'] == 'session.text'] assert [e['text'] for e in texts] == ['first answer', 'second answer', 'final answer'] assert len({e['id'] for e in texts}) == 3 + + +async def test_claude_tool_images_do_not_forward_base64_payloads(): + from types import SimpleNamespace + provider, events = _capturing(ClaudeProvider) + block = SimpleNamespace(tool_use_id='read-image', is_error=False, content=[ + {'type': 'image', 'source': { + 'type': 'base64', 'media_type': 'image/png', 'data': 'binary-secret', + }}, + {'type': 'text', 'text': 'image metadata'}, + ]) + await provider._handle_block(block) + assert 'binary-secret' not in events[-1]['content'] + assert 'image/png' in events[-1]['content'] + assert 'image metadata' in events[-1]['content'] From 1f30344bc57c2d225e8de661b756b709da6bc1dd Mon Sep 17 00:00:00 2001 From: Ace Data Cloud Dev Date: Sun, 4 Oct 2026 23:43:09 +0800 Subject: [PATCH 3/5] fix: exclude injected context from coding history titles --- coding_bridge/history.py | 2 ++ tests/test_history.py | 23 +++++++++++++++++++++++ 2 files changed, 25 insertions(+) diff --git a/coding_bridge/history.py b/coding_bridge/history.py index 3b952e1..97997d8 100644 --- a/coding_bridge/history.py +++ b/coding_bridge/history.py @@ -748,6 +748,8 @@ def _provider_title(value: Any) -> str: "# copilot instructions", "# context from my ide setup", "the following is the codex agent history", + "here is a list of plugins that are available but not installed.", + "[request interrupted by user", "the user opened the file", "the user selected the lines", "the user interrupted the previous turn", diff --git a/tests/test_history.py b/tests/test_history.py index 838c1e1..f741f38 100644 --- a/tests/test_history.py +++ b/tests/test_history.py @@ -846,3 +846,26 @@ def parse(p): path.write_text('changed') assert history._cached_summary(path, 'claude')['title'] == 'changed' assert len(calls) == 2 + + +def test_codex_history_uses_real_prompt_after_injected_plugin_catalog(tmp_path): + import json + + from coding_bridge import history + path = tmp_path / 'rollout-test.jsonl' + records = [ + {'type': 'session_meta', 'payload': {'id': 'test', 'cwd': '/fixture'}}, + {'type': 'response_item', 'payload': {'type': 'message', 'role': 'user', 'content': [ + {'type': 'input_text', 'text': ( + 'Here is a list of plugins that are available but not installed.\n- Example' + )} + ]}}, + {'type': 'response_item', 'payload': {'type': 'message', 'role': 'user', 'content': [ + {'type': 'input_text', 'text': 'Fix the calculator'} + ]}}, + ] + path.write_text('\n'.join(json.dumps(r) for r in records)) + assert history._codex_summary(path, {})['title'] == 'Fix the calculator' + header, events = history._codex_read(path, {}) + assert header['title'] == 'Fix the calculator' + assert [e['text'] for e in events if e['kind'] == 'prompt'] == ['Fix the calculator'] From 7ceaf06d16782153a5e640e443c493f9087c4427 Mon Sep 17 00:00:00 2001 From: Ace Data Cloud Dev Date: Sun, 4 Oct 2026 23:46:42 +0800 Subject: [PATCH 4/5] fix: preserve Codex patch tools in history --- coding_bridge/history.py | 6 +++--- tests/test_history.py | 16 ++++++++++++++++ 2 files changed, 19 insertions(+), 3 deletions(-) diff --git a/coding_bridge/history.py b/coding_bridge/history.py index 97997d8..3558dff 100644 --- a/coding_bridge/history.py +++ b/coding_bridge/history.py @@ -450,17 +450,17 @@ def _codex_response_event( text = _codex_reasoning_text(payload.get("summary")) if text: events.append({"kind": "thinking", "text": text, "ts": ts}) - elif ptype == "function_call": + elif ptype in ("function_call", "custom_tool_call"): events.append( { "kind": "tool_use", "tool": payload.get("name"), "tool_use_id": payload.get("call_id"), - "input": _maybe_json(payload.get("arguments")), + "input": _maybe_json(payload.get("arguments", payload.get("input"))), "ts": ts, } ) - elif ptype == "function_call_output": + elif ptype in ("function_call_output", "custom_tool_call_output"): events.append( { "kind": "tool_result", diff --git a/tests/test_history.py b/tests/test_history.py index f741f38..64800fe 100644 --- a/tests/test_history.py +++ b/tests/test_history.py @@ -869,3 +869,19 @@ def test_codex_history_uses_real_prompt_after_injected_plugin_catalog(tmp_path): header, events = history._codex_read(path, {}) assert header['title'] == 'Fix the calculator' assert [e['text'] for e in events if e['kind'] == 'prompt'] == ['Fix the calculator'] + + +def test_codex_patch_tool_is_not_lost_in_history(): + from coding_bridge import history + events = [] + header = {'title': ''} + history._codex_response_event({ + 'type': 'custom_tool_call', 'name': 'apply_patch', 'call_id': 'patch-1', + 'input': '*** Begin Patch\n*** End Patch', + }, None, events, header) + history._codex_response_event({ + 'type': 'custom_tool_call_output', 'call_id': 'patch-1', 'output': 'Success', + }, None, events, header) + assert [e['kind'] for e in events] == ['tool_use', 'tool_result'] + assert events[0]['tool'] == 'apply_patch' + assert events[1]['content'] == 'Success' From bdd0b2fcd4fdfe184e1f695f1e82aad8054b5c1c Mon Sep 17 00:00:00 2001 From: Ace Data Cloud Dev Date: Sun, 4 Oct 2026 23:59:00 +0800 Subject: [PATCH 5/5] fix: keep raw patch history input structured --- coding_bridge/history.py | 6 +++++- tests/test_history.py | 1 + 2 files changed, 6 insertions(+), 1 deletion(-) diff --git a/coding_bridge/history.py b/coding_bridge/history.py index 3558dff..1ee8ff4 100644 --- a/coding_bridge/history.py +++ b/coding_bridge/history.py @@ -456,7 +456,11 @@ def _codex_response_event( "kind": "tool_use", "tool": payload.get("name"), "tool_use_id": payload.get("call_id"), - "input": _maybe_json(payload.get("arguments", payload.get("input"))), + "input": ( + {"patch": payload.get("input")} + if ptype == "custom_tool_call" and isinstance(payload.get("input"), str) + else _maybe_json(payload.get("arguments")) + ), "ts": ts, } ) diff --git a/tests/test_history.py b/tests/test_history.py index 64800fe..0c8bcea 100644 --- a/tests/test_history.py +++ b/tests/test_history.py @@ -884,4 +884,5 @@ def test_codex_patch_tool_is_not_lost_in_history(): }, None, events, header) assert [e['kind'] for e in events] == ['tool_use', 'tool_result'] assert events[0]['tool'] == 'apply_patch' + assert events[0]['input'] == {'patch': '*** Begin Patch\n*** End Patch'} assert events[1]['content'] == 'Success'