diff --git a/.claude-plugin/plugin.json b/.claude-plugin/plugin.json index 689d499..96a32c6 100644 --- a/.claude-plugin/plugin.json +++ b/.claude-plugin/plugin.json @@ -1,7 +1,7 @@ { "name": "hermes-helmet", "displayName": "Hermes Helmet", - "version": "0.6.0", + "version": "0.7.0", "description": "First-officer skills for Hermes Helmet: setup, single-issue delivery, and dependent-issue coordination.", "author": { "name": "Machine Wisdom", diff --git a/.codex-plugin/plugin.json b/.codex-plugin/plugin.json index 74ac5e7..ca71563 100644 --- a/.codex-plugin/plugin.json +++ b/.codex-plugin/plugin.json @@ -1,6 +1,6 @@ { "name": "hermes-helmet", - "version": "0.6.0", + "version": "0.7.0", "description": "First-officer skills for Hermes Helmet: setup, single-issue delivery, and dependent-issue coordination.", "author": { "name": "Machine Wisdom", diff --git a/CHANGELOG.md b/CHANGELOG.md index ac8a0be..1f0084c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,29 @@ All notable changes to Hermes Helmet are recorded here. The project follows request on the installed host now ends with an unsupported-boundary message (no capture, dispatch or timer) unless the host advertises background delivery. +### 0.7.0 + +Claude Code record reader. + +- `helmet bridge read --session ID [--transcript PATH] [--max-bytes N]` reads + one Claude Code session's saved transcript and its subagent transcripts + read-only and prints a bounded `hermes-helmet.bridge.records/1` summary with + stable refs, origins (`person`, `hook`, `plugin`, `meta`, `assistant`, + `agent`), coverage and warnings. Thinking is dropped, every record's session + ID is verified, an incomplete final line sets `pendingTail`, and the summary + budget keeps every person prompt and assistant text while excerpting and + then dropping the oldest tool results. `--max-bytes` bounds the exact stdout + bytes. Failures exit 2 with `session-not-found`, `ambiguous-session`, + `session-mismatch`, `unsupported-format` or `unreadable`; malformed core + text blocks or an unsupported version on any record return + `unsupported-format`. +- Each ref is the record's full 32-hex UUID-derived identity, computed from that + record's own UUID only. It does not depend on other records, timestamps or + presentation order, so appends, partial-tail completion and subagent growth + never change an existing ref or `parentRef`. When one source UUID expands + into several output records, the first keeps the UUID and later blocks get a + deterministic 32-hex identity from the UUID and block index. + ### 0.5.0 - Captain’s Bridge Refresh records now rereads the exact bound chat directly diff --git a/docs/first-officer-plugins.md b/docs/first-officer-plugins.md index 05fb721..1433c95 100644 --- a/docs/first-officer-plugins.md +++ b/docs/first-officer-plugins.md @@ -37,6 +37,23 @@ is tracked in [issue #55](https://github.com/MachineWisdomAI/hermes-helmet/issues/55); the Claude plugin does not load the Bridge. +### Claude Code session records + +`helmet bridge read --session [--transcript ] [--max-bytes 600000]` +reads one Claude Code session's saved records for the Claude port of the +Bridge. It opens files read-only, finds `.jsonl` under +`$CLAUDE_CONFIG_DIR/projects` (default `~/.claude/projects`) by exact file name +only, and prints `hermes-helmet.bridge.records/1` JSON. Errors exit 2 with +`session-not-found`, `ambiguous-session`, `session-mismatch`, +`unsupported-format` or `unreadable`. Supported Claude Code versions are +2.0.0 up to (not including) 2.2.0; others are `unsupported-format`. + +Every record's `ref` (and each `parentRef`) is a full 32-hex identity derived +from that record's own source UUID, so it does not change when records are +appended, a partial final line completes, or subagent records appear. When one +source UUID expands into several output records, the first keeps the UUID-derived +ref and later blocks get a deterministic 32-hex ref from the UUID and block index. + ### Claude Code ```sh diff --git a/pyproject.toml b/pyproject.toml index a5466bd..0e154a8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "hermes-helmet" -version = "0.6.0" +version = "0.7.0" description = "Open-source software factory for coding agents: delegation, review, and repair" readme = "README.md" requires-python = ">=3.11" diff --git a/src/hermes_helmet/__init__.py b/src/hermes_helmet/__init__.py index 2adbd02..7a37429 100644 --- a/src/hermes_helmet/__init__.py +++ b/src/hermes_helmet/__init__.py @@ -7,7 +7,7 @@ from hermes_helmet.authority import Policy, load_authority, load_policy, render_crew_contract -__version__ = "0.6.0" +__version__ = "0.7.0" __all__ = [ "Policy", diff --git a/src/hermes_helmet/claude_bridge.py b/src/hermes_helmet/claude_bridge.py new file mode 100644 index 0000000..bcf53eb --- /dev/null +++ b/src/hermes_helmet/claude_bridge.py @@ -0,0 +1,710 @@ +"""Read-only Claude Code session record reader for Captain's Bridge. + +``helmet bridge read`` turns one Claude Code session's saved JSONL transcript +(plus that session's subagent transcripts) into a bounded summary of records +with stable references. Files are opened for reading only; nothing is written, +no Claude Code process is launched and no transcript is exported. + +Claude Code's transcript format is private and version dependent. The +``SUPPORTED_VERSIONS`` range and the record-type tables below are the +compatibility boundary: anything outside them is reported as +``unsupported-format`` instead of an empty success. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import re +from dataclasses import dataclass, field +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Iterator + +from hermes_helmet import __version__ + +SCHEMA = "hermes-helmet.bridge.records/1" +DEFAULT_MAX_BYTES = 600000 + +# Claude Code versions whose transcripts this reader understands (inclusive +# lower bound, exclusive upper bound). Re-verify when the range changes. +SUPPORTED_VERSIONS = ((2, 0, 0), (2, 2, 0)) + +# Record types the reader understands and must be well formed. +CORE_TYPES = frozenset({"user", "assistant", "system"}) +# Known metadata record types that carry no conversation content. They are +# skipped without being counted as unsupported. +KNOWN_NONCORE_TYPES = frozenset( + { + "summary", + "file-history-snapshot", + "queue-operation", + "progress", + "attachment", + "last-prompt", + "custom-title", + "ai-title", + "agent-name", + "mode", + "permission-mode", + "pr-link", + "tag", + "content-replacement", + } +) +# Content blocks that are model reasoning; dropped entirely. +REASONING_BLOCKS = frozenset({"thinking", "redacted_thinking", "reasoning"}) + +ERROR_CODES = ( + "session-not-found", + "ambiguous-session", + "session-mismatch", + "unsupported-format", + "unreadable", +) + +SESSION_ID_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,127}") +VERSION_RE = re.compile(r"(\d+)\.(\d+)\.(\d+)") +UUID_RE = re.compile(r"[0-9a-fA-F][0-9a-fA-F-]{7,63}") + +# Per-text read-time caps so a single enormous tool result cannot exhaust +# memory. Person prompts and assistant text are never capped. +TOOL_RESULT_READ_CAP = 20000 +TOOL_INPUT_CAP = 2000 +# Head/tail excerpt applied to tool results when the budget is exceeded. +EXCERPT_HEAD = 600 +EXCERPT_TAIL = 400 + +_HOOK_PREFIXES = ("", "") +_META_PREFIXES = ("", "") + + +class BridgeReadError(Exception): + def __init__(self, code: str, message: str) -> None: + super().__init__(message) + self.code = code + self.message = message + + +def error_payload(code: str, message: str) -> dict[str, Any]: + return {"schema": SCHEMA, "error": {"code": code, "message": message}} + + +@dataclass +class _Rec: + uuid: str + parent_uuid: str | None + agent_id: str | None + timestamp: str | None + role: str + origin: str + text: str + tool: str | None = None + tool_use_id: str | None = None + is_error: bool = False + truncated: bool = False + orig_bytes: int = 0 + primary_uuid: str | None = None # set on extra blocks of one record + order: int = 0 + ref: str = "" + + +@dataclass +class _State: + records: list[_Rec] = field(default_factory=list) + parent_of: dict[str, str | None] = field(default_factory=dict) + tool_names: dict[str, str] = field(default_factory=dict) + tool_use_uuid: dict[str, str] = field(default_factory=dict) + agent_tool_use: dict[str, str] = field(default_factory=dict) + unsupported: int = 0 + unsupported_types: set[str] = field(default_factory=set) + pending_tail: bool = False + last_main_uuid: str | None = None + counter: int = 0 + + +def claude_config_dir() -> Path: + env = os.environ.get("CLAUDE_CONFIG_DIR") + return Path(env) if env else Path.home() / ".claude" + + +def _utc_now() -> str: + return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +# -------------------------------------------------------------------------- +# Locating files +# -------------------------------------------------------------------------- + + +def find_session_file(session_id: str, projects_dir: Path) -> Path: + """Find the one transcript named exactly ``.jsonl``.""" + if not SESSION_ID_RE.fullmatch(session_id): + raise BridgeReadError( + "session-not-found", "The session id is not a valid Claude Code session id." + ) + matches: list[Path] = [] + try: + with os.scandir(projects_dir) as entries: + for entry in sorted(entries, key=lambda item: item.name): + try: + if not entry.is_dir(follow_symlinks=False): + continue + except OSError: + continue + candidate = Path(entry.path) / f"{session_id}.jsonl" + if candidate.is_file(): + matches.append(candidate) + except FileNotFoundError: + raise BridgeReadError( + "session-not-found", + "No Claude Code projects directory was found, so the session " + "transcript cannot be located. Pass --transcript with its path.", + ) from None + except OSError as exc: + raise BridgeReadError( + "unreadable", f"The Claude Code projects directory is not readable: {exc.strerror or exc}" + ) from exc + if not matches: + raise BridgeReadError( + "session-not-found", + f"No transcript named {session_id}.jsonl was found under the Claude " + "Code projects directory.", + ) + if len(matches) > 1: + raise BridgeReadError( + "ambiguous-session", + f"{len(matches)} transcripts named {session_id}.jsonl were found; " + "pass --transcript to choose one explicitly.", + ) + return matches[0] + + +def _subagent_files(transcript: Path, session_id: str) -> list[Path]: + folder = transcript.parent / session_id / "subagents" + try: + with os.scandir(folder) as entries: + names = sorted( + entry.name + for entry in entries + if entry.name.endswith(".jsonl") and entry.is_file() + ) + except (FileNotFoundError, NotADirectoryError): + return [] + except OSError as exc: + raise BridgeReadError( + "unreadable", f"The subagent transcripts are not readable: {exc.strerror or exc}" + ) from exc + return [folder / name for name in names] + + +# -------------------------------------------------------------------------- +# Parsing +# -------------------------------------------------------------------------- + + +def _iter_lines(path: Path) -> Iterator[tuple[int, bytes, bool]]: + """Yield (line number, raw line, terminated) streaming, read-only.""" + try: + with open(path, "rb") as handle: + number = 0 + for raw in handle: + number += 1 + terminated = raw.endswith(b"\n") + yield number, raw.rstrip(b"\r\n"), terminated + except FileNotFoundError: + raise BridgeReadError("session-not-found", f"Transcript {path.name} does not exist.") from None + except OSError as exc: + raise BridgeReadError( + "unreadable", f"Transcript {path.name} is not readable: {exc.strerror or exc}" + ) from exc + + +def _check_version(value: Any, where: str) -> str: + if not isinstance(value, str) or not VERSION_RE.fullmatch(value): + raise BridgeReadError( + "unsupported-format", + f"{where} has no recognizable Claude Code version (found {value!r}).", + ) + parts = tuple(int(part) for part in VERSION_RE.fullmatch(value).groups()) # type: ignore[union-attr] + low, high = SUPPORTED_VERSIONS + if not (low <= parts < high): + raise BridgeReadError( + "unsupported-format", + f"Claude Code version {value} is outside the supported range " + f"{'.'.join(map(str, low))} to below {'.'.join(map(str, high))}.", + ) + return value + + +def _clip(text: str, cap: int) -> tuple[str, bool]: + if len(text) <= cap: + return text, False + head = cap * 3 // 5 + tail = cap - head + omitted = len(text) - head - tail + return f"{text[:head]}\n…[{omitted} characters omitted]…\n{text[-tail:]}", True + + +def _blocks(content: Any, where: str) -> list[dict[str, Any]]: + if isinstance(content, str): + return [{"type": "text", "text": content}] + if isinstance(content, list): + out = [] + for block in content: + if isinstance(block, str): + out.append({"type": "text", "text": block}) + elif isinstance(block, dict): + out.append(block) + else: + raise BridgeReadError("unsupported-format", f"{where} has a malformed content block.") + return out + raise BridgeReadError("unsupported-format", f"{where} has malformed message content.") + + +def _text_field(block: dict[str, Any], where: str, version: str) -> str: + text = block.get("text") + if not isinstance(text, str): + raise BridgeReadError( + "unsupported-format", + f"{where} (Claude Code {version}) has a text block whose text is not a string.", + ) + return text + + +def _result_text(content: Any) -> str: + if isinstance(content, str): + return content + if isinstance(content, list): + parts = [] + for item in content: + if isinstance(item, str): + parts.append(item) + elif isinstance(item, dict) and item.get("type") == "text": + parts.append(str(item.get("text", ""))) + return "\n".join(parts) + return "" if content is None else json.dumps(content, ensure_ascii=False) + + +def _user_origin(record: dict[str, Any], text: str, sidechain: bool) -> str: + """Classify a user-role text record from recorded evidence only.""" + if sidechain: + return "agent" + marker = record.get("origin") + kind = marker.get("kind") if isinstance(marker, dict) else marker + if isinstance(kind, str): + lowered = kind.lower() + if lowered in {"human", "user", "person"}: + return "person" + if lowered == "hook": + return "hook" + if lowered == "plugin": + return "plugin" + return "meta" + if record.get("isMeta") is True or record.get("isCompactSummary") is True: + return "meta" + if record.get("pluginName") or record.get("plugin"): + return "plugin" + stripped = text.lstrip() + if stripped.startswith(_HOOK_PREFIXES): + return "hook" + if stripped.startswith(_META_PREFIXES): + return "meta" + return "person" + + +def _scan_file( + path: Path, + session_id: str, + state: _State, + *, + agent_id: str | None, +) -> None: + sidechain = agent_id is not None + where_base = path.name + for number, raw, terminated in _iter_lines(path): + if not raw.strip(): + continue + where = f"{where_base} line {number}" + try: + record = json.loads(raw.decode("utf-8")) + except (UnicodeDecodeError, ValueError): + if not terminated: + state.pending_tail = True + continue + raise BridgeReadError( + "unsupported-format", f"{where} is not a valid JSON record." + ) from None + if not isinstance(record, dict): + raise BridgeReadError("unsupported-format", f"{where} is not a JSON object record.") + rtype = record.get("type") + if not isinstance(rtype, str) or not rtype: + raise BridgeReadError("unsupported-format", f"{where} has no record type.") + sid = record.get("sessionId") + if sid is not None and sid != session_id: + raise BridgeReadError( + "session-mismatch", + f"{where} belongs to session {sid!r}, not the requested session {session_id!r}.", + ) + if rtype not in CORE_TYPES: + if record.get("version") is not None: + _check_version(record["version"], where) + if rtype not in KNOWN_NONCORE_TYPES: + state.unsupported += 1 + state.unsupported_types.add(rtype) + continue + version = _check_version(record.get("version"), where) + if sid is None: + raise BridgeReadError( + "unsupported-format", + f"{where} (Claude Code {version}) has no session id.", + ) + raw_uuid = record.get("uuid") + if not isinstance(raw_uuid, str) or not UUID_RE.fullmatch(raw_uuid): + raise BridgeReadError( + "unsupported-format", f"{where} (Claude Code {version}) has no valid uuid." + ) + uuid: str = raw_uuid.replace("-", "").lower() + parent = record.get("parentUuid") or record.get("logicalParentUuid") + parent = parent.replace("-", "").lower() if isinstance(parent, str) and parent else None + state.parent_of[uuid] = parent + timestamp = record.get("timestamp") + timestamp = timestamp if isinstance(timestamp, str) else None + rec_agent = record.get("agentId") if isinstance(record.get("agentId"), str) else None + agent = agent_id or rec_agent if sidechain else None + if not sidechain: + state.last_main_uuid = uuid + + emitted: list[_Rec] = [] + + def add(role: str, origin: str, text: str, **extra: Any) -> None: + emitted.append( + _Rec( + uuid=uuid, + parent_uuid=parent, + agent_id=agent, + timestamp=timestamp, + role=role, + origin=origin, + text=text, + orig_bytes=len(text.encode("utf-8")), + **extra, + ) + ) + + if rtype == "system": + content = record.get("content") + if content is not None and not isinstance(content, str): + raise BridgeReadError( + "unsupported-format", f"{where} (Claude Code {version}) has malformed system content." + ) + subtype = record.get("subtype") + text = content or (str(subtype) if subtype else "") + if subtype == "compact_boundary" and not content: + text = "Conversation compacted" + add("system", "agent" if sidechain else "meta", text) + else: + message = record.get("message") + if not isinstance(message, dict) or "content" not in message: + raise BridgeReadError( + "unsupported-format", + f"{where} (Claude Code {version}) has no message content.", + ) + for block in _blocks(message["content"], where): + btype = block.get("type") + if btype in REASONING_BLOCKS: + continue + if rtype == "assistant": + origin = "agent" if sidechain else "assistant" + if btype == "text": + add("assistant", origin, _text_field(block, where, version)) + elif btype == "tool_use": + name = block.get("name") + if not isinstance(name, str) or not name: + raise BridgeReadError( + "unsupported-format", + f"{where} (Claude Code {version}) has a tool_use block without a name.", + ) + tool_input = block.get("input", {}) + if not isinstance(tool_input, (dict, list, str)): + raise BridgeReadError( + "unsupported-format", + f"{where} (Claude Code {version}) has a malformed tool_use input.", + ) + tool_use_id = block.get("id") + tool_use_id = tool_use_id if isinstance(tool_use_id, str) else None + full_input = json.dumps(tool_input, ensure_ascii=False, sort_keys=True) + text, clipped = _clip(full_input, TOOL_INPUT_CAP) + if tool_use_id: + state.tool_names[tool_use_id] = name + add( + "assistant", + origin, + text, + tool=name, + tool_use_id=tool_use_id, + truncated=clipped, + ) + emitted[-1].orig_bytes = len(full_input.encode("utf-8")) + if emitted[-1:] and tool_use_id: + state.tool_use_uuid.setdefault(tool_use_id, uuid) + else: # user role + if btype == "tool_result": + tool_use_id = block.get("tool_use_id") + tool_use_id = tool_use_id if isinstance(tool_use_id, str) else None + raw_result = block.get("content") + if raw_result is not None and not isinstance(raw_result, (str, list)): + raise BridgeReadError( + "unsupported-format", + f"{where} (Claude Code {version}) has malformed tool_result content.", + ) + if isinstance(raw_result, list) and not all( + isinstance(item, str) + or (isinstance(item, dict) and (item.get("type") != "text" or isinstance(item.get("text"), str))) + for item in raw_result + ): + raise BridgeReadError( + "unsupported-format", + f"{where} (Claude Code {version}) has malformed tool_result content.", + ) + full = _result_text(raw_result) + text, clipped = _clip(full, TOOL_RESULT_READ_CAP) + add( + "tool_result", + "agent" if sidechain else "assistant", + text, + tool=state.tool_names.get(tool_use_id or ""), + tool_use_id=tool_use_id, + is_error=block.get("is_error") is True, + truncated=clipped, + ) + emitted[-1].orig_bytes = len(full.encode("utf-8")) + result = record.get("toolUseResult") + if ( + tool_use_id + and isinstance(result, dict) + and isinstance(result.get("agentId"), str) + ): + state.agent_tool_use[result["agentId"]] = tool_use_id + elif btype == "text": + text = _text_field(block, where, version) + add("user", _user_origin(record, text, sidechain), text) + for index, rec in enumerate(emitted): + rec.ref = _record_ref(uuid, index) + if index: + rec.primary_uuid = rec.uuid + rec.uuid = rec.ref + rec.order = state.counter + state.counter += 1 + state.records.append(rec) + + +# -------------------------------------------------------------------------- +# Refs and budget +# -------------------------------------------------------------------------- + + +def _record_ref(uuid: str, index: int) -> str: + """Full 32-hex ref derived only from the record's own source UUID. + + A 32-hex UUID is its own ref. Any other accepted UUID spelling is hashed to + 32 hex. Additional output blocks of one source UUID get a deterministic + 32-hex identity from the UUID and the block index. No other record, the + timestamp order or the presentation order takes part, so refs never change + when records are appended or discovered later. + """ + base = uuid if re.fullmatch(r"[0-9a-f]{32}", uuid) else hashlib.sha256(uuid.encode()).hexdigest()[:32] + if index == 0: + return base + return hashlib.sha256(f"{base}:{index}".encode()).hexdigest()[:32] + + +def _assign_refs(records: list[_Rec]) -> dict[int, str]: + return {id(rec): rec.ref for rec in records} + + +def _record_dict(rec: _Rec, ref: str, parent_ref: str | None) -> dict[str, Any]: + return { + "ref": ref, + "parentRef": parent_ref, + "agentId": rec.agent_id, + "timestamp": rec.timestamp, + "role": rec.role, + "origin": rec.origin, + "text": rec.text, + "tool": rec.tool, + "toolUseId": rec.tool_use_id, + "isError": rec.is_error, + "truncated": rec.truncated, + } + + +def render(payload: dict[str, Any]) -> str: + """The exact text ``helmet bridge read`` prints, newline included.""" + return json.dumps(payload, indent=2, ensure_ascii=False) + "\n" + + +def _excerpt(text: str) -> str: + if len(text) <= EXCERPT_HEAD + EXCERPT_TAIL: + return text + omitted = len(text) - EXCERPT_HEAD - EXCERPT_TAIL + return f"{text[:EXCERPT_HEAD]}\n…[{omitted} characters omitted]…\n{text[-EXCERPT_TAIL:]}" + + +def _sort_key(rec: _Rec) -> tuple[str, int]: + return (rec.timestamp or "", rec.order) + + +def read_session( + session_id: str, + *, + transcript: Path | None = None, + max_bytes: int = DEFAULT_MAX_BYTES, + claude_home: Path | None = None, +) -> dict[str, Any]: + """Return the ``hermes-helmet.bridge.records/1`` payload or raise.""" + if transcript is None: + home = claude_home if claude_home is not None else claude_config_dir() + path = find_session_file(session_id, home / "projects") + else: + path = Path(transcript) + if not path.is_file(): + raise BridgeReadError( + "session-not-found", f"Transcript {path.name} does not exist or is not a file." + ) + state = _State() + _scan_file(path, session_id, state, agent_id=None) + main_count = len(state.records) + for sub in _subagent_files(path, session_id): + agent = sub.stem[len("agent-") :] if sub.stem.startswith("agent-") else sub.stem + _scan_file(sub, session_id, state, agent_id=agent) + if main_count == 0: + raise BridgeReadError( + "unsupported-format", + "The transcript has no conversation records this reader recognizes.", + ) + + records = sorted(state.records, key=_sort_key) + refs = _assign_refs(records) + ref_by_uuid: dict[str, str] = {} + for rec in records: + if rec.primary_uuid is None: + ref_by_uuid[rec.uuid] = rec.ref + first_ref_of_agent: dict[str, str] = {} + for rec in records: + if rec.agent_id: + first_ref_of_agent.setdefault(rec.agent_id, refs[id(rec)]) + + def parent_ref_for(rec: _Rec, included: set[str] | None) -> str | None: + if rec.primary_uuid: + return ref_by_uuid.get(rec.primary_uuid) + current = rec.parent_uuid + seen: set[str] = set() + while current and current not in seen: + seen.add(current) + candidate = ref_by_uuid.get(current) + if candidate and (included is None or candidate in included): + return candidate + current = state.parent_of.get(current) + if rec.agent_id and rec.parent_uuid is None: + tool_use_id = state.agent_tool_use.get(rec.agent_id) + origin_uuid = state.tool_use_uuid.get(tool_use_id or "") + if origin_uuid: + candidate = ref_by_uuid.get(origin_uuid) + if candidate and (included is None or candidate in included): + return candidate + return None + + total = len(records) + protected = {"person", "assistant", "agent"} + original_bytes = sum(rec.orig_bytes for rec in records) + last_uuid = state.last_main_uuid or (records[-1].primary_uuid or records[-1].uuid) + texts = {id(rec): rec.text for rec in records} + truncated = {id(rec): rec.truncated for rec in records} + extra_warnings: list[dict[str, str]] = [] + if state.unsupported: + extra_warnings.append( + { + "code": "unsupported-records-skipped", + "message": ( + f"Skipped {state.unsupported} records of unknown type: " + + ", ".join(sorted(state.unsupported_types)) + + "." + ), + } + ) + if state.pending_tail: + extra_warnings.append( + { + "code": "pending-tail", + "message": "An incomplete final line was ignored; Claude Code may still be writing.", + } + ) + + def build(keep: list[_Rec], budget_warnings: list[dict[str, str]]) -> dict[str, Any]: + included = {refs[id(rec)] for rec in keep} + out_records = [] + for rec in keep: + item = _record_dict(rec, refs[id(rec)], parent_ref_for(rec, included)) + item["text"] = texts[id(rec)] + item["truncated"] = truncated[id(rec)] + out_records.append(item) + final_bytes = sum(len(item["text"].encode("utf-8")) for item in out_records) + return { + "schema": SCHEMA, + "helmetVersion": __version__, + "sessionId": session_id, + "readAt": _utc_now(), + "fingerprint": f"{total}:{last_uuid}", + "records": out_records, + "coverage": { + "recordsTotal": total, + "recordsIncluded": len(out_records), + "bytesOmitted": max(0, original_bytes - final_bytes), + "unsupportedSkipped": state.unsupported, + "pendingTail": state.pending_tail, + }, + "warnings": budget_warnings + extra_warnings, + } + + def measure(keep: list[_Rec]) -> int: + return len(render(build(keep, [])).encode("utf-8")) + + keep = list(records) + budget_warnings: list[dict[str, str]] = [] + if measure(keep) > max_bytes: + for rec in keep: + if rec.role == "tool_result" and len(texts[id(rec)]) > EXCERPT_HEAD + EXCERPT_TAIL: + texts[id(rec)] = _excerpt(texts[id(rec)]) + truncated[id(rec)] = True + # Drop the oldest tool results until the serialized output fits. + droppable = [rec for rec in records if rec.role == "tool_result"] + low, high = 0, len(droppable) # smallest count of oldest results to drop + while low < high: + mid = (low + high) // 2 + gone = {id(rec) for rec in droppable[:mid]} + if measure([rec for rec in records if id(rec) not in gone]) <= max_bytes: + high = mid + else: + low = mid + 1 + gone = {id(rec) for rec in droppable[:low]} + keep = [rec for rec in records if id(rec) not in gone] + running = measure(keep) + if running > max_bytes: + protected_bytes = sum( + len(texts[id(rec)].encode("utf-8")) + for rec in keep + if rec.origin in protected and rec.role in {"user", "assistant"} + ) + budget_warnings.append( + { + "code": "budget-conflict", + "message": ( + f"Person prompts and assistant text ({protected_bytes} bytes) are " + f"protected and are returned in full, so the output is {running} " + f"bytes, over --max-bytes {max_bytes}." + ), + } + ) + return build(keep, budget_warnings) diff --git a/src/hermes_helmet/cli.py b/src/hermes_helmet/cli.py index ff39352..ffc693c 100644 --- a/src/hermes_helmet/cli.py +++ b/src/hermes_helmet/cli.py @@ -431,6 +431,27 @@ def cmd_prepare_repo(args: argparse.Namespace) -> int: return 0 +def cmd_bridge_read(args: argparse.Namespace) -> int: + """Read one Claude Code session's records (read-only); JSON on stdout.""" + from hermes_helmet import claude_bridge + + try: + if args.max_bytes < 1: + raise claude_bridge.BridgeReadError( + "unreadable", "--max-bytes must be a positive integer." + ) + payload = claude_bridge.read_session( + args.session, + transcript=args.transcript, + max_bytes=args.max_bytes, + ) + except claude_bridge.BridgeReadError as exc: + print(json.dumps(claude_bridge.error_payload(exc.code, exc.message), indent=2)) + return 2 + sys.stdout.write(claude_bridge.render(payload)) + return 0 + + def cmd_install_skills(args: argparse.Namespace) -> int: try: results = install_skills( @@ -1344,6 +1365,33 @@ def build_parser() -> argparse.ArgumentParser: ) prep_p.set_defaults(func=cmd_prepare_repo) + bridge_p = sub.add_parser( + "bridge", + help="Captain's Bridge record helpers (read-only)", + ) + bridge_sub = bridge_p.add_subparsers(dest="bridge_command", required=True) + bridge_read = bridge_sub.add_parser( + "read", + help=( + "Read one Claude Code session's saved records as a bounded JSON " + "summary (never writes, launches Claude Code or exports)" + ), + ) + bridge_read.add_argument("--session", required=True, help="Exact Claude Code session id") + bridge_read.add_argument( + "--transcript", + type=Path, + default=None, + help="Explicit session transcript path (else found by exact session id)", + ) + bridge_read.add_argument( + "--max-bytes", + type=int, + default=600000, + help="Summary budget in bytes (default: 600000)", + ) + bridge_read.set_defaults(func=cmd_bridge_read) + setup_p = sub.add_parser( "setup", help="Create or resume a validated adopter-owned Hermes Helmet installation", diff --git a/tests/test_claude_bridge_read.py b/tests/test_claude_bridge_read.py new file mode 100644 index 0000000..86767c0 --- /dev/null +++ b/tests/test_claude_bridge_read.py @@ -0,0 +1,751 @@ +#!/usr/bin/env python3 +"""``helmet bridge read``: Claude Code session records through the CLI seam. + +All transcripts are synthetic and generated here. Tests drive argv, stdout and +the exit code of ``hermes_helmet.cli.main``. +""" + +from __future__ import annotations + +import contextlib +import io +import json +import os +import sys +import tempfile +import unittest +import uuid as uuidlib +from pathlib import Path +from unittest import mock + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / "src")) + +from hermes_helmet import __version__ # noqa: E402 +from hermes_helmet import cli # noqa: E402 + +SESSION = "11111111-2222-4333-8444-555555555555" +VERSION = "2.1.296" + + +def make_uuid(n: int) -> str: + return str(uuidlib.UUID(int=0x4F1C2A9E << 96 | n)) + + +class Transcript: + """Builds synthetic Claude Code JSONL records with a running clock.""" + + def __init__(self, session: str = SESSION, version: str = VERSION) -> None: + self.session = session + self.version = version + self.counter = 0 + self.last: str | None = None + self.lines: list[str] = [] + + def _base(self, rtype: str, **extra) -> dict: + self.counter += 1 + uid = make_uuid(self.counter) + record = { + "type": rtype, + "uuid": uid, + "parentUuid": self.last, + "sessionId": self.session, + "version": self.version, + "timestamp": f"2026-10-09T16:{self.counter // 60:02d}:{self.counter % 60:02d}Z", + } + record.update(extra) + self.last = uid + return record + + def add(self, record: dict) -> dict: + self.lines.append(json.dumps(record)) + return record + + def user(self, text, **extra): + return self.add(self._base("user", message={"role": "user", "content": text}, **extra)) + + def assistant(self, blocks): + if isinstance(blocks, str): + blocks = [{"type": "text", "text": blocks}] + return self.add(self._base("assistant", message={"role": "assistant", "content": blocks})) + + def tool_use(self, tool_use_id, name, tool_input): + return self.assistant([{"type": "tool_use", "id": tool_use_id, "name": name, "input": tool_input}]) + + def tool_result(self, tool_use_id, text, is_error=False, **extra): + return self.add( + self._base( + "user", + message={ + "role": "user", + "content": [ + {"type": "tool_result", "tool_use_id": tool_use_id, "content": text, "is_error": is_error} + ], + }, + **extra, + ) + ) + + def text(self, trailing_newline=True, partial: str = "") -> str: + body = "\n".join(self.lines) + return body + ("\n" if trailing_newline else "") + partial + + +class BridgeReadTests(unittest.TestCase): + def setUp(self) -> None: + self._tmp = tempfile.TemporaryDirectory() + self.addCleanup(self._tmp.cleanup) + self.home = Path(self._tmp.name) / "claude-home" + self.projects = self.home / "projects" + self.project = self.projects / "-synthetic-project" + self.project.mkdir(parents=True) + env = mock.patch.dict(os.environ, {"CLAUDE_CONFIG_DIR": str(self.home)}) + env.start() + self.addCleanup(env.stop) + + # helpers --------------------------------------------------------- + def write(self, transcript: Transcript, *, project: Path | None = None, session: str = SESSION, **kw) -> Path: + path = (project or self.project) / f"{session}.jsonl" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(transcript.text(**kw), encoding="utf-8") + return path + + def run_cli(self, *args: str) -> tuple[int, dict]: + out, err = io.StringIO(), io.StringIO() + with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err): + code = cli.main(["bridge", "read", *args]) + self.assertEqual(err.getvalue(), "") + return code, json.loads(out.getvalue()) + + def read_ok(self, *args: str) -> dict: + code, payload = self.run_cli("--session", SESSION, *args) + self.assertEqual(code, 0, payload) + return payload + + def assert_error(self, payload: dict, code: str) -> str: + self.assertEqual(payload["schema"], "hermes-helmet.bridge.records/1") + self.assertEqual(payload["error"]["code"], code) + self.assertNotIn("records", payload) + return payload["error"]["message"] + + def basic(self) -> Transcript: + t = Transcript() + t.user("Please fix the parser.") + t.assistant([ + {"type": "thinking", "thinking": "SECRET-REASONING-ONE", "signature": "x"}, + {"type": "text", "text": "I will run the tests."}, + ]) + t.tool_use("toolu_1", "Bash", {"command": "pytest"}) + t.tool_result("toolu_1", "3 passed") + t.assistant("Done.") + return t + + # documented output ------------------------------------------------- + def test_documented_output_shape(self) -> None: + self.write(self.basic()) + payload = self.read_ok() + self.assertEqual(payload["schema"], "hermes-helmet.bridge.records/1") + self.assertEqual(payload["helmetVersion"], __version__) + self.assertEqual(payload["sessionId"], SESSION) + self.assertRegex(payload["readAt"], r"^\d{4}-\d\d-\d\dT\d\d:\d\d:\d\dZ$") + self.assertEqual(payload["fingerprint"], f"5:{make_uuid(5).replace('-', '')}") + self.assertEqual(payload["warnings"], []) + self.assertEqual( + payload["coverage"], + {"recordsTotal": 5, "recordsIncluded": 5, "bytesOmitted": 0, + "unsupportedSkipped": 0, "pendingTail": False}, + ) + records = payload["records"] + self.assertEqual( + [(r["role"], r["origin"]) for r in records], + [("user", "person"), ("assistant", "assistant"), ("assistant", "assistant"), + ("tool_result", "assistant"), ("assistant", "assistant")], + ) + keys = {"ref", "parentRef", "agentId", "timestamp", "role", "origin", "text", + "tool", "toolUseId", "isError", "truncated"} + for record in records: + self.assertEqual(set(record), keys) + self.assertRegex(record["ref"], r"^[0-9a-f]{32}$") + refs = [r["ref"] for r in records] + self.assertEqual(len(set(refs)), len(refs)) + self.assertIsNone(records[0]["parentRef"]) + self.assertEqual(records[1]["parentRef"], refs[0]) + call, result = records[2], records[3] + self.assertEqual((call["tool"], call["toolUseId"]), ("Bash", "toolu_1")) + self.assertEqual(json.loads(call["text"]), {"command": "pytest"}) + self.assertEqual((result["tool"], result["toolUseId"], result["text"]), ("Bash", "toolu_1", "3 passed")) + self.assertFalse(result["isError"]) + self.assertEqual(records[0]["timestamp"], "2026-10-09T16:00:01Z") + + def test_tool_error_flag(self) -> None: + t = Transcript() + t.user("go") + t.tool_use("toolu_9", "Bash", {"command": "false"}) + t.tool_result("toolu_9", "boom", is_error=True) + self.write(t) + self.assertTrue(self.read_ok()["records"][-1]["isError"]) + + def test_thinking_is_absent(self) -> None: + t = self.basic() + t.assistant([{"type": "redacted_thinking", "data": "SECRET-REASONING-TWO"}]) + self.write(t) + raw = json.dumps(self.read_ok()) + self.assertNotIn("SECRET-REASONING", raw) + self.assertNotIn("thinking", raw) + + # subagents ---------------------------------------------------------- + def test_subagent_transcripts_carry_agent_ids_and_parents(self) -> None: + t = Transcript() + t.user("delegate") + t.tool_use("toolu_task", "Task", {"prompt": "look"}) + t.tool_result("toolu_task", "found it", toolUseResult={"agentId": "a1b2c3"}) + self.write(t) + sub = Transcript() + sub.counter = 100 + sub.user("look", isSidechain=True, agentId="a1b2c3", parentUuid=None) + sub.assistant("Searching.") + sub_lines = [json.loads(line) for line in sub.lines] + for line in sub_lines: + line["isSidechain"] = True + line["agentId"] = "a1b2c3" + sub_lines[0]["parentUuid"] = None + folder = self.project / SESSION / "subagents" + folder.mkdir(parents=True) + (folder / "agent-a1b2c3.jsonl").write_text( + "\n".join(json.dumps(line) for line in sub_lines) + "\n", encoding="utf-8" + ) + payload = self.read_ok() + agents = [r for r in payload["records"] if r["agentId"]] + self.assertEqual(len(agents), 2) + self.assertEqual({r["agentId"] for r in agents}, {"a1b2c3"}) + self.assertEqual({r["origin"] for r in agents}, {"agent"}) + task_call = next(r for r in payload["records"] if r["tool"] == "Task" and not r["agentId"]) + self.assertEqual(agents[0]["parentRef"], task_call["ref"]) + self.assertEqual(agents[1]["parentRef"], agents[0]["ref"]) + self.assertEqual(payload["coverage"]["recordsTotal"], 5) + + # compaction --------------------------------------------------------- + def test_records_on_both_sides_of_compaction_are_kept(self) -> None: + t = Transcript() + t.user("before compaction") + t.assistant("early answer") + t.add(t._base("system", subtype="compact_boundary", content="Conversation compacted", + compactMetadata={"trigger": "auto"})) + t.user("This session is being continued from a previous conversation...", isCompactSummary=True) + t.assistant("late answer") + self.write(t) + payload = self.read_ok() + texts = [r["text"] for r in payload["records"]] + self.assertIn("before compaction", texts) + self.assertIn("early answer", texts) + self.assertIn("late answer", texts) + self.assertIn("Conversation compacted", texts) + summary = next(r for r in payload["records"] if r["text"].startswith("This session is being")) + self.assertEqual(summary["origin"], "meta") + self.assertEqual(payload["coverage"]["recordsTotal"], 5) + + # partial tail ------------------------------------------------------- + def test_incomplete_final_line_is_ignored(self) -> None: + self.write(self.basic(), partial='{"type":"assistant","uuid":"abc') + payload = self.read_ok() + self.assertTrue(payload["coverage"]["pendingTail"]) + self.assertEqual(payload["coverage"]["recordsTotal"], 5) + self.assertTrue(any(w["code"] == "pending-tail" for w in payload["warnings"])) + + def test_complete_unterminated_final_line_is_read(self) -> None: + self.write(self.basic(), trailing_newline=False) + payload = self.read_ok() + self.assertFalse(payload["coverage"]["pendingTail"]) + self.assertEqual(payload["coverage"]["recordsTotal"], 5) + + def test_corrupt_middle_line_is_unsupported_format(self) -> None: + t = self.basic() + t.lines.insert(2, "{not json") + self.write(t) + code, payload = self.run_cli("--session", SESSION) + self.assertEqual(code, 2) + self.assert_error(payload, "unsupported-format") + + # origins ------------------------------------------------------------ + def test_injected_user_records_are_not_person(self) -> None: + t = Transcript() + t.user("typed by the Captain") + t.user("hook says hi") + t.user("injected by a plugin", origin={"kind": "plugin"}) + t.user("hook output", origin={"kind": "hook"}) + t.user("caveat text", isMeta=True) + t.user("/clear") + t.user("explicit human", origin={"kind": "human"}) + t.assistant("ok") + self.write(t) + origins = [r["origin"] for r in self.read_ok()["records"]] + self.assertEqual(origins, ["person", "hook", "plugin", "hook", "meta", "meta", "person", "assistant"]) + + # identity ----------------------------------------------------------- + def test_session_mismatch_inside_file(self) -> None: + t = self.basic() + t.session = "99999999-2222-4333-8444-555555555555" + t.user("someone else") + self.write(t) + code, payload = self.run_cli("--session", SESSION) + self.assertEqual(code, 2) + message = self.assert_error(payload, "session-mismatch") + self.assertIn(SESSION, message) + + def test_session_mismatch_with_explicit_transcript(self) -> None: + other = "99999999-2222-4333-8444-555555555555" + path = self.write(Transcript(session=other), session=other) + t = Transcript(session=other) + t.user("x") + path.write_text(t.text(), encoding="utf-8") + code, payload = self.run_cli("--session", SESSION, "--transcript", str(path)) + self.assertEqual(code, 2) + self.assert_error(payload, "session-mismatch") + + def test_zero_matches_is_session_not_found_even_with_newer_unrelated_file(self) -> None: + other = "99999999-2222-4333-8444-555555555555" + t = Transcript(session=other) + t.user("newer and unrelated") + self.write(t, session=other) + code, payload = self.run_cli("--session", SESSION) + self.assertEqual(code, 2) + self.assert_error(payload, "session-not-found") + + def test_missing_projects_directory_is_session_not_found(self) -> None: + with mock.patch.dict(os.environ, {"CLAUDE_CONFIG_DIR": str(self.home / "absent")}): + code, payload = self.run_cli("--session", SESSION) + self.assertEqual(code, 2) + self.assert_error(payload, "session-not-found") + + def test_two_exact_matches_is_ambiguous(self) -> None: + self.write(self.basic()) + self.write(self.basic(), project=self.projects / "-another-project") + code, payload = self.run_cli("--session", SESSION) + self.assertEqual(code, 2) + self.assert_error(payload, "ambiguous-session") + + def test_explicit_transcript_resolves_ambiguity(self) -> None: + chosen = self.write(self.basic()) + self.write(self.basic(), project=self.projects / "-another-project") + payload = self.read_ok("--transcript", str(chosen)) + self.assertEqual(payload["coverage"]["recordsTotal"], 5) + + def test_exact_filename_only_and_newer_file_not_chosen(self) -> None: + self.write(self.basic()) + newer = Transcript() + newer.user("a decoy that names the session in its content: " + SESSION) + decoy = self.project / f"{SESSION}-copy.jsonl" + decoy.write_text(newer.text(), encoding="utf-8") + os.utime(decoy, (4102444800, 4102444800)) + payload = self.read_ok() + self.assertEqual(payload["records"][0]["text"], "Please fix the parser.") + + def test_missing_explicit_transcript_is_not_found(self) -> None: + code, payload = self.run_cli("--session", SESSION, "--transcript", str(self.home / "nope.jsonl")) + self.assertEqual(code, 2) + self.assert_error(payload, "session-not-found") + + def test_unreadable_transcript(self) -> None: + path = self.write(self.basic()) + real_open = open + + def fake_open(file, *args, **kwargs): + if str(file) == str(path): + raise PermissionError(13, "Permission denied") + return real_open(file, *args, **kwargs) + + with mock.patch("builtins.open", fake_open): + code, payload = self.run_cli("--session", SESSION, "--transcript", str(path)) + self.assertEqual(code, 2) + self.assert_error(payload, "unreadable") + + # format support ----------------------------------------------------- + def test_unsupported_version_names_the_version(self) -> None: + for bad in ("1.0.12", "9.9.9"): + with self.subTest(version=bad): + self.write(self._versioned(bad)) + code, payload = self.run_cli("--session", SESSION) + self.assertEqual(code, 2) + self.assertIn(bad, self.assert_error(payload, "unsupported-format")) + + def _versioned(self, version: str) -> Transcript: + t = Transcript(version=version) + t.user("hello") + return t + + def test_missing_version_is_unsupported(self) -> None: + t = Transcript() + record = t.user("hello") + del record["version"] + t.lines = [json.dumps(record)] + self.write(t) + code, payload = self.run_cli("--session", SESSION) + self.assertEqual(code, 2) + self.assert_error(payload, "unsupported-format") + + def test_malformed_core_records_are_unsupported(self) -> None: + cases = { + "no-message": lambda r: r.pop("message"), + "bad-content": lambda r: r["message"].update(content=42), + "bad-uuid": lambda r: r.update(uuid=7), + "bad-block": lambda r: r["message"].update(content=[7]), + } + for name, mutate in cases.items(): + with self.subTest(name): + t = Transcript() + record = t.user("hello") + mutate(record) + t.lines = [json.dumps(record)] + self.write(t) + code, payload = self.run_cli("--session", SESSION) + self.assertEqual(code, 2) + self.assert_error(payload, "unsupported-format") + + def test_unknown_noncore_types_are_skipped_and_counted(self) -> None: + t = self.basic() + t.add({"type": "hologram", "sessionId": SESSION, "version": VERSION}) + t.add({"type": "teleport", "sessionId": SESSION}) + t.add({"type": "file-history-snapshot", "messageId": "m", "snapshot": {}}) + self.write(t) + payload = self.read_ok() + self.assertEqual(payload["coverage"]["unsupportedSkipped"], 2) + self.assertEqual(payload["coverage"]["recordsTotal"], 5) + warning = next(w for w in payload["warnings"] if w["code"] == "unsupported-records-skipped") + self.assertIn("hologram", warning["message"]) + + def test_noncore_record_with_foreign_session_is_mismatch(self) -> None: + t = self.basic() + t.add({"type": "summary", "sessionId": "other-session", "summary": "x"}) + self.write(t) + code, payload = self.run_cli("--session", SESSION) + self.assertEqual(code, 2) + self.assert_error(payload, "session-mismatch") + + def test_empty_transcript_is_never_empty_success(self) -> None: + for body in ("", "\n"): + with self.subTest(body=body): + (self.project / f"{SESSION}.jsonl").write_text(body, encoding="utf-8") + code, payload = self.run_cli("--session", SESSION) + self.assertEqual(code, 2) + self.assert_error(payload, "unsupported-format") + + def test_only_noncore_records_is_not_empty_success(self) -> None: + t = Transcript() + t.add({"type": "hologram", "sessionId": SESSION}) + self.write(t) + code, payload = self.run_cli("--session", SESSION) + self.assertEqual(code, 2) + self.assert_error(payload, "unsupported-format") + + # budget ------------------------------------------------------------- + def big_transcript(self, results: int, size: int) -> Transcript: + t = Transcript() + for index in range(results): + t.user(f"prompt {index}") + t.assistant(f"answer {index}") + t.tool_use(f"toolu_{index}", "Bash", {"command": f"cmd {index}"}) + head, tail = f"HEAD{index}-", f"-TAIL{index}" + t.tool_result(f"toolu_{index}", head + ("x" * size) + tail) + return t + + def test_large_transcript_over_4_mib_streams_and_respects_budget(self) -> None: + t = self.big_transcript(results=24, size=200_000) + path = self.write(t) + self.assertGreater(path.stat().st_size, 4 * 1024 * 1024) + payload = self.read_ok("--max-bytes", "40000") + cov = payload["coverage"] + self.assertEqual(cov["recordsTotal"], 24 * 4) + self.assertLess(cov["recordsIncluded"], cov["recordsTotal"]) + self.assertGreater(cov["bytesOmitted"], 4 * 1024 * 1024 - 100_000) + prompts = [r["text"] for r in payload["records"] if r["origin"] == "person"] + answers = [r["text"] for r in payload["records"] if r["role"] == "assistant" and not r["tool"]] + self.assertEqual(prompts, [f"prompt {i}" for i in range(24)]) + self.assertEqual(answers, [f"answer {i}" for i in range(24)]) + results = [r for r in payload["records"] if r["role"] == "tool_result"] + self.assertTrue(results) + self.assertTrue(all(r["truncated"] for r in results)) + # Oldest tool results go first; the retained ones are the newest. + kept = [int(r["toolUseId"].split("_")[1]) for r in results] + self.assertEqual(kept, sorted(kept)) + self.assertEqual(kept[-1], 23) + self.assertLess(kept[0], 23) if len(kept) > 1 else None + self.assertLessEqual(len(json.dumps(payload).encode("utf-8")), 40000 + 4000) + + def test_excerpt_keeps_head_and_tail_before_dropping(self) -> None: + t = self.big_transcript(results=3, size=5000) + self.write(t) + full = self.read_ok() + self.assertTrue(all(not r["truncated"] for r in full["records"] if r["role"] == "tool_result")) + payload = self.read_ok("--max-bytes", "9000") + self.assertEqual(payload["coverage"]["recordsIncluded"], payload["coverage"]["recordsTotal"]) + results = [r for r in payload["records"] if r["role"] == "tool_result"] + self.assertEqual(len(results), 3) + for index, record in enumerate(results): + self.assertTrue(record["truncated"]) + self.assertTrue(record["text"].startswith(f"HEAD{index}-")) + self.assertTrue(record["text"].endswith(f"-TAIL{index}")) + self.assertGreater(payload["coverage"]["bytesOmitted"], 10000) + + def test_protected_text_is_never_deleted_and_conflict_is_reported(self) -> None: + t = Transcript() + long_prompt = "P" * 6000 + t.user(long_prompt) + t.assistant("A" * 6000) + t.tool_use("toolu_1", "Bash", {"command": "ls"}) + t.tool_result("toolu_1", "R" * 3000) + self.write(t) + payload = self.read_ok("--max-bytes", "1000") + texts = [r["text"] for r in payload["records"]] + self.assertIn(long_prompt, texts) + self.assertIn("A" * 6000, texts) + self.assertFalse([r for r in payload["records"] if r["role"] == "tool_result"]) + self.assertTrue(any(w["code"] == "budget-conflict" for w in payload["warnings"])) + self.assertEqual(payload["coverage"]["recordsTotal"], 4) + + def test_invalid_max_bytes(self) -> None: + self.write(self.basic()) + code, payload = self.run_cli("--session", SESSION, "--max-bytes", "0") + self.assertEqual(code, 2) + self.assert_error(payload, "unreadable") + + # stable refs ---------------------------------------------------------- + def test_refs_are_stable_after_append(self) -> None: + t = self.basic() + path = self.write(t) + first = self.read_ok() + t.user("a later prompt") + t.assistant("a later answer") + path.write_text(t.text(), encoding="utf-8") + second = self.read_ok() + before = [(r["ref"], r["text"], r["parentRef"]) for r in first["records"]] + after = [(r["ref"], r["text"], r["parentRef"]) for r in second["records"]] + self.assertEqual(after[: len(before)], before) + self.assertEqual(second["coverage"]["recordsTotal"], 7) + self.assertNotEqual(first["fingerprint"], second["fingerprint"]) + + def refs_of(self, payload: dict) -> dict[str, dict]: + return {r["text"]: r for r in payload["records"]} + + def test_refs_are_full_32_hex_uuid_derived(self) -> None: + self.write(self.basic()) + records = self.read_ok()["records"] + for record in records: + self.assertRegex(record["ref"], r"^[0-9a-f]{32}$") + self.assertEqual(records[0]["ref"], make_uuid(1).replace("-", "")) + self.assertEqual(records[1]["parentRef"], records[0]["ref"]) + + def test_ref_is_stable_when_an_appended_uuid_collides_with_its_prefix(self) -> None: + t = Transcript() + t.user("single prompt") + path = self.write(t) + before = self.read_ok()["records"] + t.assistant("next answer") # shares the 4f1c2a9e prefix + path.write_text(t.text(), encoding="utf-8") + after = self.read_ok()["records"] + self.assertEqual(after[0], before[0]) + self.assertEqual(after[1]["parentRef"], before[0]["ref"]) + self.assertEqual(len({r["ref"] for r in after}), 2) + + def test_refs_do_not_depend_on_earlier_or_tied_timestamps(self) -> None: + t = Transcript() + t.user("first prompt") + path = self.write(t) + before = self.read_ok()["records"][0] + t.assistant("answer with earlier timestamp") + t.assistant("answer with tied timestamp") + records = [json.loads(line) for line in t.lines] + records[1]["timestamp"] = "2000-01-01T00:00:00Z" + records[2]["timestamp"] = records[0]["timestamp"] + path.write_text("\n".join(json.dumps(r) for r in records) + "\n", encoding="utf-8") + after = self.refs_of(self.read_ok()) + self.assertEqual(after["first prompt"]["ref"], before["ref"]) + self.assertEqual(after["answer with earlier timestamp"]["parentRef"], before["ref"]) + self.assertEqual( + after["answer with tied timestamp"]["parentRef"], + after["answer with earlier timestamp"]["ref"], + ) + + def test_refs_are_stable_when_a_partial_tail_completes(self) -> None: + t = Transcript() + t.user("kept prompt") + t.assistant("tail answer") + complete = t.text() + first_line, tail_line = complete.rstrip("\n").split("\n") + path = self.write(t) + path.write_text(first_line + "\n" + tail_line[:20], encoding="utf-8") + partial = self.read_ok() + self.assertTrue(partial["coverage"]["pendingTail"]) + path.write_text(complete, encoding="utf-8") + done = self.read_ok() + self.assertEqual(done["records"][0], partial["records"][0]) + self.assertEqual(done["records"][1]["parentRef"], partial["records"][0]["ref"]) + + def test_refs_are_stable_when_subagent_records_grow_or_appear(self) -> None: + t = Transcript() + t.user("delegate") + t.tool_use("toolu_task", "Task", {"prompt": "look"}) + t.tool_result("toolu_task", "found it", toolUseResult={"agentId": "a1b2c3"}) + self.write(t) + before = self.read_ok()["records"] + + def sub_line(uid: str, parent, text: str, stamp: str) -> str: + return json.dumps({ + "type": "assistant", "uuid": uid, "parentUuid": parent, "sessionId": SESSION, + "version": VERSION, "timestamp": stamp, "isSidechain": True, "agentId": "a1b2c3", + "message": {"role": "assistant", "content": [{"type": "text", "text": text}]}, + }) + + folder = self.project / SESSION / "subagents" + folder.mkdir(parents=True) + sub = folder / "agent-a1b2c3.jsonl" + one = make_uuid(900) + sub.write_text(sub_line(one, None, "sub one", "2000-01-01T00:00:00Z") + "\n", encoding="utf-8") + discovered = self.read_ok()["records"] + self.assertEqual([r for r in discovered if not r["agentId"]], before) + sub_one = next(r for r in discovered if r["agentId"]) + self.assertEqual(sub_one["ref"], one.replace("-", "")) + task_call = next(r for r in before if r["tool"] == "Task") + self.assertEqual(sub_one["parentRef"], task_call["ref"]) + sub.write_text( + sub.read_text(encoding="utf-8") + + sub_line(make_uuid(901), one, "sub two", "1999-01-01T00:00:00Z") + "\n", + encoding="utf-8", + ) + grown = self.refs_of(self.read_ok()) + self.assertEqual(grown["sub one"], sub_one) + self.assertEqual(grown["sub two"]["parentRef"], sub_one["ref"]) + + def test_multiple_blocks_of_one_uuid_have_deterministic_distinct_refs(self) -> None: + t = Transcript() + t.user("ask") + t.assistant([ + {"type": "text", "text": "block one"}, + {"type": "tool_use", "id": "toolu_x", "name": "Bash", "input": {"command": "ls"}}, + {"type": "text", "text": "block three"}, + ]) + path = self.write(t) + first = self.read_ok()["records"] + refs = [r["ref"] for r in first] + self.assertEqual(len(set(refs)), 4) + for ref in refs: + self.assertRegex(ref, r"^[0-9a-f]{32}$") + self.assertEqual(first[1]["ref"], make_uuid(2).replace("-", "")) + self.assertEqual(first[2]["parentRef"], first[1]["ref"]) + self.assertEqual(first[3]["parentRef"], first[1]["ref"]) + t.user("later") + path.write_text(t.text(), encoding="utf-8") + second = self.read_ok()["records"] + self.assertEqual(second[:4], first) + + # serialized budget ------------------------------------------------------ + def cli_stdout(self, *args: str) -> tuple[int, str]: + out = io.StringIO() + with contextlib.redirect_stdout(out): + code = cli.main(["bridge", "read", "--session", SESSION, *args]) + return code, out.getvalue() + + def test_stdout_bytes_stay_within_max_bytes(self) -> None: + self.write(self.big_transcript(results=20, size=5000)) + code, text = self.cli_stdout("--max-bytes", "40000") + self.assertEqual(code, 0) + payload = json.loads(text) + self.assertLessEqual(len(text.encode("utf-8")), 40000) + self.assertEqual(payload["warnings"], []) + prompts = [r["text"] for r in payload["records"] if r["origin"] == "person"] + self.assertEqual(prompts, [f"prompt {i}" for i in range(20)]) + + def test_over_budget_protected_text_still_warns_with_true_size(self) -> None: + t = Transcript() + t.user("P" * 6000) + t.assistant("A" * 6000) + self.write(t) + code, text = self.cli_stdout("--max-bytes", "1000") + payload = json.loads(text) + self.assertEqual([w["code"] for w in payload["warnings"]], ["budget-conflict"]) + self.assertGreater(len(text.encode("utf-8")), 12000) + + def test_truncated_tool_input_reports_omitted_bytes(self) -> None: + t = Transcript() + t.user("go") + t.tool_use("toolu_1", "Write", {"content": "x" * 10000}) + self.write(t) + payload = self.read_ok() + record = [r for r in payload["records"] if r["tool"] == "Write"][0] + self.assertTrue(record["truncated"]) + self.assertGreater(payload["coverage"]["bytesOmitted"], 7000) + + # fail closed ------------------------------------------------------------ + def test_non_string_core_text_is_unsupported_format(self) -> None: + for name, rtype in (("assistant", "assistant"), ("user", "user")): + with self.subTest(name): + t = Transcript() + if rtype == "assistant": + t.assistant([{"type": "text", "text": {"unexpected": "object"}}]) + else: + t.add(t._base("user", message={"role": "user", "content": [{"type": "text", "text": {"unexpected": "object"}}]})) + self.write(t) + code, payload = self.run_cli("--session", SESSION) + self.assertEqual(code, 2) + self.assert_error(payload, "unsupported-format") + + def test_unsupported_version_on_skipped_record_is_unsupported_format(self) -> None: + t = Transcript() + t.user("hello") + t.add({"type": "future-thing", "sessionId": SESSION, "version": "9.9.9"}) + self.write(t) + code, payload = self.run_cli("--session", SESSION) + self.assertEqual(code, 2) + self.assertIn("9.9.9", self.assert_error(payload, "unsupported-format")) + + def test_versionless_metadata_and_unknown_types_are_still_skipped(self) -> None: + t = Transcript() + t.user("hello") + t.add({"type": "summary", "summary": "x"}) + t.add({"type": "future-thing", "sessionId": SESSION}) + t.add({"type": "future-thing", "sessionId": SESSION, "version": VERSION}) + self.write(t) + payload = self.read_ok() + self.assertEqual(payload["coverage"]["unsupportedSkipped"], 2) + + # read-only ------------------------------------------------------------ + def snapshot(self) -> dict[str, tuple[int, int, bytes | None]]: + state = {} + for path in sorted(Path(self._tmp.name).rglob("*")): + stat = path.stat() + state[str(path)] = (stat.st_size, stat.st_mtime_ns, path.read_bytes() if path.is_file() else None) + return state + + def test_reader_leaves_the_fixture_directory_unchanged(self) -> None: + self.write(self.basic()) + folder = self.project / SESSION / "subagents" + folder.mkdir(parents=True) + sub = Transcript() + sub.user("sub", agentId="zz", isSidechain=True) + (folder / "agent-zz.jsonl").write_text(sub.text(), encoding="utf-8") + before = self.snapshot() + self.read_ok() + self.run_cli("--session", "no-such-session") + self.assertEqual(self.snapshot(), before) + + def test_reader_opens_only_for_reading(self) -> None: + self.write(self.basic()) + modes = [] + real_open = open + + def spy(file, mode="r", *args, **kwargs): + modes.append(mode) + return real_open(file, mode, *args, **kwargs) + + with mock.patch("builtins.open", spy): + self.read_ok() + self.assertTrue(modes) + self.assertTrue(all(set(m) <= {"r", "b"} for m in modes), modes) + + def test_cli_requires_session(self) -> None: + with contextlib.redirect_stderr(io.StringIO()), self.assertRaises(SystemExit) as raised: + cli.main(["bridge", "read"]) + self.assertEqual(raised.exception.code, 2) + + +if __name__ == "__main__": + unittest.main()