diff --git a/README.md b/README.md index c870bcc..f97875c 100644 --- a/README.md +++ b/README.md @@ -42,7 +42,8 @@ Run `substrate-claude-code configure` to print the required wiring without writi If the variables are absent or the API is temporarily unreachable, capture remains enabled and new events are placed in the local spool beneath `~/.substrate/claude_code_memory/`. Set `SUBSTRATE_STATE_HOME` to relocate the state root, which is particularly useful for isolated -tests. +tests. Sidechain capture is enabled by default. Set `SUBSTRATE_CAPTURE_SIDECHAINS=0` +to use the emergency exclusion kill-switch. ## MCP tools @@ -62,8 +63,8 @@ The repository also adds `/substrate-status` and `/substrate-recall` slash comma Claude Code invokes four command hooks from `hooks/hooks.json`: -- **Stop** captures only user and assistant messages not previously checkpointed and emits a - `turn` event. +- **Stop** captures normalized user, assistant, tool-call, tool-result, and system blocks not + previously checkpointed and emits a `turn` event. - **PreCompact** captures the same incremental transcript window as `pre_compress` before Claude Code compacts its context. - **SessionEnd** emits a content-free `session_end` event containing only the normalized message @@ -76,29 +77,24 @@ under the plugin state directory. Capture events are durably spooled before netw ## Privacy boundary -The transcript reader keeps only top-level `user` and `assistant` records. It excludes sidechain -records produced by subagents and ignores system, attachment, mode, permission-mode, and -last-prompt records. +The transcript reader captures top-level and sidechain `user`, `assistant`, and `system` records. +Sidechain records carry record/block coordinates and session ancestry. Tool calls and results are +separate text-only messages paired by `tool_call_id`. A paired result receives its tool name; an +orphaned or ambiguous result receives a reason code and no source identity. -For captured user and assistant messages, the plugin may send: - -- visible text from string content and `text`, `input_text`, or `output_text` blocks; -- short markers such as `[tool_use: Read]` and `[tool_result]`; -- role, message index, bounded message identifier, session boundary, and capture metadata. - -It does not intentionally send: - -- raw tool inputs or raw tool-result payloads; -- hidden reasoning, token usage, billing fields, or arbitrary provider metadata; -- attachment bodies, binary media, data URLs, or sidechain transcript text; -- `SUBSTRATE_API_KEY` or other recognized credential-shaped values. +For every captured block, full credential detection runs before the 65,536-byte UTF-8 ceiling. If +a recognized credential occurs anywhere in a block, the whole block becomes content-free. This +prevents a secret from being cut at a former truncation boundary. Binary and media bodies, hidden +reasoning, token usage, billing fields, and arbitrary provider metadata are not captured. The shared capture core redacts recognized secrets before persistence and transfer. Redaction is defense in depth, not proof that arbitrary sensitive prose is absent. Visible prompts and assistant output can themselves contain confidential material, so configure only a trusted Substrate server and review its access and retention policy. Failed deliveries remain in a -bounded owner-private local spool until delivered, evicted by the bound, quarantined, or removed -by the operator. +bounded owner-private local spool. The spool reserves capacity for +boundary events and refuses newest events under pressure instead of evicting older evidence. +`substrate-claude-code status` exposes persistent `evicted`, `quarantined`, `dropped`, and +`duplicates` counters without exposing content. ## Fail-open behavior diff --git a/src/claude_code_memory/hook.py b/src/claude_code_memory/hook.py index 081328e..fe88258 100644 --- a/src/claude_code_memory/hook.py +++ b/src/claude_code_memory/hook.py @@ -47,7 +47,12 @@ def _capture_transcript( ) -> None: session_id = str(data.get("session_id") or "")[:512] transcript_path = data.get("transcript_path") - messages = read_messages(transcript_path) if isinstance(transcript_path, str) else [] + settings = config.resolved(PROVIDER_ID) + messages = ( + read_messages(transcript_path, include_sidechains=bool(settings["capture_sidechains"])) + if isinstance(transcript_path, str) + else [] + ) client, spool, deliverer, builder = runtime_factory() del client, spool checkpoint = _checkpoint(session_id) diff --git a/src/claude_code_memory/transcript.py b/src/claude_code_memory/transcript.py index ed46a0c..591dd37 100644 --- a/src/claude_code_memory/transcript.py +++ b/src/claude_code_memory/transcript.py @@ -2,65 +2,418 @@ from __future__ import annotations +import hashlib import json +from collections.abc import Iterator from pathlib import Path from typing import Any from substrate_capture import normalize_message +from substrate_capture.redaction import ( + configured_secret_values, + iter_redacted_text_chunks, + redact_text, +) MAX_MESSAGES = 2000 -MAX_MESSAGE_CHARS = 65_536 +MAX_BLOCK_BYTES = 65_536 +# Kept as a compatibility alias for callers which imported the old character bound. +MAX_MESSAGE_CHARS = MAX_BLOCK_BYTES _MAX_LINE_BYTES = 1024 * 1024 +_SCAN_CHARS = 16 * 1024 _TEXT_BLOCK_TYPES = frozenset({"text", "input_text", "output_text"}) +_BINARY_KEYS = frozenset( + { + "attachment", + "attachments", + "base64", + "binary", + "blob", + "bytes", + "file_content", + "image", + "images", + } +) +SOURCE_PROTOCOL_TRANSCRIPT = "claude_code_transcript" +SOURCE_PROTOCOL_TOOL = "claude_code_tool" +# Reserved for the approved Phase 5 edit path. No current parser emits it. +SOURCE_PROTOCOL_USER_EDIT = "user_edit" -def _bounded_text(value: str) -> str: - if len(value) <= MAX_MESSAGE_CHARS: +def _visible_json(value: Any, *, depth: int = 0) -> Any: + """Return JSON-safe text source material without binary/media values.""" + if depth > 12: + return "[NESTED_CONTENT_OMITTED]" + if value is None or isinstance(value, (bool, int, float, str)): return value - return value[:MAX_MESSAGE_CHARS] + "\n[CONTENT_TRUNCATED]" + if isinstance(value, list): + return [_visible_json(item, depth=depth + 1) for item in value] + if isinstance(value, dict): + return { + str(key): ( + "[BINARY_CONTENT_OMITTED]" + if str(key).casefold() in _BINARY_KEYS + else _visible_json(item, depth=depth + 1) + ) + for key, item in value.items() + } + return "[NON_JSON_CONTENT_OMITTED]" -def _tool_marker(block: dict[str, Any]) -> str: - block_type = str(block.get("type") or "").lower() +def _canonical_text(value: Any) -> str: + if isinstance(value, str): + return value + return json.dumps( + _visible_json(value), ensure_ascii=False, separators=(",", ":"), sort_keys=True + ) + + +def _tool_result_text(value: Any) -> str: + """Extract textual tool output while omitting media blocks.""" + if isinstance(value, str): + return value + if isinstance(value, list): + parts: list[str] = [] + for item in value: + if isinstance(item, str): + parts.append(item) + elif isinstance(item, dict): + block_type = str(item.get("type") or "").casefold() + if block_type in _TEXT_BLOCK_TYPES and isinstance(item.get("text"), str): + parts.append(item["text"]) + return "".join(parts) + if isinstance(value, dict): + block_type = str(value.get("type") or "").casefold() + if block_type in _TEXT_BLOCK_TYPES and isinstance(value.get("text"), str): + return value["text"] + return _canonical_text(value) + + +def _utf8_prefix(value: str, maximum: int) -> bytes: + """Return the longest valid UTF-8 prefix within ``maximum`` bytes.""" + if not value or maximum <= 0: + return b"" + low, high = 0, min(len(value), maximum) + while low < high: + middle = (low + high + 1) // 2 + if len(value[:middle].encode("utf-8")) <= maximum: + low = middle + else: + high = middle - 1 + return value[:low].encode("utf-8") + + +def _safe_block(value: str, secrets: tuple[str, ...]) -> dict[str, Any]: + """Scan the complete block, then redact-or-bound it without releasing a prefix.""" + raw_digest = hashlib.sha256() + safe_digest = hashlib.sha256() + original_bytes = 0 + safe_bytes = 0 + retained = bytearray() + retention_open = True + + def chunks() -> Iterator[str]: + nonlocal original_bytes + for offset in range(0, len(value), _SCAN_CHARS): + piece = value[offset : offset + _SCAN_CHARS] + encoded = piece.encode("utf-8") + raw_digest.update(encoded) + original_bytes += len(encoded) + yield piece + + for piece in iter_redacted_text_chunks(chunks(), secrets): + encoded = piece.encode("utf-8") + safe_digest.update(encoded) + safe_bytes += len(encoded) + if retention_open and len(retained) < MAX_BLOCK_BYTES: + selected = _utf8_prefix(piece, MAX_BLOCK_BYTES - len(retained)) + retained.extend(selected) + if len(selected) < len(encoded): + retention_open = False + + credential_detected = ( + original_bytes != safe_bytes or raw_digest.digest() != safe_digest.digest() + ) + if credential_detected: + content = "" + retained_bytes = 0 + codes = ["credential_detected"] + truncated = False + else: + content = retained.decode("utf-8", errors="strict") + retained_bytes = len(retained) + codes = [] + truncated = original_bytes > retained_bytes + return { + "content": content, + "original_bytes": original_bytes, + "retained_bytes": retained_bytes, + "redaction_codes": codes, + "truncated": truncated, + "content_digest": hashlib.sha256(content.encode("utf-8")).hexdigest(), + } + + +def _redacted_bound(value: Any, secrets: tuple[str, ...], maximum: int = 512) -> str: + """Redact a complete scalar before applying its wire-size bound.""" + return redact_text(str(value or ""), secrets)[:maximum] + + +def _coordinates( + record: dict[str, Any], + record_index: int, + block_index: int, + secrets: tuple[str, ...], +) -> dict[str, Any]: + return { + "transcript_coordinate": {"record": record_index, "block": block_index}, + "session_ancestry": { + "session_id": _redacted_bound(record.get("sessionId"), secrets), + "message_id": _redacted_bound(record.get("uuid"), secrets), + "parent_message_id": _redacted_bound(record.get("parentUuid"), secrets), + "agent_id": _redacted_bound(record.get("agentId"), secrets), + "is_sidechain": record.get("isSidechain") is True, + }, + } + + +def _envelope( + *, + role: str, + content: str, + record: dict[str, Any], + record_index: int, + block_index: int, + secrets: tuple[str, ...], + tool_call_id: str | None = None, + tool_name: str | None = None, + source_protocol: str = SOURCE_PROTOCOL_TRANSCRIPT, + source_identity: str | None = None, + attribution_reason_code: str | None = None, + quarantine: bool = False, +) -> dict[str, Any]: + safe = _safe_block(content, secrets) + if quarantine and "credential_detected" not in safe["redaction_codes"]: + safe.update( + content="", + retained_bytes=0, + redaction_codes=["credential_detected"], + truncated=False, + content_digest=hashlib.sha256(b"").hexdigest(), + ) + if quarantine: + tool_call_id = None + tool_name = None + source_identity = None + message: dict[str, Any] = { + "role": role, + **safe, + "tool_call_id": tool_call_id, + "tool_name": tool_name, + "source_protocol": source_protocol, + "source_identity": source_identity, + "observed_at": record.get("timestamp"), + "platform_message_id": record.get("uuid"), + **_coordinates(record, record_index, block_index, secrets), + } + if attribution_reason_code: + message["attribution_reason_code"] = attribution_reason_code + return message + + +def _block_content(block: Any) -> str | None: + """Return the exact visible content stream represented by one transcript block.""" + if isinstance(block, str): + return block + if not isinstance(block, dict): + return None + block_type = str(block.get("type") or "").casefold() + if block_type in _TEXT_BLOCK_TYPES and isinstance(block.get("text"), str): + return block["text"] if block_type == "tool_use": - name = block.get("name") - label = str(name)[:128] if isinstance(name, str) and name else "tool" - return f"[tool_use: {label}]" + return _canonical_text(block.get("input")) if block_type == "tool_result": - return "[tool_result: error]" if block.get("is_error") is True else "[tool_result]" - return "" + return _tool_result_text(block.get("content")) + return None + +def _credential_detected(value: str, secrets: tuple[str, ...]) -> bool: + return "credential_detected" in _safe_block(value, secrets)["redaction_codes"] -def _flatten_content(content: Any) -> str: + +def _parse_record( + record: dict[str, Any], record_index: int, secrets: tuple[str, ...] +) -> list[dict[str, Any]]: + message = record.get("message") + if not isinstance(message, dict): + return [] + default_role = str(message.get("role") or record.get("type") or "").casefold() + content = message.get("content") if isinstance(content, str): - return _bounded_text(content) + return [ + _envelope( + role=default_role, + content=content, + record=record, + record_index=record_index, + block_index=0, + secrets=secrets, + ) + ] if not isinstance(content, list): - return "" + return [] + + # Scan the complete visible content stream before structural boundaries are + # restored. This catches configured credentials split across top-level or + # nested text blocks. A joint detection conservatively quarantines every + # block in the record so no receiver can reconstruct either fragment. + block_contents = [_block_content(block) for block in content] + joint_content = "".join(piece for piece in block_contents if piece is not None) + joint_quarantine = _credential_detected(joint_content, secrets) - parts: list[str] = [] - size = 0 - for block in content: - text = "" + parsed: list[dict[str, Any]] = [] + for block_index, (block, block_content) in enumerate(zip(content, block_contents, strict=True)): if isinstance(block, str): - text = block - elif isinstance(block, dict): - block_type = str(block.get("type") or "").lower() - if block_type in _TEXT_BLOCK_TYPES and isinstance(block.get("text"), str): - text = block["text"] - else: - text = _tool_marker(block) - if not text: + parsed.append( + _envelope( + role=default_role, + content=block, + record=record, + record_index=record_index, + block_index=block_index, + secrets=secrets, + quarantine=joint_quarantine, + ) + ) + continue + if not isinstance(block, dict): continue - remaining = MAX_MESSAGE_CHARS - size - if remaining <= 0: - break - selected = text[:remaining] - parts.append(selected) - size += len(selected) - flattened = "\n".join(parts) - if size >= MAX_MESSAGE_CHARS: - flattened += "\n[CONTENT_TRUNCATED]" - return flattened + block_type = str(block.get("type") or "").casefold() + if block_type in _TEXT_BLOCK_TYPES and isinstance(block.get("text"), str): + parsed.append( + _envelope( + role=default_role, + content=block["text"], + record=record, + record_index=record_index, + block_index=block_index, + secrets=secrets, + quarantine=joint_quarantine, + ) + ) + elif block_type == "tool_use": + raw_call_id = str(block.get("id") or "") + raw_name = str(block.get("name") or "") + raw_input = block_content or "" + block_quarantine = joint_quarantine or _credential_detected( + raw_call_id + raw_name + raw_input, secrets + ) + call_id = raw_call_id[:512] or None + name = raw_name[:512] or None + parsed.append( + _envelope( + role="tool_call", + content=raw_input, + record=record, + record_index=record_index, + block_index=block_index, + secrets=secrets, + tool_call_id=call_id, + tool_name=name, + source_protocol=SOURCE_PROTOCOL_TOOL, + source_identity=name, + attribution_reason_code=(None if call_id else "missing_tool_call_id"), + quarantine=block_quarantine, + ) + ) + elif block_type == "tool_result": + raw_result_id = str(block.get("tool_use_id") or "") + raw_result = block_content or "" + block_quarantine = joint_quarantine or _credential_detected( + raw_result_id + raw_result, secrets + ) + result_id = raw_result_id[:512] or None + parsed.append( + _envelope( + role="tool_result", + content=raw_result, + record=record, + record_index=record_index, + block_index=block_index, + secrets=secrets, + tool_call_id=result_id, + source_protocol=SOURCE_PROTOCOL_TOOL, + attribution_reason_code="pairing_pending", + quarantine=block_quarantine, + ) + ) + return parsed + + +def _lineage_key(message: dict[str, Any]) -> tuple[str, str, bool]: + ancestry = message.get("session_ancestry") + if not isinstance(ancestry, dict): + return ("", "", False) + return ( + str(ancestry.get("session_id") or ""), + str(ancestry.get("agent_id") or ""), + ancestry.get("is_sidechain") is True, + ) + + +def _clear_tool_result_identity(message: dict[str, Any], reason: str) -> None: + # The deployed schema-v2 server treats tool_call_id as an identity fallback. + # Unverified results therefore retain no field from which identity can be minted. + message["tool_call_id"] = None + message["tool_name"] = None + message["source_identity"] = None + message["attribution_reason_code"] = reason + + +def _pair_tool_results(messages: list[dict[str, Any]]) -> None: + calls: dict[tuple[tuple[str, str, bool], str], list[dict[str, Any]]] = {} + global_calls: dict[str, list[dict[str, Any]]] = {} + results: dict[tuple[tuple[str, str, bool], str], list[dict[str, Any]]] = {} + for message in messages: + call_id = message.get("tool_call_id") + if not isinstance(call_id, str) or not call_id: + continue + key = (_lineage_key(message), call_id) + if message.get("role") == "tool_call": + calls.setdefault(key, []).append(message) + global_calls.setdefault(call_id, []).append(message) + elif message.get("role") == "tool_result": + results.setdefault(key, []).append(message) + + for message in messages: + if message.get("role") != "tool_result": + continue + call_id = message.get("tool_call_id") + if not isinstance(call_id, str) or not call_id: + _clear_tool_result_identity(message, "missing_tool_call_id") + continue + key = (_lineage_key(message), call_id) + same_results = results.get(key, []) + same_calls = calls.get(key, []) + if len(same_results) != 1: + reason = "duplicate_tool_result" + elif not same_calls and global_calls.get(call_id): + reason = "cross_stream_tool_result" + elif not same_calls: + reason = "orphaned_tool_result" + elif len(same_calls) != 1: + reason = "ambiguous_tool_call_id" + else: + name = same_calls[0].get("tool_name") + if not isinstance(name, str) or not name: + reason = "missing_tool_name" + else: + message["tool_name"] = name + message["source_identity"] = name + message.pop("attribution_reason_code", None) + continue + _clear_tool_result_identity(message, reason) def _discard_line_remainder(stream: Any, chunk: bytes) -> None: @@ -68,49 +421,60 @@ def _discard_line_remainder(stream: Any, chunk: bytes) -> None: chunk = stream.readline(_MAX_LINE_BYTES + 1) -def read_messages(path: str | Path) -> list[dict[str, Any]]: - """Return normalized user and assistant messages from a transcript. +def read_messages(path: str | Path, *, include_sidechains: bool = True) -> list[dict[str, Any]]: + """Return bounded, redacted transcript blocks in transcript order. - Invalid records, sidechains, oversized lines, truncated final JSON, and all - filesystem errors are ignored. This function never raises. + Invalid records, oversized lines, truncated final JSON, and filesystem errors + are ignored. Sidechains are captured by default and can be disabled with the + explicit configuration kill-switch. This function never raises. """ - messages: list[dict[str, Any]] = [] + parsed: list[dict[str, Any]] = [] try: + secrets = configured_secret_values() with Path(path).open("rb") as stream: - while len(messages) < MAX_MESSAGES: + record_index = 0 + while len(parsed) < MAX_MESSAGES: raw = stream.readline(_MAX_LINE_BYTES + 1) if not raw: break if len(raw) > _MAX_LINE_BYTES: _discard_line_remainder(stream, raw) + record_index += 1 continue try: record = json.loads(raw.decode("utf-8", errors="strict")) except (UnicodeDecodeError, json.JSONDecodeError): + record_index += 1 continue if not isinstance(record, dict): + record_index += 1 continue - if record.get("type") not in {"user", "assistant"}: + if record.get("type") not in {"user", "assistant", "system"}: + record_index += 1 continue - if record.get("isSidechain") is True: + if record.get("isSidechain") is True and not include_sidechains: + record_index += 1 continue - message = record.get("message") - if not isinstance(message, dict): - continue - role = str(message.get("role") or record.get("type") or "").lower() - normalized = normalize_message( - { - "role": role, - "content": _flatten_content(message.get("content")), - "platform_message_id": record.get("uuid"), - }, - index=len(messages), - ) - if normalized is not None: - messages.append(normalized) + for message in _parse_record(record, record_index, secrets): + if len(parsed) >= MAX_MESSAGES: + break + parsed.append(message) + record_index += 1 + _pair_tool_results(parsed) + normalized: list[dict[str, Any]] = [] + for index, message in enumerate(parsed): + item = normalize_message(message, index=index, secrets=secrets) + if item is not None: + normalized.append(item) + return normalized except Exception: # noqa: BLE001 - transcript capture must always fail open return [] - return messages -__all__ = ["MAX_MESSAGES", "MAX_MESSAGE_CHARS", "read_messages"] +__all__ = [ + "MAX_BLOCK_BYTES", + "MAX_MESSAGES", + "MAX_MESSAGE_CHARS", + "SOURCE_PROTOCOL_USER_EDIT", + "read_messages", +] diff --git a/src/substrate_capture/_vendor.json b/src/substrate_capture/_vendor.json index 7bb60f2..10e0f44 100644 --- a/src/substrate_capture/_vendor.json +++ b/src/substrate_capture/_vendor.json @@ -1,16 +1,16 @@ -{ - "files": { - "__init__.py": "ee06117bc4b7a94c1f2e9c5c8b9a25e5e65a7c23bf7db26b00ba9990dccb3088", - "checkpoint.py": "e7e687a119a756c5e5992c1dec6e34c792839bc07d9332563003241a2ab7d030", - "client.py": "9e4df698fb3919b502766d835c9ee6e3c9995381c75772a6452942bc0f54cd4a", - "config.py": "5aeffa59a6856ead25ad5ccb21b19de126a9c52189adbc968c2fdd54c04d4727", - "delivery.py": "743c1da8fd4c658b9272907aabd07b4ac7c69a61073733e15674dc2341f98a51", - "events.py": "bd4ed86b8ec3f202af03ae77d3af99a7094739f9f64be4efb8ee1528b75050ab", - "mcp.py": "50b2f046792ed7da20644c167a1bd38da464d2025ac1a0949599f2e5ac90097e", - "py.typed": "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855", - "redaction.py": "e9bec198aa7ad018da359d2e9aa6df1dab717881bb41b1001348911b23e6439b", - "spool.py": "b3c31c6f124d0f57d23c11e9a0b76626921ef0d48846cfbdec59ac117ae402db", - "tools.py": "1e4887131c95d10e99550de4a3d4f38953661adbceb1fbec26bd5986552f5902" - }, - "vendor_version": "1.0.0" -} +{ + "files": { + "__init__.py": "ee06117bc4b7a94c1f2e9c5c8b9a25e5e65a7c23bf7db26b00ba9990dccb3088", + "checkpoint.py": "e7e687a119a756c5e5992c1dec6e34c792839bc07d9332563003241a2ab7d030", + "client.py": "9e4df698fb3919b502766d835c9ee6e3c9995381c75772a6452942bc0f54cd4a", + "config.py": "ea34befc39ee48b11c011cf7890a53ba0eaadd57a5e4c395068489319d638d3e", + "delivery.py": "6c48d989d47fedbda01fdc966915bd39ba47c8f1fefb3815b28a74b8e01dd740", + "events.py": "fb1bb57e52a804f9a17a48b8657737d6c761c1db9c839abd7a2eb76e328bd6ff", + "mcp.py": "50b2f046792ed7da20644c167a1bd38da464d2025ac1a0949599f2e5ac90097e", + "py.typed": "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855", + "redaction.py": "e9bec198aa7ad018da359d2e9aa6df1dab717881bb41b1001348911b23e6439b", + "spool.py": "8ab56569e61da99f7fd8e1f539f4ecf3cdbf68cce5f50406212f5da4929aa977", + "tools.py": "1e4887131c95d10e99550de4a3d4f38953661adbceb1fbec26bd5986552f5902" + }, + "vendor_version": "1.0.0" +} diff --git a/src/substrate_capture/config.py b/src/substrate_capture/config.py index edc15f6..5d75634 100644 --- a/src/substrate_capture/config.py +++ b/src/substrate_capture/config.py @@ -30,7 +30,14 @@ "token", } ) -_PERSISTED_KEYS = ("api_url", "timeout", "max_capture_bytes", "recall_limit", "enabled") +_PERSISTED_KEYS = ( + "api_url", + "timeout", + "max_capture_bytes", + "recall_limit", + "enabled", + "capture_sidechains", +) def api_url() -> str: @@ -102,6 +109,9 @@ def resolved(provider_id: str) -> dict[str, Any]: "timeout": _positive_float(stored.get("timeout"), 10.0), "recall_limit": _bounded_int(stored.get("recall_limit"), 5, 1, 25), "enabled": stored.get("enabled") is not False, + "capture_sidechains": _environment_bool( + "SUBSTRATE_CAPTURE_SIDECHAINS", stored.get("capture_sidechains") is not False + ), } @@ -115,3 +125,10 @@ def _bounded_int(value: Any, default: int, minimum: int, maximum: int) -> int: if isinstance(value, int) and not isinstance(value, bool) and minimum <= value <= maximum: return value return default + + +def _environment_bool(name: str, default: bool) -> bool: + value = os.environ.get(name) + if value is None: + return default + return value.strip().casefold() not in {"0", "false", "no", "off"} diff --git a/src/substrate_capture/delivery.py b/src/substrate_capture/delivery.py index f986de9..1300de2 100644 --- a/src/substrate_capture/delivery.py +++ b/src/substrate_capture/delivery.py @@ -133,4 +133,5 @@ def status(self) -> dict[str, Any]: "configured": self.client is not None, "last_category": self.last_category, **self.counters, + **self.spool.statistics(), } diff --git a/src/substrate_capture/events.py b/src/substrate_capture/events.py index c830c31..9379b9f 100644 --- a/src/substrate_capture/events.py +++ b/src/substrate_capture/events.py @@ -15,7 +15,7 @@ from collections.abc import Iterable, Iterator, Sequence from typing import Any, Protocol, runtime_checkable -from .redaction import iter_redacted_text_chunks, redact +from .redaction import iter_redacted_text_chunks, redact, redact_text SCHEMA_VERSION = 2 RETENTION_DAYS = 90 @@ -27,6 +27,17 @@ "tool_call_id", "tool_calls", "tool_name", + "source_protocol", + "source_identity", + "observed_at", + "original_bytes", + "retained_bytes", + "redaction_codes", + "truncated", + "content_digest", + "attribution_reason_code", + "transcript_coordinate", + "session_ancestry", "timestamp", "platform_message_id", "name", @@ -117,13 +128,13 @@ def normalize_message( index: int, secrets: Sequence[str] = (), ) -> dict[str, Any] | None: - """Return a redacted, inference-safe message or ``None`` for system data.""" + """Return a redacted message containing only the approved envelope fields.""" role = str(message.get("role") or "").strip().lower() - if role not in {"user", "assistant", "tool"}: + if role not in {"user", "assistant", "tool", "tool_call", "tool_result", "system"}: return None selected: dict[str, Any] = {"role": role, "message_index": int(index)} for field in _MESSAGE_FIELDS[1:]: - if field not in message or message[field] is None: + if field not in message: continue selected[field] = ( _visible_content(message[field]) @@ -187,9 +198,7 @@ def _fragment_message(message: dict[str, Any], maximum: int) -> Iterator[dict[st encoded = canonical_bytes(message).decode("utf-8") encoded_digest = hashlib.sha256(encoded.encode("utf-8")).hexdigest() count = sum(1 for _ in _utf8_boundaries(encoded, max(1024, maximum // 4))) - for index, (start, end) in enumerate( - _utf8_boundaries(encoded, max(1024, maximum // 4)) - ): + for index, (start, end) in enumerate(_utf8_boundaries(encoded, max(1024, maximum // 4))): yield { "role": message["role"], "message_index": message["message_index"], @@ -334,12 +343,15 @@ def __init__( secrets: Sequence[str] = (), max_capture_bytes: int = MAX_CAPTURE_BYTES, ) -> None: - provider = str(provider_id)[:512] + self.secrets = tuple(secrets) + provider = redact_text(str(provider_id), self.secrets)[:512] if not provider: raise ValueError("provider_id must be a non-empty string") self.provider_id = provider self.scope = { - str(key): str(value)[:512] + redact_text(str(key), self.secrets)[:512]: redact_text( + str(value), self.secrets + )[:512] for key, value in scope.items() if value is not None and str(value) } @@ -358,7 +370,6 @@ def __init__( self.scope["subject_id"] = hashlib.sha256( f"{platform}\0{user_id}".encode() ).hexdigest()[:24] - self.secrets = tuple(secrets) self.max_capture_bytes = max(16 * 1024, min(max_capture_bytes, MAX_CAPTURE_BYTES)) def message_events( @@ -595,16 +606,19 @@ def _event( deterministic: bool, event_id: str | None = None, ) -> dict[str, Any]: - safe_session = str(session_id)[:512] - safe_payload = {"session_id": safe_session, **payload} + safe_session = redact_text(str(session_id), self.secrets)[:512] + safe_kind = redact_text(str(kind), self.secrets)[:512] + safe_origin = redact_text(str(capture_origin), self.secrets)[:64] + safe_batch = redact_text(str(batch_id), self.secrets)[:128] + safe_payload = redact({"session_id": safe_session, **payload}, self.secrets) identity = { "provider_id": self.provider_id, - "kind": kind, + "kind": safe_kind, "scope": {**self.scope, "session_id": safe_session}, "session_lineage": {"session_id": safe_session}, "capture_boundary": boundary, - "capture_origin": capture_origin, - "batch_id": batch_id, + "capture_origin": safe_origin, + "batch_id": safe_batch, "payload": safe_payload, } resolved_id = event_id @@ -618,19 +632,19 @@ def _event( "schema_version": SCHEMA_VERSION, "event_id": resolved_id, "provider_id": self.provider_id, - "kind": kind, - "capture_kind": kind, + "kind": safe_kind, + "capture_kind": safe_kind, "capture_boundary": boundary, "session_lineage": {"session_id": safe_session}, "scope": {**self.scope, "session_id": safe_session}, - "capture_origin": str(capture_origin)[:64], + "capture_origin": safe_origin, "created_at": 0 if deterministic else time.time(), "retention_days": RETENTION_DAYS, "content_sha256": content_digest(safe_payload), "payload": safe_payload, } - if batch_id: - event["batch_id"] = str(batch_id)[:128] + if safe_batch: + event["batch_id"] = safe_batch messages = safe_payload.get("messages") if isinstance(messages, list): event["message_hashes"] = [content_digest(message) for message in messages] diff --git a/src/substrate_capture/spool.py b/src/substrate_capture/spool.py index ba51b77..242e638 100644 --- a/src/substrate_capture/spool.py +++ b/src/substrate_capture/spool.py @@ -1,4 +1,4 @@ -"""Bounded, private, durable JSON spool for nonblocking Hermes delivery.""" +"""Bounded, private, durable JSON spool for nonblocking delivery.""" from __future__ import annotations @@ -8,8 +8,38 @@ import threading import time import uuid +from contextlib import contextmanager from pathlib import Path -from typing import Any +from typing import Any, Iterator + +from .events import MAX_CAPTURE_BYTES + +try: + import fcntl +except ImportError: # pragma: no cover - exercised on Windows + fcntl = None # type: ignore[assignment] + import msvcrt +else: # pragma: no cover - keeps the Windows-only name explicit + msvcrt = None # type: ignore[assignment] + + +def _lock_file(descriptor: int) -> None: + if fcntl is not None: + fcntl.flock(descriptor, fcntl.LOCK_EX) + return + if os.fstat(descriptor).st_size == 0: + os.write(descriptor, b"\0") + os.fsync(descriptor) + os.lseek(descriptor, 0, os.SEEK_SET) + msvcrt.locking(descriptor, msvcrt.LK_LOCK, 1) + + +def _unlock_file(descriptor: int) -> None: + if fcntl is not None: + fcntl.flock(descriptor, fcntl.LOCK_UN) + return + os.lseek(descriptor, 0, os.SEEK_SET) + msvcrt.locking(descriptor, msvcrt.LK_UNLCK, 1) def _chmod_private(path: Path, mode: int) -> None: @@ -54,7 +84,9 @@ def secure_atomic_json_write(target: Path, value: Any) -> None: """Write JSON with exclusive temp creation, fsync, replace, and directory fsync.""" root = _safe_root(target.parent) target = _safe_child(root, target) - payload = (json.dumps(value, ensure_ascii=False, indent=2, sort_keys=True) + "\n").encode("utf-8") + payload = (json.dumps(value, ensure_ascii=False, indent=2, sort_keys=True) + "\n").encode( + "utf-8" + ) temporary = root / f".{target.name}.{os.getpid()}.{uuid.uuid4().hex}.tmp" flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL if hasattr(os, "O_NOFOLLOW"): @@ -79,47 +111,76 @@ def secure_atomic_json_write(target: Path, value: Any) -> None: class DurableSpool: - def __init__(self, root: Path, *, max_items: int = 1000, max_bytes: int = 10 * 1024 * 1024) -> None: + """A bounded FIFO which preserves old evidence and reserves boundary capacity.""" + + _COUNTER_KEYS = ("evicted", "quarantined", "dropped", "duplicates") + _BOUNDARY_KINDS = frozenset({"pre_compress", "session_boundary", "session_end"}) + + def __init__( + self, root: Path, *, max_items: int = 1000, max_bytes: int = 10 * 1024 * 1024 + ) -> None: self.root = _safe_root(root) self.max_items = max(1, max_items) - self.max_bytes = max(1024, max_bytes) + if max_bytes < MAX_CAPTURE_BYTES: + raise ValueError("max_bytes must reserve one maximum capture event") + self.max_bytes = max_bytes self._lock = threading.Lock() + self._lock_path = self.root / ".spool.lock" self._sequence = 0 self._claimed: set[Path] = set() - self.evicted_count = 0 - self.quarantined_count = 0 self._quarantine = _safe_root(self.root / "corrupt") + self._stats_path = self.root / ".spool-stats" + self._counters = {key: 0 for key in self._COUNTER_KEYS} + with self._transaction(refresh_counters=True): + pass def append(self, event: dict[str, Any]) -> Path: - payload = json.dumps(event, ensure_ascii=False, separators=(",", ":"), sort_keys=True).encode("utf-8") - if len(payload) > self.max_bytes: - raise ValueError("event exceeds spool limit") - with self._lock: + try: + payload = json.dumps( + event, ensure_ascii=False, separators=(",", ":"), sort_keys=True + ).encode("utf-8") + except (TypeError, ValueError): + with self._transaction(refresh_counters=True): + self._increment_locked("dropped") + raise ValueError("event is not JSON serializable") from None + with self._transaction(refresh_counters=True): + if len(payload) > self.max_bytes: + self._increment_locked("dropped") + raise ValueError("event exceeds spool limit") + duplicate = self._duplicate_locked(event) + if duplicate is not None: + self._increment_locked("duplicates") + return duplicate + files = self._files_locked() + total = self._total_bytes_locked(files) + boundary = self._is_boundary(event) + item_limit, byte_limit = self._admission_limits(boundary) + if len(files) + 1 > item_limit or total + len(payload) > byte_limit: + # Refuse the newest event. Never punch an unobservable hole in + # already-spooled evidence, and never evict a boundary record. + self._increment_locked("dropped") + raise ValueError("spool capacity unavailable") self._sequence += 1 target = self.root / ( - f"{time.time_ns():020d}-{os.getpid()}-{threading.get_ident()}-{self._sequence:08d}.json" + f"{time.time_ns():020d}-{os.getpid()}-{threading.get_ident()}-" + f"{self._sequence:08d}.json" ) self._write_payload_locked(target, payload) - self._trim_locked(protected=target) - files = self._files_locked() - total = sum(path.stat(follow_symlinks=False).st_size for path in files) - if len(files) > self.max_items or total > self.max_bytes: - try: - target.unlink() - _fsync_directory(self.root) - except FileNotFoundError: - pass - raise ValueError("spool capacity unavailable") return target + def statistics(self) -> dict[str, int]: + """Return persistent content-free loss and quarantine counters.""" + with self._transaction(refresh_counters=True): + return dict(self._counters) + def oldest(self) -> Path | None: - with self._lock: + with self._transaction(refresh_counters=True): files = [path for path in self._files_locked() if path not in self._claimed] return files[0] if files else None def claim_oldest(self) -> Path | None: - """Reserve the oldest event so capacity trimming cannot remove it in flight.""" - with self._lock: + """Reserve the oldest event while it is in flight.""" + with self._transaction(refresh_counters=True): files = [path for path in self._files_locked() if path not in self._claimed] if not files: return None @@ -128,7 +189,7 @@ def claim_oldest(self) -> Path | None: return path def release(self, path: Path) -> None: - with self._lock: + with self._transaction(refresh_counters=True): self._claimed.discard(_safe_child(self.root, path)) def load(self, path: Path) -> dict[str, Any]: @@ -147,7 +208,7 @@ def load(self, path: Path) -> dict[str, Any]: return value def remove(self, path: Path) -> None: - with self._lock: + with self._transaction(refresh_counters=True): safe = _safe_child(self.root, path) self._claimed.discard(safe) try: @@ -158,7 +219,7 @@ def remove(self, path: Path) -> None: def quarantine(self, path: Path) -> None: """Move corrupt data aside without parsing or exposing its contents.""" - with self._lock: + with self._transaction(refresh_counters=True): safe = _safe_child(self.root, path) self._claimed.discard(safe) if not safe.exists() or safe.is_symlink(): @@ -166,15 +227,102 @@ def quarantine(self, path: Path) -> None: destination = self._quarantine / f"{safe.stem}-{uuid.uuid4().hex}.bad" os.replace(safe, destination) _chmod_private(destination, 0o600) - self.quarantined_count += 1 + self._counters["quarantined"] += 1 self._trim_quarantine_locked() + self._persist_counters_locked() _fsync_directory(self.root) _fsync_directory(self._quarantine) def __len__(self) -> int: - with self._lock: + with self._transaction(refresh_counters=True): return len(self._files_locked()) + @contextmanager + def _transaction(self, *, refresh_counters: bool) -> Iterator[None]: + """Serialize root-scoped accounting across hook processes and instances.""" + with self._lock: + lock_path = _safe_child(self.root, self._lock_path) + flags = os.O_RDWR | os.O_CREAT + if hasattr(os, "O_NOFOLLOW"): + flags |= os.O_NOFOLLOW + descriptor = os.open(lock_path, flags, 0o600) + locked = False + try: + info = os.fstat(descriptor) + if not stat.S_ISREG(info.st_mode): + raise ValueError("spool lock is not a regular file") + _chmod_private(lock_path, 0o600) + _lock_file(descriptor) + locked = True + if refresh_counters: + self._counters = self._load_counters() + self._sync_counter_attributes() + yield + finally: + if locked: + _unlock_file(descriptor) + os.close(descriptor) + + def _admission_limits(self, boundary: bool) -> tuple[int, int]: + if boundary: + return self.max_items, self.max_bytes + # Ordinary writes may never consume the one slot and the full byte + # budget needed by a maximum legal capture event. + reserved_items = 1 + reserved_bytes = min(MAX_CAPTURE_BYTES, self.max_bytes) + return self.max_items - reserved_items, self.max_bytes - reserved_bytes + + @classmethod + def _is_boundary(cls, event: dict[str, Any]) -> bool: + kind = str(event.get("kind") or event.get("capture_kind") or "") + return kind in cls._BOUNDARY_KINDS + + def _duplicate_locked(self, event: dict[str, Any]) -> Path | None: + event_id = event.get("event_id") + if not isinstance(event_id, str) or not event_id: + return None + for path in self._files_locked(): + try: + descriptor = self._open_readonly(path) + with os.fdopen(descriptor, "rb") as stream: + raw = stream.read(self.max_bytes + 1) + value = json.loads(raw.decode("utf-8", errors="strict")) + except (OSError, ValueError, UnicodeDecodeError, json.JSONDecodeError): + continue + if isinstance(value, dict) and value.get("event_id") == event_id: + return path + return None + + def _load_counters(self) -> dict[str, int]: + counters = {key: 0 for key in self._COUNTER_KEYS} + try: + if self._stats_path.is_symlink() or self._stats_path.stat().st_size > 4096: + return counters + value = json.loads(self._stats_path.read_text(encoding="utf-8")) + except (FileNotFoundError, OSError, UnicodeDecodeError, json.JSONDecodeError): + return counters + if not isinstance(value, dict): + return counters + for key in self._COUNTER_KEYS: + count = value.get(key) + if isinstance(count, int) and not isinstance(count, bool) and count >= 0: + counters[key] = count + return counters + + def _increment_locked(self, key: str, amount: int = 1) -> None: + self._counters[key] += amount + self._persist_counters_locked() + + def _persist_counters_locked(self) -> None: + secure_atomic_json_write(self._stats_path, self._counters) + self._sync_counter_attributes() + + def _sync_counter_attributes(self) -> None: + self.evicted_count = self._counters["evicted"] + self.quarantined_count = self._counters["quarantined"] + self.dropped_count = self._counters["dropped"] + self.duplicate_count = self._counters["duplicates"] + def _write_payload_locked(self, target: Path, payload: bytes) -> None: target = _safe_child(self.root, target) temporary = self.root / f".{target.name}.{uuid.uuid4().hex}.tmp" @@ -216,33 +364,15 @@ def _files_locked(self) -> list[Path]: files.append(item) return sorted(files, key=lambda item: item.name) - def _trim_locked(self, *, protected: Path | None = None) -> None: - protected_paths = set(self._claimed) - if protected is not None: - protected_paths.add(protected) - files = self._files_locked() - sizes: dict[Path, int] = {} + @staticmethod + def _total_bytes_locked(files: list[Path]) -> int: + total = 0 for path in files: try: - sizes[path] = path.stat(follow_symlinks=False).st_size - except FileNotFoundError: - sizes[path] = 0 - total = sum(sizes.values()) - changed = False - while files and (len(files) > self.max_items or total > self.max_bytes): - victim = next((path for path in files if path not in protected_paths), None) - if victim is None: - break - files.remove(victim) - try: - victim.unlink() - total -= sizes[victim] - self.evicted_count += 1 - changed = True + total += path.stat(follow_symlinks=False).st_size except FileNotFoundError: continue - if changed: - _fsync_directory(self.root) + return total def _trim_quarantine_locked(self) -> None: files = sorted( @@ -260,10 +390,13 @@ def _trim_quarantine_locked(self) -> None: except FileNotFoundError: sizes[item] = 0 total = sum(sizes.values()) + evicted = 0 while files and (len(files) > self.max_items or total > self.max_bytes): victim = files.pop(0) try: victim.unlink() total -= sizes[victim] + evicted += 1 except FileNotFoundError: continue + self._counters["evicted"] += evicted diff --git a/tests/test_delivery.py b/tests/test_delivery.py index 407f1c0..c29fb97 100644 --- a/tests/test_delivery.py +++ b/tests/test_delivery.py @@ -58,9 +58,7 @@ def spool(tmp_path: Path) -> DurableSpool: ("memory_write", "/api/v1/hermes/memory-write-events"), ], ) -def test_kind_routes_to_the_documented_endpoint( - spool: DurableSpool, kind: str, path: str -) -> None: +def test_kind_routes_to_the_documented_endpoint(spool: DurableSpool, kind: str, path: str) -> None: client = FakeClient() assert Deliverer(spool, client).deliver(_event(kind)) == "" assert client.requests[0]["path"] == path @@ -154,6 +152,7 @@ def test_status_is_content_free(spool: DurableSpool) -> None: status = deliverer.status() assert status["pending"] == 1 assert status["configured"] is True + assert {"evicted", "quarantined", "dropped", "duplicates"} <= status.keys() serialized = repr(status) assert "payload" not in serialized assert "e1" not in serialized diff --git a/tests/test_events.py b/tests/test_events.py index 28e7947..7f26846 100644 --- a/tests/test_events.py +++ b/tests/test_events.py @@ -97,3 +97,16 @@ def test_normalize_message_keeps_only_known_fields() -> None: def test_content_digest_is_order_independent() -> None: assert content_digest({"a": 1, "b": 2}) == content_digest({"b": 2, "a": 1}) + + +def test_schema_v2_and_envelope_limit_remain_unchanged() -> None: + assert SCHEMA_VERSION == 2 + assert MAX_CAPTURE_BYTES == 262_144 + + +def test_normalizer_accepts_the_approved_five_roles() -> None: + roles = ["user", "assistant", "tool_call", "tool_result", "system"] + assert [ + normalize_message({"role": role, "content": "x"}, index=index)["role"] + for index, role in enumerate(roles) + ] == roles diff --git a/tests/test_hooks.py b/tests/test_hooks.py index 6d21526..73893ee 100644 --- a/tests/test_hooks.py +++ b/tests/test_hooks.py @@ -16,7 +16,7 @@ from claude_code_memory import PROVIDER_ID # noqa: E402 from claude_code_memory import hook # noqa: E402 -from substrate_capture import CaptureEventBuilder, Deliverer, DurableSpool # noqa: E402 +from substrate_capture import CaptureEventBuilder, Deliverer, DurableSpool, config # noqa: E402 class FakeClient: @@ -201,3 +201,26 @@ def test_malformed_hook_input_fails_open_without_stdout() -> None: stdout = io.StringIO() assert hook.run("stop", stdin=io.StringIO("{"), stdout=stdout, stderr=io.StringIO()) == 0 assert stdout.getvalue() == "" + + +def test_hook_honors_sidechain_capture_kill_switch( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + state = tmp_path / "state" + monkeypatch.setenv("SUBSTRATE_STATE_HOME", str(state)) + config.save_config(PROVIDER_ID, {"capture_sidechains": False}) + transcript = tmp_path / "sidechain.jsonl" + transcript.write_text( + json.dumps( + { + "type": "assistant", + "isSidechain": True, + "message": {"role": "assistant", "content": "excluded by kill switch"}, + } + ) + + "\n", + encoding="utf-8", + ) + harness = RuntimeHarness(tmp_path / "runtime") + assert _invoke("stop", transcript, harness)[0] == 0 + assert harness.client.requests == [] diff --git a/tests/test_review185_regressions.py b/tests/test_review185_regressions.py new file mode 100644 index 0000000..2a46986 --- /dev/null +++ b/tests/test_review185_regressions.py @@ -0,0 +1,229 @@ +from __future__ import annotations + +import hashlib +import json +import multiprocessing +import os +import sys +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).parents[1] / "src")) + +from claude_code_memory.transcript import MAX_BLOCK_BYTES, read_messages +from substrate_capture.events import CaptureEventBuilder, MAX_CAPTURE_BYTES, canonical_bytes +from substrate_capture.spool import DurableSpool + +SYNTHETIC_CREDENTIAL = "super-secret-value-0123456789" + + +def write_record(path: Path, content, *, record_type="user", **record_fields) -> Path: + record = { + "type": record_type, + **record_fields, + "message": {"role": "user" if record_type == "user" else "assistant", "content": content}, + } + path.write_text(json.dumps(record, ensure_ascii=False) + "\n", encoding="utf-8") + return path + + +def test_old_boundary_full_scan_passes(tmp_path, monkeypatch): + monkeypatch.setenv("ANTHROPIC_API_KEY", SYNTHETIC_CREDENTIAL) + text = "x" * (MAX_BLOCK_BYTES - 5) + SYNTHETIC_CREDENTIAL + "tail" + msg = read_messages(write_record(tmp_path / "t.jsonl", text))[0] + assert msg["content"] == "" + assert msg["retained_bytes"] == 0 + assert msg["content_digest"] == hashlib.sha256(b"").hexdigest() + assert SYNTHETIC_CREDENTIAL not in json.dumps(msg) + + +def test_secret_split_across_top_level_tool_result_blocks_must_not_survive(tmp_path, monkeypatch): + monkeypatch.setenv("ANTHROPIC_API_KEY", SYNTHETIC_CREDENTIAL) + left, right = SYNTHETIC_CREDENTIAL[:13], SYNTHETIC_CREDENTIAL[13:] + blocks = [ + {"type": "tool_result", "tool_use_id": "orphan-a", "content": left}, + {"type": "tool_result", "tool_use_id": "orphan-b", "content": right}, + ] + messages = read_messages(write_record(tmp_path / "split.jsonl", blocks)) + # Security invariant: no raw portion of a configured secret is retained across capture blocks. + assert all(m["content"] == "" for m in messages) + + +def test_secret_split_across_inner_tool_result_text_blocks_must_not_survive(tmp_path, monkeypatch): + monkeypatch.setenv("ANTHROPIC_API_KEY", SYNTHETIC_CREDENTIAL) + left, right = SYNTHETIC_CREDENTIAL[:13], SYNTHETIC_CREDENTIAL[13:] + blocks = [{ + "type": "tool_result", "tool_use_id": "orphan", + "content": [{"type": "text", "text": left}, {"type": "text", "text": right}], + }] + message = read_messages(write_record(tmp_path / "inner-split.jsonl", blocks))[0] + assert message["content"] == "" + + +def test_tool_use_input_full_quarantine_passes(tmp_path, monkeypatch): + monkeypatch.setenv("ANTHROPIC_API_KEY", SYNTHETIC_CREDENTIAL) + block = [{"type": "tool_use", "id": "call", "name": "Fetch", "input": {"x": "safe", "arg": SYNTHETIC_CREDENTIAL, "tail": "safe"}}] + msg = read_messages(write_record(tmp_path / "input.jsonl", block, record_type="assistant"))[0] + assert msg["content"] == "" + assert msg["retained_bytes"] == 0 + assert msg["content_digest"] == hashlib.sha256(b"").hexdigest() + assert SYNTHETIC_CREDENTIAL not in json.dumps(msg) + assert "safe" not in msg["content"] + + +def test_multibyte_adjacent_secret_passes(tmp_path, monkeypatch): + monkeypatch.setenv("ANTHROPIC_API_KEY", SYNTHETIC_CREDENTIAL) + text = "é🙂" * 40000 + SYNTHETIC_CREDENTIAL + "界" + msg = read_messages(write_record(tmp_path / "utf8.jsonl", text))[0] + assert msg["content"] == "" + assert msg["original_bytes"] == len(text.encode()) + assert msg["content_digest"] == hashlib.sha256(b"").hexdigest() + + +def test_orphan_secret_full_quarantine_passes(tmp_path, monkeypatch): + monkeypatch.setenv("ANTHROPIC_API_KEY", SYNTHETIC_CREDENTIAL) + block = [{"type": "tool_result", "tool_use_id": "no-call", "content": "head " + SYNTHETIC_CREDENTIAL + " tail"}] + msg = read_messages(write_record(tmp_path / "orphan-secret.jsonl", block))[0] + assert msg["content"] == "" + assert msg["retained_bytes"] == 0 + assert msg["redaction_codes"] == ["credential_detected"] + assert msg["content_digest"] == hashlib.sha256(b"").hexdigest() + assert SYNTHETIC_CREDENTIAL not in json.dumps(msg) + + +def test_orphan_must_not_retain_server_attributable_tool_call_id(tmp_path): + block = [{"type": "tool_result", "tool_use_id": "forged-id", "content": "safe"}] + msg = read_messages(write_record(tmp_path / "orphan.jsonl", block))[0] + assert msg["tool_name"] is None and msg["source_identity"] is None + # The deployed server's v2 compatibility path falls back to tool_call_id as identity. + assert msg["tool_call_id"] is None + assert msg["attribution_reason_code"] == "orphaned_tool_result" + + +def test_duplicate_results_do_not_both_pair_to_one_call(tmp_path): + records = [ + {"type": "assistant", "message": {"role": "assistant", "content": [{"type": "tool_use", "id": "one", "name": "Read", "input": {}}]}}, + {"type": "user", "message": {"role": "user", "content": [ + {"type": "tool_result", "tool_use_id": "one", "content": "real"}, + {"type": "tool_result", "tool_use_id": "one", "content": "forged duplicate"}, + ]}}, + ] + path = tmp_path / "dup-results.jsonl" + path.write_text("".join(json.dumps(r) + "\n" for r in records)) + results = [m for m in read_messages(path) if m["role"] == "tool_result"] + assert all(m["tool_name"] is None and m["source_identity"] is None and m["tool_call_id"] is None for m in results) + assert {m["attribution_reason_code"] for m in results} == {"duplicate_tool_result"} + + +def test_cross_sidechain_result_does_not_pair_with_root_call(tmp_path): + records = [ + {"type": "assistant", "sessionId": "same-session", "agentId": "root", "isSidechain": False, + "message": {"role": "assistant", "content": [{"type": "tool_use", "id": "shared", "name": "TrustedTool", "input": {}}]}}, + {"type": "user", "sessionId": "same-session", "agentId": "side-A", "isSidechain": True, + "message": {"role": "user", "content": [{"type": "tool_result", "tool_use_id": "shared", "content": "forged cross-lineage result"}]}}, + ] + path = tmp_path / "cross-lineage.jsonl" + path.write_text("".join(json.dumps(r) + "\n" for r in records)) + result = [m for m in read_messages(path) if m["role"] == "tool_result"][0] + assert result["tool_name"] is None and result["source_identity"] is None and result["tool_call_id"] is None + assert result["attribution_reason_code"] == "cross_stream_tool_result" + + +def test_sidechain_kill_switch_reader_passes(tmp_path): + path = write_record(tmp_path / "side.jsonl", "side", sessionId="s", agentId="a", isSidechain=True) + assert len(read_messages(path)) == 1 + assert read_messages(path, include_sidechains=False) == [] + + +def test_boundary_reservation_must_cover_max_sized_precompress(tmp_path): + spool = DurableSpool(tmp_path / "spool", max_items=3, max_bytes=300_000) + # The old quarter-reserve admitted this item and then rejected the legal boundary event. + with pytest.raises(ValueError, match="capacity"): + spool.append({"event_id": "ordinary", "kind": "turn", "content": "x" * 220_000}) + boundary = {"event_id": "boundary", "kind": "pre_compress", "content": "y" * 80_000} + assert len(json.dumps(boundary, separators=(",", ":"), sort_keys=True).encode()) < MAX_CAPTURE_BYTES + spool.append(boundary) + + +def test_loss_counters_survive_normal_restart_passes(tmp_path): + root = tmp_path / "spool" + spool = DurableSpool(root, max_items=2) + spool.append({"event_id": "a", "kind": "turn"}) + with pytest.raises(ValueError): + spool.append({"event_id": "b", "kind": "turn"}) + before = spool.statistics() + after = DurableSpool(root, max_items=2).statistics() + assert after == before and after["dropped"] == 1 + + +def test_post_redaction_capture_cap_passes(): + builder = CaptureEventBuilder({"platform": "cli"}, provider_id="claude_code_memory", secrets=(SYNTHETIC_CREDENTIAL,)) + events = builder.message_events("turn", "s", [{"role": "user", "content": "x" * 300_000 + SYNTHETIC_CREDENTIAL}]) + assert events + assert all(len(canonical_bytes(event)) <= MAX_CAPTURE_BYTES for event in events) + assert SYNTHETIC_CREDENTIAL not in json.dumps(events) + + +def test_long_configured_secret_in_new_tool_metadata_is_not_sliced_before_scan(tmp_path, monkeypatch): + long_secret = "Q" * 600 + monkeypatch.setenv("ANTHROPIC_API_KEY", long_secret) + block = [{"type": "tool_use", "id": long_secret, "name": long_secret, "input": {}}] + msg = read_messages(write_record(tmp_path / "metadata.jsonl", block, record_type="assistant"))[0] + serialized = json.dumps(msg) + assert "Q" * 32 not in serialized + assert msg["content"] == "" + + +def test_builder_does_not_emit_unscanned_session_secret(): + builder = CaptureEventBuilder({"platform": "cli"}, provider_id="claude_code_memory", secrets=(SYNTHETIC_CREDENTIAL,)) + event = builder.payload_event("session_end", SYNTHETIC_CREDENTIAL, {"summary": {"message_count": 0}}) + assert SYNTHETIC_CREDENTIAL not in json.dumps(event) + + +def _process_append(root: str, start, outcomes, index: int) -> None: + spool = DurableSpool(Path(root), max_items=3) + start.wait() + try: + spool.append({"event_id": f"ordinary-{index}", "kind": "turn"}) + except ValueError: + outcomes.put("refused") + else: + outcomes.put("accepted") + + +@pytest.mark.skipif(os.name != "posix", reason="cross-process flock is POSIX-specific") +def test_two_hook_processes_cannot_race_away_boundary_reservation(tmp_path): + root = tmp_path / "racy-spool" + DurableSpool(root, max_items=3).append({"event_id": "ordinary-0", "kind": "turn"}) + context = multiprocessing.get_context("fork") + start = context.Event() + outcomes = context.Queue() + processes = [ + context.Process(target=_process_append, args=(str(root), start, outcomes, index)) + for index in (1, 2) + ] + for process in processes: + process.start() + start.set() + for process in processes: + process.join(timeout=10) + assert process.exitcode == 0 + assert sorted(outcomes.get(timeout=2) for _ in processes) == ["accepted", "refused"] + spool = DurableSpool(root, max_items=3) + assert len(spool) == 2 + spool.append({"event_id": "boundary", "kind": "session_end"}) + assert len(spool) == 3 + + +def test_concurrent_instances_do_not_overwrite_persisted_loss_counters(tmp_path): + root = tmp_path / "stats-race" + first = DurableSpool(root, max_items=2) + second = DurableSpool(root, max_items=2) # stale zero-valued counter snapshot + first.append({"event_id": "one", "kind": "turn"}) + with pytest.raises(ValueError): + first.append({"event_id": "dropped", "kind": "turn"}) + # This instance writes its stale `dropped=0` snapshot and erases the recorded loss. + assert second.append({"event_id": "one", "kind": "turn"}) + stats = DurableSpool(root, max_items=2).statistics() + assert stats["dropped"] == 1 and stats["duplicates"] == 1 diff --git a/tests/test_state.py b/tests/test_state.py index f31eb21..2806ae2 100644 --- a/tests/test_state.py +++ b/tests/test_state.py @@ -78,11 +78,35 @@ def test_spool_quarantines_rather_than_losing_a_corrupt_file(tmp_path: Path) -> assert len(spool) == 0 -def test_spool_enforces_a_capacity_bound(tmp_path: Path) -> None: +def test_spool_refuses_newest_and_reserves_boundary_capacity(tmp_path: Path) -> None: spool = DurableSpool(tmp_path / "s", max_items=3) - for index in range(10): - spool.append({"event_id": f"e{index}", "kind": "turn"}) - assert len(spool) <= 3 + spool.append({"event_id": "ordinary-1", "kind": "turn"}) + spool.append({"event_id": "ordinary-2", "kind": "turn"}) + with pytest.raises(ValueError, match="capacity"): + spool.append({"event_id": "ordinary-refused", "kind": "turn"}) + boundary = spool.append({"event_id": "boundary", "kind": "session_end"}) + with pytest.raises(ValueError, match="capacity"): + spool.append({"event_id": "newest-refused", "kind": "turn"}) + assert spool.load(boundary)["event_id"] == "boundary" + assert {spool.load(path)["event_id"] for path in spool._files_locked()} == { # noqa: SLF001 + "ordinary-1", + "ordinary-2", + "boundary", + } + assert spool.statistics()["dropped"] == 2 + assert spool.statistics()["evicted"] == 0 + + +def test_spool_duplicate_and_quarantine_counters_persist(tmp_path: Path) -> None: + root = tmp_path / "s" + spool = DurableSpool(root) + original = spool.append({"event_id": "same", "kind": "turn"}) + assert spool.append({"event_id": "same", "kind": "turn"}) == original + original.write_text("{not json", encoding="utf-8") + spool.quarantine(original) + reloaded = DurableSpool(root) + assert reloaded.statistics()["duplicates"] == 1 + assert reloaded.statistics()["quarantined"] == 1 def test_atomic_write_leaves_no_temporary_file(tmp_path: Path) -> None: @@ -169,3 +193,14 @@ def test_state_home_rejects_a_traversing_provider_id( for bad in ("..", "a/b", "a\\b", ""): with pytest.raises(ValueError): config.state_home(bad) + + +def test_sidechain_capture_defaults_on_and_has_config_kill_switch( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setenv("SUBSTRATE_STATE_HOME", str(tmp_path)) + assert config.resolved("claude_code_memory")["capture_sidechains"] is True + config.save_config("claude_code_memory", {"capture_sidechains": False}) + assert config.resolved("claude_code_memory")["capture_sidechains"] is False + monkeypatch.setenv("SUBSTRATE_CAPTURE_SIDECHAINS", "1") + assert config.resolved("claude_code_memory")["capture_sidechains"] is True diff --git a/tests/test_transcript.py b/tests/test_transcript.py index e477ed4..a06790a 100644 --- a/tests/test_transcript.py +++ b/tests/test_transcript.py @@ -2,6 +2,7 @@ from __future__ import annotations +import hashlib import json import sys from pathlib import Path @@ -9,40 +10,216 @@ sys.path.insert(0, str(Path(__file__).parents[1] / "src")) from claude_code_memory.transcript import ( # noqa: E402 + MAX_BLOCK_BYTES, MAX_MESSAGES, - MAX_MESSAGE_CHARS, + SOURCE_PROTOCOL_USER_EDIT, read_messages, ) FIXTURE = Path(__file__).parent / "fixtures" / "transcript_sample.jsonl" -def test_sidechains_and_non_message_records_are_excluded() -> None: - messages = read_messages(FIXTURE) - assert [message["role"] for message in messages] == ["user", "assistant"] - serialized = json.dumps(messages) - assert "subagent noise" not in serialized - assert "not captured" not in serialized +def _write(path: Path, records: list[dict[str, object]]) -> Path: + path.write_text( + "".join(json.dumps(record, ensure_ascii=False) + "\n" for record in records), + encoding="utf-8", + ) + return path -def test_content_blocks_are_flattened_without_raw_tool_payloads() -> None: +def test_tool_blocks_are_separate_text_messages_and_pair_by_call_id() -> None: messages = read_messages(FIXTURE) - content = messages[1]["content"] - assert "I will inspect it." in content - assert "[tool_use: Read]" in content - assert "[tool_result]" in content - assert "The parser is bounded." in content - assert "must-not-leak" not in content - assert "raw tool payload" not in content + assert [message["role"] for message in messages] == [ + "system", + "user", + "assistant", + "tool_call", + "tool_result", + "assistant", + "assistant", + ] + call = messages[3] + result = messages[4] + assert call["tool_call_id"] == result["tool_call_id"] == "tool-1" + assert call["tool_name"] == result["tool_name"] == "Read" + assert result["source_identity"] == "Read" + assert result["content"] == "raw tool payload must not leak" + assert isinstance(result["content"], str) + assert "attribution_reason_code" not in result + + +def test_every_block_carries_the_approved_envelope() -> None: + required = { + "role", + "content", + "tool_call_id", + "tool_name", + "source_protocol", + "source_identity", + "observed_at", + "original_bytes", + "retained_bytes", + "redaction_codes", + "truncated", + "content_digest", + } + for message in read_messages(FIXTURE): + assert required <= message.keys() + assert ( + message["content_digest"] + == hashlib.sha256(message["content"].encode("utf-8")).hexdigest() + ) + assert SOURCE_PROTOCOL_USER_EDIT == "user_edit" + assert all( + message["source_protocol"] != SOURCE_PROTOCOL_USER_EDIT + for message in read_messages(FIXTURE) + ) + + +def test_secret_spanning_the_former_truncation_boundary_quarantines_whole_block( + tmp_path: Path, +) -> None: + secret = "sk-live-" + "a" * 200 + content = "x" * (MAX_BLOCK_BYTES - 12) + " api_key=" + secret + " safe suffix" + transcript = _write( + tmp_path / "boundary.jsonl", + [{"type": "user", "message": {"role": "user", "content": content}}], + ) + message = read_messages(transcript)[0] + assert message["content"] == "" + assert message["retained_bytes"] == 0 + assert message["original_bytes"] == len(content.encode("utf-8")) + assert message["redaction_codes"] == ["credential_detected"] + assert secret not in json.dumps(message) + assert secret[:16] not in json.dumps(message) + + +def test_unsafe_tool_input_is_quarantined_as_a_whole_block(tmp_path: Path) -> None: + transcript = _write( + tmp_path / "tool-secret.jsonl", + [ + { + "type": "assistant", + "message": { + "role": "assistant", + "content": [ + { + "type": "tool_use", + "id": "secret-call", + "name": "Fetch", + "input": {"api_key": "synthetic-credential-value"}, + } + ], + }, + } + ], + ) + call = read_messages(transcript)[0] + assert call["content"] == "" + assert call["redaction_codes"] == ["credential_detected"] + assert "synthetic-credential-value" not in json.dumps(call) + + +def test_utf8_byte_ceiling_is_applied_after_full_scan(tmp_path: Path) -> None: + content = "é" * (MAX_BLOCK_BYTES // 2 + 50) + transcript = _write( + tmp_path / "utf8.jsonl", + [{"type": "assistant", "message": {"role": "assistant", "content": content}}], + ) + message = read_messages(transcript)[0] + assert len(message["content"].encode("utf-8")) == MAX_BLOCK_BYTES + assert message["retained_bytes"] == MAX_BLOCK_BYTES + assert message["original_bytes"] > message["retained_bytes"] + assert message["truncated"] is True + message["content"].encode("utf-8", errors="strict") + + +def test_orphaned_tool_result_has_no_attributable_provenance(tmp_path: Path) -> None: + transcript = _write( + tmp_path / "orphan.jsonl", + [ + { + "type": "user", + "message": { + "role": "user", + "content": [{"type": "tool_result", "tool_use_id": "missing", "content": "ok"}], + }, + } + ], + ) + result = read_messages(transcript)[0] + assert result["role"] == "tool_result" + assert result["tool_name"] is None + assert result["source_identity"] is None + assert result["attribution_reason_code"] == "orphaned_tool_result" + + +def test_ambiguous_tool_result_has_no_attributable_provenance(tmp_path: Path) -> None: + transcript = _write( + tmp_path / "ambiguous.jsonl", + [ + { + "type": "assistant", + "message": { + "role": "assistant", + "content": [ + {"type": "tool_use", "id": "same", "name": "Read", "input": {}}, + {"type": "tool_use", "id": "same", "name": "Web", "input": {}}, + ], + }, + }, + { + "type": "user", + "message": { + "role": "user", + "content": [{"type": "tool_result", "tool_use_id": "same", "content": "ok"}], + }, + }, + ], + ) + result = read_messages(transcript)[-1] + assert result["tool_name"] is None + assert result["source_identity"] is None + assert result["attribution_reason_code"] == "ambiguous_tool_call_id" + + +def test_sidechains_are_bounded_by_default_with_collision_free_coordinates( + tmp_path: Path, +) -> None: + transcript = _write( + tmp_path / "sidechains.jsonl", + [ + { + "type": "assistant", + "uuid": "same-looking-record", + "sessionId": "root", + "agentId": "a", + "isSidechain": True, + "message": {"role": "assistant", "content": "one"}, + }, + { + "type": "assistant", + "uuid": "same-looking-record", + "sessionId": "root", + "agentId": "b", + "isSidechain": True, + "message": {"role": "assistant", "content": "two"}, + }, + ], + ) + messages = read_messages(transcript) + coordinates = [json.dumps(item["transcript_coordinate"], sort_keys=True) for item in messages] + assert len(messages) == 2 + assert len(set(coordinates)) == len(coordinates) + assert all(item["session_ancestry"]["is_sidechain"] is True for item in messages) + assert read_messages(transcript, include_sidechains=False) == [] def test_garbage_and_truncated_final_lines_are_tolerated(tmp_path: Path) -> None: transcript = tmp_path / "transcript.jsonl" transcript.write_text( "garbage\n" - + json.dumps( - {"type": "user", "message": {"role": "user", "content": "kept"}} - ) + + json.dumps({"type": "user", "message": {"role": "user", "content": "kept"}}) + "\n" + '{"type":"assistant","message":', encoding="utf-8", @@ -55,16 +232,12 @@ def test_unreadable_or_missing_file_never_raises(tmp_path: Path) -> None: assert read_messages(tmp_path) == [] -def test_message_count_and_content_are_bounded(tmp_path: Path) -> None: +def test_message_count_is_bounded(tmp_path: Path) -> None: transcript = tmp_path / "large.jsonl" with transcript.open("w", encoding="utf-8") as stream: for index in range(MAX_MESSAGES + 5): - content = "x" * (MAX_MESSAGE_CHARS + 50) if index == 0 else str(index) stream.write( - json.dumps({"type": "user", "message": {"role": "user", "content": content}}) + json.dumps({"type": "user", "message": {"role": "user", "content": str(index)}}) + "\n" ) - messages = read_messages(transcript) - assert len(messages) == MAX_MESSAGES - assert messages[0]["content"].endswith("[CONTENT_TRUNCATED]") - assert len(messages[0]["content"]) <= MAX_MESSAGE_CHARS + 32 + assert len(read_messages(transcript)) == MAX_MESSAGES