#!/usr/bin/env python3 """Cross-host, deterministic reasoning follow-through guard. The guard observes host-visible hook payloads plus explicit visible-summary fields from trusted host transcript paths. It never persists thinking blocks, raw prompts, transcripts, or tool output. """ from __future__ import annotations import hashlib import json import math import os import re import sys import tempfile import time import unicodedata from collections import deque from contextlib import contextmanager from datetime import datetime, timezone from pathlib import Path from typing import Any SCHEMA_VERSION = "3.0" PREVIOUS_SCHEMA_VERSION = "2.0" LEGACY_SCHEMA_VERSION = "1.0" LEDGER_SCHEMA_VERSION = "1.0" REPORT_RELATIVE_PATH = Path(".prd_plugin/local/reason_guard.json") LOCK_RELATIVE_PATH = Path(".prd_plugin/local/reason_guard.lock") HOST_REPORT_FILENAMES = { "claude": "claude_reasoning_guard.json", "codex": "codex_reasoning_guard.json", "opencode": "opencode_reasoning_guard.json", } VALID_MODES = {"on", "report", "off"} VALID_CATEGORY_MODES = VALID_MODES | {"inherit"} VALID_BACKFILL_MODES = {"live", "recent", "full"} DEFAULT_BACKFILL_MODE = "recent" DEFAULT_BACKFILL_LIMIT = 20 MAX_BACKFILL_LIMIT = 100_000 CATEGORIES = ( "evidence_follow_through", "causal_claims", "permanent_mutations", "completion_claims", ) OPEN_STATES = {"IDENTIFIED", "PLANNED", "EXECUTED", "EVIDENCED"} TEXT_FIELDS = ( "commentary", "reasoning_summary", "plan", "message", "last_assistant_message", ) CHECK_RE = re.compile( r"\b(compar(?:e|ing)|distinguish|benchmark|test|verify|measure|inspect|reproduce|check)\b", re.IGNORECASE, ) ACTIVE_CHECK_RE = re.compile( r"(?:^|[.;]\s*|(? str: return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") def _digest(value: Any) -> str: encoded = json.dumps(value, sort_keys=True, default=str, ensure_ascii=True).encode("utf-8") return hashlib.sha256(encoded).hexdigest() def _sanitize(text: Any) -> str: if not isinstance(text, str): text = json.dumps(text, sort_keys=True, default=str, ensure_ascii=True) text = SECRET_ASSIGNMENT_RE.sub(lambda m: f"{m.group(1)}=[REDACTED]", text) text = SECRET_FLAG_RE.sub(lambda m: f"{m.group(1)}[REDACTED]", text) text = re.sub(r"\s+", " ", text).strip() return text[:MAX_EXCERPT] def _mojibake_score(value: str) -> int: return sum(value.count(marker) for marker in MOJIBAKE_MARKERS) def _normalize_visible_unicode(value: str) -> str: """Conservatively repair recognizable UTF-8 mojibake before hashing.""" current = unicodedata.normalize("NFC", value) for _ in range(2): current_score = _mojibake_score(current) if current_score == 0: break best = current best_score = current_score for encoding in ("cp1252", "latin1"): try: candidate = current.encode(encoding).decode("utf-8") except (UnicodeEncodeError, UnicodeDecodeError): continue candidate = unicodedata.normalize("NFC", candidate) candidate_score = _mojibake_score(candidate) if candidate_score < best_score: best = candidate best_score = candidate_score if best == current: break current = best return current def _normalized_surface(value: Any, host: str) -> str: """Normalize only allowlisted surface markers for the matching host.""" if not isinstance(value, str): return "unknown" normalized = SURFACE_ALIASES.get(value.strip().lower(), "unknown") expected_host = str(host or "").strip().lower() if normalized == "unknown": return normalized return normalized if SURFACE_HOSTS.get(normalized) == expected_host else "unknown" def _declared_surface(host: str, payload: dict[str, Any]) -> str: """Prefer bounded hook declarations and host-owned environment markers.""" candidates = [ payload.get("surface"), payload.get("client_surface"), os.environ.get("PRD_HOOK_SURFACE"), ] normalized_host = str(host or "").strip().lower() if normalized_host == "codex": candidates.append(os.environ.get("CODEX_INTERNAL_ORIGINATOR_OVERRIDE")) elif normalized_host == "claude": candidates.append(os.environ.get("CLAUDE_CODE_ENTRYPOINT")) for candidate in candidates: surface = _normalized_surface(candidate, normalized_host) if surface != "unknown": return surface return "opencode" if normalized_host == "opencode" else "unknown" def _transcript_surface(transcript: Path, host: str) -> str: """Read bounded host-owned metadata only; never inspect message content.""" try: with transcript.open("rb") as handle: data = handle.read(MAX_SURFACE_METADATA_BYTES) except OSError: return "unknown" for raw_line in data.splitlines()[:MAX_SURFACE_METADATA_LINES]: try: envelope = json.loads(raw_line) except (UnicodeDecodeError, json.JSONDecodeError): continue if not isinstance(envelope, dict): continue if host == "codex" and envelope.get("type") == "session_meta": payload = envelope.get("payload") if isinstance(payload, dict): surface = _normalized_surface(payload.get("originator"), host) if surface != "unknown": return surface elif host == "claude": surface = _normalized_surface(envelope.get("entrypoint"), host) if surface != "unknown": return surface return "unknown" def _session_surface( host: str, payload: dict[str, Any], ingestion: dict[str, Any], previous: Any = "unknown", ) -> str: declared = _declared_surface(host, payload) if declared != "unknown": return declared ingested = _normalized_surface(ingestion.get("surface"), host) if ingested != "unknown": return ingested retained = _normalized_surface(previous, host) if retained != "unknown": return retained return "opencode" if str(host).lower() == "opencode" else "unknown" def _normalized_model(value: Any) -> str: if not isinstance(value, str): return "unknown" normalized = value.strip() if MODEL_ID_RE.fullmatch(normalized) is None: return "unknown" return normalized def _normalized_effort(value: Any) -> str: if not isinstance(value, str): return "unknown" normalized = value.strip().lower() return normalized if normalized in VALID_EFFORTS else "unknown" def _declared_model_effort(payload: dict[str, Any]) -> tuple[str, str]: model = "unknown" effort = "unknown" for candidate in ( payload.get("model"), payload.get("model_id"), os.environ.get("PRD_HOOK_MODEL"), ): model = _normalized_model(candidate) if model != "unknown": break for candidate in ( payload.get("effort"), payload.get("reasoning_effort"), os.environ.get("PRD_HOOK_EFFORT"), ): effort = _normalized_effort(candidate) if effort != "unknown": break return model, effort def _session_model_effort( payload: dict[str, Any], ingestion: dict[str, Any], previous_model: Any = "unknown", previous_effort: Any = "unknown", ) -> tuple[str, str]: model, effort = _declared_model_effort(payload) if model == "unknown": model = _normalized_model(ingestion.get("model")) if model == "unknown": model = _normalized_model(previous_model) if effort == "unknown": effort = _normalized_effort(ingestion.get("effort")) if effort == "unknown": effort = _normalized_effort(previous_effort) return model, effort def _transcript_model_effort( transcript: Path, host: str, *, start: int, end: int, ) -> tuple[str, str, bool]: """Read only a bounded newly covered tail and retain the latest metadata.""" if end <= start or start < 0: return "unknown", "unknown", end == start and start >= 0 read_start = max(start, end - MAX_SESSION_METADATA_BYTES) dropped_prefix = read_start > start try: with transcript.open("rb") as handle: handle.seek(read_start) data = handle.read(end - read_start) except OSError: return "unknown", "unknown", False lines = data.splitlines() if dropped_prefix and lines: lines = lines[1:] lines = lines[-MAX_SESSION_METADATA_LINES:] model = "unknown" effort = "unknown" for raw_line in lines: try: envelope = json.loads(raw_line) except (UnicodeDecodeError, json.JSONDecodeError): continue if not isinstance(envelope, dict): continue candidate_model: Any = None candidate_effort: Any = None if host == "codex": payload = envelope.get("payload") if not isinstance(payload, dict): continue if envelope.get("type") == "turn_context": candidate_model = payload.get("model") candidate_effort = payload.get("effort") elif ( envelope.get("type") == "event_msg" and payload.get("type") == "thread_settings_applied" ): settings = payload.get("thread_settings") if isinstance(settings, dict): candidate_model = settings.get("model") candidate_effort = settings.get("reasoning_effort") elif host == "claude" and envelope.get("type") == "assistant": message = envelope.get("message") if not isinstance(message, dict) or message.get("role") != "assistant": continue candidate_model = message.get("model") candidate_effort = envelope.get("effort") normalized_model = _normalized_model(candidate_model) normalized_effort = _normalized_effort(candidate_effort) if normalized_model != "unknown": model = normalized_model if normalized_effort != "unknown": effort = normalized_effort return model, effort, True def _bounded_json(value: Any, depth: int = 0) -> Any: """Return bounded JSON-safe data or reject it without executing anything.""" if depth > MAX_EXPECTED_DEPTH: raise ValueError("structured value is too deeply nested") if value is None or isinstance(value, bool): return value if isinstance(value, (int, float)) and not isinstance(value, bool): if isinstance(value, float) and not math.isfinite(value): raise ValueError("structured numbers must be finite") return value if isinstance(value, str): if len(value) > MAX_EXCERPT: raise ValueError("structured string exceeds the field bound") return _sanitize(value) if isinstance(value, list): if len(value) > MAX_EXPECTED_COLLECTION: raise ValueError("structured list exceeds the collection bound") return [_bounded_json(item, depth + 1) for item in value] if isinstance(value, dict): if len(value) > MAX_EXPECTED_COLLECTION: raise ValueError("structured object exceeds the collection bound") result = {} for key, item in value.items(): if ( not isinstance(key, str) or not key or len(key) > 64 or SENSITIVE_KEY_RE.search(key) ): raise ValueError("structured object contains an unsafe key") result[key] = _bounded_json(item, depth + 1) return result raise ValueError("structured value is not JSON-safe") def _directive_candidate( payload: dict[str, Any], ) -> tuple[bool, Any, str]: direct = payload.get("reasoning_evidence_check") if direct is not None: if isinstance(direct, dict): try: encoded_direct = json.dumps( direct, sort_keys=True, ensure_ascii=True, separators=(",", ":"), ).encode("utf-8") except (TypeError, ValueError): return True, None, "reasoning_evidence_check must be JSON-safe" if len(encoded_direct) > MAX_DIRECTIVE_BYTES: return True, None, "reasoning_evidence_check exceeds the byte bound" return True, direct, "" if not isinstance(direct, str) or len(direct.encode("utf-8")) > MAX_DIRECTIVE_BYTES: return True, None, "reasoning_evidence_check must be a bounded object" try: return True, json.loads(direct), "" except json.JSONDecodeError: return True, None, "reasoning_evidence_check is not valid JSON" for key in TEXT_FIELDS: raw = payload.get(key) if not isinstance(raw, str): continue match = DIRECTIVE_MARKER_RE.search(raw) if not match: continue encoded = raw[match.end() :].strip() if not encoded or len(encoded.encode("utf-8")) > MAX_DIRECTIVE_BYTES: return True, None, "Evidence-Check JSON is empty or exceeds the bound" try: return True, json.loads(encoded), "" except json.JSONDecodeError: return True, None, "Evidence-Check is not valid JSON" return False, None, "" def _normalize_acceptance(item: Any) -> dict[str, Any]: if not isinstance(item, dict): raise ValueError("acceptance predicate must be an object") predicate = item.get("predicate") if predicate not in ALLOWED_ACCEPTANCE_PREDICATES: raise ValueError("acceptance predicate is not allowlisted") if predicate == "outcome_is": if set(item) != {"predicate", "value"} or item.get("value") not in { "success", "error", }: raise ValueError("outcome_is requires success or error") return {"predicate": predicate, "value": item["value"]} if predicate == "json_field_equals": if set(item) != {"predicate", "path", "value"}: raise ValueError("json_field_equals has an invalid shape") path = item.get("path") if ( not isinstance(path, str) or not path or len(path) > 128 or len(path.split(".")) > MAX_EXPECTED_DEPTH or any(not SAFE_PATH_RE.fullmatch(part) for part in path.split(".")) or any(SENSITIVE_KEY_RE.search(part) for part in path.split(".")) ): raise ValueError("json_field_equals path is unsafe") value = _bounded_json(item.get("value")) if isinstance(value, (dict, list)): raise ValueError("json_field_equals value must be scalar") return {"predicate": predicate, "path": path, "value": value} if set(item) != {"predicate", "values"}: raise ValueError(f"{predicate} has an invalid shape") values = item.get("values") if ( not isinstance(values, list) or not values or len(values) > MAX_ACCEPTANCE_PREDICATES or any(not isinstance(value, str) or not value or len(value) > 128 for value in values) ): raise ValueError(f"{predicate} requires bounded string values") return {"predicate": predicate, "values": [_sanitize(value) for value in values]} def _normalize_directive(value: Any) -> dict[str, Any]: if not isinstance(value, dict): raise ValueError("evidence-check directive must be an object") required_keys = { "version", "kind", "claim", "required_evidence", "expected_result", "blocking", } if set(value) != required_keys: raise ValueError("evidence-check directive fields do not match version 1") if value.get("version") != 1 or value.get("kind") != "evidence-check": raise ValueError("unsupported evidence-check version or kind") claim = value.get("claim") if not isinstance(claim, str) or not claim.strip() or len(claim) > MAX_EXCERPT: raise ValueError("evidence-check claim is empty or exceeds the bound") if not isinstance(value.get("blocking"), bool): raise ValueError("evidence-check blocking must be boolean") required = value.get("required_evidence") if ( not isinstance(required, list) or not required or len(required) > MAX_REQUIRED_EVIDENCE ): raise ValueError("required_evidence must be a bounded non-empty list") normalized_required = [] total_predicates = 0 for evidence in required: if not isinstance(evidence, dict) or set(evidence) != { "type", "tool", "acceptance", }: raise ValueError("required evidence has an invalid shape") tool = evidence.get("tool") predicates = evidence.get("acceptance") if ( evidence.get("type") != "tool-result" or not isinstance(tool, str) or not tool.strip() or len(tool) > 128 or not isinstance(predicates, list) or not predicates or len(predicates) > MAX_ACCEPTANCE_PREDICATES ): raise ValueError("required tool evidence is invalid or unbounded") total_predicates += len(predicates) if total_predicates > MAX_ACCEPTANCE_PREDICATES: raise ValueError("evidence-check exceeds the total predicate bound") normalized_required.append( { "type": "tool-result", "tool": _sanitize(tool), "acceptance": [ _normalize_acceptance(predicate) for predicate in predicates ], } ) return { "version": 1, "kind": "evidence-check", "claim": _sanitize(claim), "required_evidence": normalized_required, "expected_result": _bounded_json(value.get("expected_result")), "blocking": value["blocking"], } def _trusted_codex_transcript_path(payload: dict[str, Any]) -> Path | None: codex_home = Path(os.environ.get("CODEX_HOME") or (Path.home() / ".codex")) try: sessions_root = (codex_home / "sessions").resolve() except (OSError, RuntimeError): return None raw_path = payload.get("transcript_path") if isinstance(raw_path, str) and raw_path.strip(): transcript = _trusted_codex_rollout(Path(raw_path), sessions_root) if transcript is not None: return transcript raw_session_id = payload.get("session_id") or payload.get("sessionId") if not isinstance(raw_session_id, str): return None session_id = raw_session_id.strip().lower() if CODEX_SESSION_ID_RE.fullmatch(session_id) is None: return None matches: list[Path] = [] try: candidates = sessions_root.glob( f"*/*/*/rollout-*-{session_id}.jsonl" ) for candidate in candidates: transcript = _trusted_codex_rollout(candidate, sessions_root) if transcript is not None and transcript not in matches: matches.append(transcript) if len(matches) > 1: return None except OSError: return None return matches[0] if matches else None def _trusted_codex_rollout( candidate: Path, sessions_root: Path ) -> Path | None: try: transcript = candidate.resolve(strict=True) transcript.relative_to(sessions_root) except (OSError, RuntimeError, ValueError): return None if ( not transcript.is_file() or transcript.suffix.lower() != ".jsonl" or not transcript.name.startswith("rollout-") ): return None return transcript def _trusted_claude_transcript_path(payload: dict[str, Any]) -> Path | None: """Resolve one host-owned Claude transcript without trusting repo paths.""" claude_home = Path( os.environ.get("CLAUDE_CONFIG_DIR") or (Path.home() / ".claude") ) try: projects_root = (claude_home / "projects").resolve() except (OSError, RuntimeError): return None raw_session_id = payload.get("session_id") or payload.get("sessionId") if not isinstance(raw_session_id, str): return None session_id = raw_session_id.strip().lower() if CLAUDE_SESSION_ID_RE.fullmatch(session_id) is None: return None raw_path = payload.get("transcript_path") if isinstance(raw_path, str) and raw_path.strip(): transcript = _trusted_claude_transcript( Path(raw_path), projects_root, session_id ) if transcript is not None: return transcript matches: list[Path] = [] try: for candidate in projects_root.glob(f"**/{session_id}.jsonl"): transcript = _trusted_claude_transcript( candidate, projects_root, session_id ) if transcript is not None and transcript not in matches: matches.append(transcript) if len(matches) > 1: return None except OSError: return None return matches[0] if matches else None def _trusted_claude_transcript( candidate: Path, projects_root: Path, session_id: str ) -> Path | None: try: transcript = candidate.resolve(strict=True) transcript.relative_to(projects_root) except (OSError, RuntimeError, ValueError): return None if ( not transcript.is_file() or transcript.suffix.lower() != ".jsonl" or transcript.stem.lower() != session_id ): return None return transcript def _visible_summary_item( raw_line: bytes, source_digest: str, line_offset: int ) -> dict[str, Any] | None: """Extract one allowlisted host-visible summary item.""" try: envelope = json.loads(raw_line) except (UnicodeDecodeError, json.JSONDecodeError): return None if not isinstance(envelope, dict) or envelope.get("type") != "response_item": return None reasoning = envelope.get("payload") if not isinstance(reasoning, dict) or reasoning.get("type") != "reasoning": return None summaries = reasoning.get("summary") if not isinstance(summaries, list): return None visible = [ item.get("text", "") for item in summaries if isinstance(item, dict) and item.get("type") == "summary_text" and isinstance(item.get("text"), str) and item.get("text", "").strip() ] if not visible: return None text = _sanitize(" ".join(visible)) return { "text": text, "offset": line_offset, "digest": _digest( { "source": source_digest, "offset": line_offset, "text": text, } ), } def _visible_claude_text_item( raw_line: bytes, source_digest: str, line_offset: int ) -> dict[str, Any] | None: """Extract visible Claude assistant text while rejecting every other block.""" try: envelope = json.loads(raw_line) except (UnicodeDecodeError, json.JSONDecodeError): return None if not isinstance(envelope, dict) or envelope.get("type") != "assistant": return None message = envelope.get("message") if not isinstance(message, dict) or message.get("role") != "assistant": return None content = message.get("content") if not isinstance(content, list): return None visible = [ item.get("text", "") for item in content if isinstance(item, dict) and item.get("type") == "text" and set(item) == {"type", "text"} and isinstance(item.get("text"), str) and item.get("text", "").strip() ] if not visible: return None text = _sanitize( " ".join(_normalize_visible_unicode(item) for item in visible) ) return { "text": text, "offset": line_offset, "digest": _digest( { "source": source_digest, "offset": line_offset, "text": text, } ), } def _scan_codex_rollout( transcript: Path, source_digest: str, *, start: int, recent_limit: int | None = None, item_parser=_visible_summary_item, ) -> tuple[list[dict[str, Any]], int, bool]: """Stream complete JSONL records and return allowlisted visible summaries.""" selected: list[dict[str, Any]] | deque[dict[str, Any]] selected = deque(maxlen=recent_limit) if recent_limit else [] cursor = start complete = True try: with transcript.open("rb") as handle: handle.seek(0, os.SEEK_END) size = handle.tell() if start < 0 or start > size: raise ValueError("stored rollout cursor is outside the current file") handle.seek(start) while True: line_offset = handle.tell() raw_line = handle.readline() if not raw_line: break if not raw_line.endswith(b"\n"): complete = False break cursor = handle.tell() item = item_parser(raw_line, source_digest, line_offset) if item is not None: selected.append(item) except (OSError, ValueError): return [], start, False return list(selected), cursor, complete def _rollout_source_anchor(transcript: Path, cursor: int) -> str | None: """Hash bounded head/cursor bytes so same-path replacement fails safe.""" try: with transcript.open("rb") as handle: handle.seek(0, os.SEEK_END) size = handle.tell() if cursor < 0 or cursor > size: return None handle.seek(0) head = handle.read(min(cursor, CODEX_SOURCE_ANCHOR_BYTES)) tail_start = max(0, cursor - CODEX_SOURCE_ANCHOR_BYTES) handle.seek(tail_start) tail = handle.read(cursor - tail_start) except OSError: return None digest = hashlib.sha256() digest.update(str(cursor).encode("ascii")) digest.update(b"\0") digest.update(head) digest.update(b"\0") digest.update(tail) return digest.hexdigest() def _latest_codex_summary( transcript: Path, source_digest: str, item_parser=_visible_summary_item, ) -> tuple[list[dict[str, Any]], int, bool]: """Keep live-mode startup bounded while retaining an exact append cursor.""" try: with transcript.open("rb") as handle: handle.seek(0, os.SEEK_END) size = handle.tell() start = max(0, size - MAX_CODEX_TRANSCRIPT_BYTES) handle.seek(start) data = handle.read(MAX_CODEX_TRANSCRIPT_BYTES) except OSError: return [], 0, False final_newline = data.rfind(b"\n") if final_newline < 0: return [], 0, False cursor = start + final_newline + 1 data = data[: final_newline + 1] if start: first_newline = data.find(b"\n") if first_newline < 0: return [], cursor, cursor == size start += first_newline + 1 data = data[first_newline + 1 :] lines = data.splitlines(keepends=True) if len(lines) > MAX_CODEX_TRANSCRIPT_LINES: discarded = lines[:-MAX_CODEX_TRANSCRIPT_LINES] start += sum(len(line) for line in discarded) lines = lines[-MAX_CODEX_TRANSCRIPT_LINES:] offsets = [] offset = start for raw_line in lines: offsets.append(offset) offset += len(raw_line) for raw_line, line_offset in reversed(list(zip(lines, offsets))): item = item_parser(raw_line, source_digest, line_offset) if item is not None: return [item], cursor, cursor == size return [], cursor, cursor == size def _empty_ingestion(mode: str, limit: int) -> dict[str, Any]: return { "mode": mode, "limit": limit, "surface": "unknown", "model": "unknown", "effort": "unknown", "metadata_initialized": False, "source_digest": "", "cursor_bytes": 0, "source_anchor_bytes": 0, "source_anchor_digest": "", "covered_from_bytes": 0, "covered_through_bytes": 0, "has_covered_interval": False, "initialized": False, "backfill_complete": False, "historically_complete": False, "summaries_discovered": 0, "summaries_processed": 0, "last_error": "", } def _reset_ingestion_mode( ingestion: dict[str, Any], mode: str, limit: int ) -> dict[str, Any]: """Restart selection while retaining compact exactly-once history.""" reset = _empty_ingestion(mode, limit) for key in ( "surface", "model", "effort", "metadata_initialized", "source_digest", "source_anchor_bytes", "source_anchor_digest", "covered_from_bytes", "covered_through_bytes", "has_covered_interval", "summaries_discovered", "summaries_processed", "historically_complete", ): reset[key] = ingestion.get(key, reset[key]) return reset def _transcript_visible_reasoning_summaries( payload: dict[str, Any], ingestion: dict[str, Any], mode: str, limit: int, *, transcript_resolver, item_parser, surface_host: str, source_label: str, record_label: str, ) -> tuple[list[dict[str, Any]], dict[str, Any]]: """Select initial host history once, then follow appended complete records.""" updated = dict(ingestion) updated["mode"] = mode updated["limit"] = limit transcript = transcript_resolver(payload) if transcript is None: updated["last_error"] = f"trusted {source_label} is unavailable" updated["backfill_complete"] = False updated["historically_complete"] = False return [], updated discovered_surface = "unknown" if _normalized_surface(updated.get("surface"), surface_host) == "unknown": discovered_surface = _transcript_surface(transcript, surface_host) source_digest = _digest(str(transcript)) initialized = bool(updated.get("initialized")) prior_source = updated.get("source_digest", "") if prior_source and prior_source != source_digest: updated["last_error"] = ( f"trusted {source_label} changed during the session" ) updated["backfill_complete"] = False updated["historically_complete"] = False return [], updated try: size = transcript.stat().st_size except OSError: updated["last_error"] = f"trusted {source_label} cannot be inspected" updated["backfill_complete"] = False updated["historically_complete"] = False return [], updated anchor_bytes = int(updated.get("source_anchor_bytes", 0)) anchor_digest = updated.get("source_anchor_digest", "") if prior_source and anchor_digest: if anchor_bytes > size: updated["last_error"] = ( f"trusted {source_label} shrank below the saved cursor" ) updated["backfill_complete"] = False updated["historically_complete"] = False return [], updated current_anchor = _rollout_source_anchor(transcript, anchor_bytes) if current_anchor is None: updated["last_error"] = ( f"trusted {source_label} anchor cannot be inspected" ) updated["backfill_complete"] = False updated["historically_complete"] = False return [], updated if current_anchor != anchor_digest: updated["last_error"] = ( f"trusted {source_label} changed below the saved cursor" ) updated["backfill_complete"] = False updated["historically_complete"] = False return [], updated if initialized and int(updated.get("cursor_bytes", 0)) > size: updated["last_error"] = ( f"trusted {source_label} shrank below the saved cursor" ) updated["backfill_complete"] = False updated["historically_complete"] = False return [], updated if discovered_surface != "unknown": updated["surface"] = discovered_surface if initialized: metadata_start = ( int(updated.get("cursor_bytes", 0)) if updated.get("metadata_initialized") else 0 ) raw_summaries, cursor, complete = _scan_codex_rollout( transcript, source_digest, start=int(updated.get("cursor_bytes", 0)), item_parser=item_parser, ) elif mode == "live": metadata_start = 0 raw_summaries, cursor, complete = _latest_codex_summary( transcript, source_digest, item_parser=item_parser ) else: metadata_start = 0 raw_summaries, cursor, complete = _scan_codex_rollout( transcript, source_digest, start=0, recent_limit=limit if mode == "recent" else None, item_parser=item_parser, ) observed_model, observed_effort, metadata_complete = _transcript_model_effort( transcript, surface_host, start=metadata_start, end=cursor, ) if observed_model != "unknown": updated["model"] = observed_model if observed_effort != "unknown": updated["effort"] = observed_effort if metadata_complete: updated["metadata_initialized"] = True prior_has_interval = bool(updated.get("has_covered_interval")) prior_from = int(updated.get("covered_from_bytes", 0)) prior_through = int(updated.get("covered_through_bytes", 0)) summaries = [ item for item in raw_summaries if not ( prior_source == source_digest and prior_has_interval and prior_from <= int(item.get("offset", -1)) < prior_through ) ] interval_from: int | None = None if initialized: if prior_has_interval: interval_from = prior_from elif raw_summaries: interval_from = min(int(item["offset"]) for item in raw_summaries) elif mode == "full": interval_from = 0 elif raw_summaries: interval_from = min(int(item["offset"]) for item in raw_summaries) if interval_from is not None: if prior_source == source_digest and prior_has_interval: interval_from = min(interval_from, prior_from) updated["covered_from_bytes"] = interval_from updated["covered_through_bytes"] = max(cursor, prior_through) updated["has_covered_interval"] = True scan_error = ( "" if complete else f"{record_label} ended with an incomplete record" ) anchor = _rollout_source_anchor(transcript, cursor) if anchor is None: complete = False scan_error = f"trusted {source_label} anchor cannot be inspected" updated["source_digest"] = source_digest updated["cursor_bytes"] = cursor updated["source_anchor_bytes"] = cursor updated["source_anchor_digest"] = anchor or "" updated["initialized"] = True updated["backfill_complete"] = complete updated["historically_complete"] = complete and ( mode == "full" or bool(updated.get("historically_complete")) ) updated["summaries_discovered"] = int( updated.get("summaries_discovered", 0) ) + len(summaries) updated["last_error"] = scan_error return summaries, updated def _codex_visible_reasoning_summaries( payload: dict[str, Any], ingestion: dict[str, Any], mode: str, limit: int, ) -> tuple[list[dict[str, Any]], dict[str, Any]]: """REQ-175: read explicit visible Codex summary records.""" return _transcript_visible_reasoning_summaries( payload, ingestion, mode, limit, transcript_resolver=_trusted_codex_transcript_path, item_parser=_visible_summary_item, surface_host="codex", source_label="Codex rollout", record_label="rollout", ) def _claude_visible_reasoning_summaries( payload: dict[str, Any], ingestion: dict[str, Any], mode: str, limit: int, ) -> tuple[list[dict[str, Any]], dict[str, Any]]: """REQ-176: read only visible Claude assistant text records.""" return _transcript_visible_reasoning_summaries( payload, ingestion, mode, limit, transcript_resolver=_trusted_claude_transcript_path, item_parser=_visible_claude_text_item, surface_host="claude", source_label="Claude transcript", record_label="transcript", ) def _with_visible_reasoning_summaries( host: str, payload: dict[str, Any], ingestion: dict[str, Any], backfill_mode: str, backfill_limit: int, ) -> tuple[dict[str, Any], dict[str, Any]]: enriched = dict(payload) summaries: list[dict[str, Any]] = [] updated = dict(ingestion) if str(host).lower() == "codex": summaries, updated = _codex_visible_reasoning_summaries( payload, ingestion, backfill_mode, backfill_limit ) elif str(host).lower() == "claude": summaries, updated = _claude_visible_reasoning_summaries( payload, ingestion, backfill_mode, backfill_limit ) if not summaries and payload.get("reasoning_summary"): text = _sanitize(payload["reasoning_summary"]) summaries = [ { "text": text, "digest": _digest( { "host": host, "session_id": _session_id(payload), "event_id": payload.get("event_id") or payload.get("tool_use_id") or _digest(payload), "text": text, } ), } ] updated["mode"] = "live" updated["limit"] = backfill_limit updated["initialized"] = True updated["backfill_complete"] = False updated["historically_complete"] = False updated["summaries_discovered"] = int( updated.get("summaries_discovered", 0) ) + 1 updated["last_error"] = "" if summaries: enriched["reasoning_summary"] = summaries[-1]["text"] enriched["_reasoning_summaries"] = summaries return enriched, updated def _visible_text(payload: dict[str, Any]) -> str: parts = [_sanitize(payload[key]) for key in TEXT_FIELDS if payload.get(key)] return " ".join(part for part in parts if part)[:MAX_EXCERPT] def _tool_text(payload: dict[str, Any]) -> str: parts = [ _sanitize(payload.get("tool_name", "")), _sanitize(payload.get("tool_input", "")), ] return " ".join(parts)[:MAX_EXCERPT] def _tool_output(payload: dict[str, Any]) -> str: for key in ("tool_response", "tool_output", "output", "result"): if payload.get(key) not in (None, "", {}, []): return _sanitize(payload[key]) return "" def _tool_result_value(payload: dict[str, Any]) -> Any: for key in ("tool_response", "tool_output", "output", "result"): if key in payload and payload[key] is not None: return payload[key] return None def _tool_name(payload: dict[str, Any]) -> str: return _sanitize(payload.get("tool_name", "")).casefold() def _tool_outcome(payload: dict[str, Any], result: Any) -> str: if payload.get("is_error") is True or payload.get("error"): return "error" values = [payload, result] if isinstance(result, dict) else [payload] for value in values: if not isinstance(value, dict): continue if value.get("is_error") is True or value.get("error"): return "error" status = str(value.get("status", "")).casefold() if status in {"error", "failed", "failure", "cancelled"}: return "error" exit_code = value.get("exit_code", value.get("returncode")) if isinstance(exit_code, int) and not isinstance(exit_code, bool) and exit_code != 0: return "error" return "success" def _json_path_value(value: Any, path: str) -> tuple[bool, Any]: current = value for part in path.split("."): if not isinstance(current, dict) or part not in current: return False, None current = current[part] return True, current def _evaluate_acceptance( predicate: dict[str, Any], payload: dict[str, Any], result: Any, outcome: str, ) -> bool: kind = predicate["predicate"] if kind == "outcome_is": return outcome == predicate["value"] if kind == "json_field_equals": found, observed = _json_path_value(result, predicate["path"]) return found and observed == predicate["value"] output = _tool_output(payload).casefold() values = [value.casefold() for value in predicate["values"]] if kind == "output_contains_all": return all(value in output for value in values) if kind == "output_contains_none": return all(value not in output for value in values) return False def _read_config(root: Path) -> dict[str, Any]: path = root / ".prd_plugin/config.json" try: value = json.loads(path.read_text(encoding="utf-8")) return value if isinstance(value, dict) else {} except (OSError, json.JSONDecodeError): return {} def _configured_modes( root: Path, ) -> tuple[str, dict[str, str], str, int]: guard = _read_config(root).get("reasoning_guard", {}) if not isinstance(guard, dict): guard = {} mode = guard.get("mode", "report") if mode not in VALID_MODES: mode = "report" raw_categories = guard.get("categories", {}) if not isinstance(raw_categories, dict): raw_categories = {} categories = {} for category in CATEGORIES: value = raw_categories.get(category, "inherit") categories[category] = value if value in VALID_CATEGORY_MODES else "inherit" raw_backfill = guard.get("backfill", {}) if not isinstance(raw_backfill, dict): raw_backfill = {} backfill_mode = raw_backfill.get("mode", DEFAULT_BACKFILL_MODE) if backfill_mode not in VALID_BACKFILL_MODES: backfill_mode = DEFAULT_BACKFILL_MODE backfill_limit = raw_backfill.get("limit", DEFAULT_BACKFILL_LIMIT) if ( not isinstance(backfill_limit, int) or isinstance(backfill_limit, bool) or not 1 <= backfill_limit <= MAX_BACKFILL_LIMIT ): backfill_limit = DEFAULT_BACKFILL_LIMIT return mode, categories, backfill_mode, backfill_limit def _effective_mode(global_mode: str, categories: dict[str, str], category: str) -> str: if global_mode == "off": return "off" override = categories.get(category, "inherit") return global_mode if override == "inherit" else override def _coverage( host: str, payload: dict[str, Any], ingestion: dict[str, Any], previous: dict[str, Any] | None = None, ) -> dict[str, Any]: observed = {"visible hook events"} if isinstance(previous, dict): observed.update( item for item in previous.get("observed", []) if isinstance(item, str) and item in COVERAGE_SIGNALS ) if payload.get("reasoning_summary"): observed.add("visible reasoning summaries") if payload.get("tool_input"): observed.add("tool inputs") if _tool_output(payload): observed.add("tool results") missing = [] ingestion_mode = ingestion.get("mode", "live") historically_complete = bool(ingestion.get("historically_complete")) ingestion_error = bool(ingestion.get("last_error")) if ingestion_error: level = "reduced" elif historically_complete: level = "historically_complete" elif "visible reasoning summaries" not in observed: level = "reduced" elif ingestion_mode == "recent": level = "recent_backfill" else: level = "live_full" if level == "reduced" and "visible reasoning summaries" not in observed: missing.append( "no visible reasoning summaries have reached the guard in this session" ) elif level == "live_full": missing.append( "historical visible reasoning summaries were not backfilled in live mode" ) elif level == "recent_backfill": missing.append( "historical visible reasoning coverage is limited to the last " f"{ingestion.get('limit', DEFAULT_BACKFILL_LIMIT)} summaries" ) if ingestion.get("last_error"): missing.append(_sanitize(ingestion["last_error"])) return { "host": host or "unknown", "level": level, "observed": [item for item in COVERAGE_SIGNALS if item in observed], "missing": missing, "history": { "mode": ingestion_mode, "limit": ingestion.get("limit", DEFAULT_BACKFILL_LIMIT), "backfill_complete": bool(ingestion.get("backfill_complete")), "historically_complete": historically_complete, }, } def _empty_state( mode: str, host: str, payload: dict[str, Any], backfill_mode: str = DEFAULT_BACKFILL_MODE, backfill_limit: int = DEFAULT_BACKFILL_LIMIT, ) -> dict[str, Any]: ingestion = _empty_ingestion(backfill_mode, backfill_limit) model, effort = _session_model_effort(payload, ingestion) return { "schema_version": SCHEMA_VERSION, "mode": mode, "surface": _session_surface(host, payload, ingestion), "model": model, "effort": effort, "session_id": _sanitize( payload.get("session_id") or payload.get("sessionId") or "" ), "turn_id": _sanitize(payload.get("turn_id") or payload.get("turnId") or ""), "started_at": _utc_now(), "updated_at": _utc_now(), "coverage": _coverage(host, payload, ingestion), "obligations": [], "violations": [], "guard_actions": [], "processed_event_digests": [], "processed_summary_digests": [], "reasoning_summary_ingestion": ingestion, "diagnostics": [], "metrics": { "samples_ms": [], "p50_ms": 0.0, "p95_ms": 0.0, "classification": { "summaries_seen": 0, "candidates": 0, "created": 0, "reconciled": 0, "ignored": 0, "uncertain": 0, }, }, "summary": {"open": 0, "deferred": 0, "violations": 0, "blocked": 0}, } def _valid_state(value: Any) -> bool: if not isinstance(value, dict) or value.get("schema_version") != SCHEMA_VERSION: return False list_fields = ( "obligations", "violations", "guard_actions", "processed_event_digests", "processed_summary_digests", "diagnostics", ) if any(not isinstance(value.get(field), list) for field in list_fields): return False if ( "surface" in value and value.get("surface") not in VALID_SURFACES ): return False if ( ("model" in value and _normalized_model(value.get("model")) != value.get("model")) or ( "effort" in value and _normalized_effort(value.get("effort")) != value.get("effort") ) ): return False valid_states = OPEN_STATES | {"RESOLVED", "DEFERRED"} if any( not isinstance(item, dict) or item.get("state") not in valid_states or not isinstance(item.get("kind"), str) or not isinstance(item.get("claim"), str) or not isinstance(item.get("required_evidence"), list) or not isinstance(item.get("observed_evidence"), list) or not isinstance(item.get("blocking"), bool) or not isinstance(item.get("source"), dict) for item in value["obligations"] ): return False if any(not isinstance(item, dict) for item in value["violations"]): return False if any( not isinstance(item, str) for field in ("processed_event_digests", "processed_summary_digests") for item in value[field] ): return False ingestion = value.get("reasoning_summary_ingestion") if ( not isinstance(ingestion, dict) or ingestion.get("mode") not in VALID_BACKFILL_MODES or not isinstance(ingestion.get("limit"), int) or isinstance(ingestion.get("limit"), bool) or not 1 <= ingestion.get("limit") <= MAX_BACKFILL_LIMIT or not isinstance(ingestion.get("source_digest"), str) or not isinstance(ingestion.get("cursor_bytes"), int) or ingestion.get("cursor_bytes") < 0 or not isinstance(ingestion.get("source_anchor_bytes"), int) or ingestion.get("source_anchor_bytes") < 0 or not isinstance(ingestion.get("source_anchor_digest"), str) or not isinstance(ingestion.get("covered_from_bytes"), int) or ingestion.get("covered_from_bytes") < 0 or not isinstance(ingestion.get("covered_through_bytes"), int) or ingestion.get("covered_through_bytes") < 0 or any( not isinstance(ingestion.get(key), bool) for key in ( "initialized", "backfill_complete", "historically_complete", "has_covered_interval", ) ) or ( "metadata_initialized" in ingestion and not isinstance(ingestion.get("metadata_initialized"), bool) ) or any( not isinstance(ingestion.get(key), int) or isinstance(ingestion.get(key), bool) or ingestion.get(key) < 0 for key in ("summaries_discovered", "summaries_processed") ) or not isinstance(ingestion.get("last_error"), str) or ( "surface" in ingestion and ingestion.get("surface") not in VALID_SURFACES ) or ( "model" in ingestion and _normalized_model(ingestion.get("model")) != ingestion.get("model") ) or ( "effort" in ingestion and _normalized_effort(ingestion.get("effort")) != ingestion.get("effort") ) ): return False metrics = value.get("metrics") if not isinstance(metrics, dict) or not isinstance(metrics.get("samples_ms"), list): return False if any(not isinstance(sample, (int, float)) for sample in metrics["samples_ms"]): return False classification = metrics.get("classification") if classification is not None and ( not isinstance(classification, dict) or any( not isinstance(classification.get(key), int) or isinstance(classification.get(key), bool) or classification.get(key) < 0 for key in ( "summaries_seen", "candidates", "created", "reconciled", "ignored", "uncertain", ) ) ): return False return True def _valid_legacy_state(value: Any) -> bool: if ( not isinstance(value, dict) or value.get("schema_version") != LEGACY_SCHEMA_VERSION ): return False list_fields = ( "obligations", "violations", "guard_actions", "processed_event_digests", "diagnostics", ) if any(not isinstance(value.get(field), list) for field in list_fields): return False valid_states = OPEN_STATES | {"RESOLVED", "DEFERRED"} if any( not isinstance(item, dict) or item.get("state") not in valid_states for item in value["obligations"] ): return False if any(not isinstance(item, dict) for item in value["violations"]): return False if any(not isinstance(item, str) for item in value["processed_event_digests"]): return False metrics = value.get("metrics") return bool( isinstance(metrics, dict) and isinstance(metrics.get("samples_ms"), list) and all( isinstance(sample, (int, float)) for sample in metrics.get("samples_ms", []) ) ) def _migrate_legacy_state(value: dict[str, Any]) -> dict[str, Any]: state = json.loads(json.dumps(value, ensure_ascii=True)) migrated_obligations = [] for legacy in state.get("obligations", [])[-MAX_RECORDS:]: obligation = dict(legacy) claim = _sanitize(obligation.get("summary", "")) observed_evidence = [] if obligation.get("evidence_digest"): observed_evidence.append( { "tool_use_id": _sanitize(obligation.get("tool_use_id", "")), "tool_name": "", "outcome": "success", "evidence_digest": _sanitize( obligation.get("evidence_digest", "") ), "acceptance_passed": obligation.get("state") in {"EVIDENCED", "RESOLVED"}, "predicate_results": [], "observed_at": _sanitize( obligation.get("created_at", "") ), "source": "legacy_migration", } ) obligation.update( { "kind": "evidence-check", "claim": claim, "summary": claim, "required_evidence": [], "blocking": False, "observed_evidence": observed_evidence, "source": {"type": "legacy_migration", "confidence": "low"}, } ) migrated_obligations.append(obligation) state["schema_version"] = SCHEMA_VERSION state["obligations"] = migrated_obligations default_obligation_id = ( migrated_obligations[0].get("id", "") if len(migrated_obligations) == 1 else "" ) migrated_findings: list[dict[str, Any]] = [] findings_by_key: dict[tuple[str, str], dict[str, Any]] = {} for legacy in state.get("violations", [])[-MAX_RECORDS:]: category = legacy.get("category", "") code = VIOLATION_CODES.get(category, "RG-LEGACY-FINDING-001") obligation_id = default_obligation_id key = (code, obligation_id) source_digest = _sanitize(legacy.get("source_event_digest", "")) at = _sanitize(legacy.get("at", "")) or _utc_now() existing = findings_by_key.get(key) if existing: existing["occurrences"] += 1 existing["last_seen"] = at if source_digest and source_digest not in existing["source_event_digests"]: existing["source_event_digests"].append(source_digest) continue finding = { "code": code, "category": _sanitize(category), "mode": _sanitize(legacy.get("mode", "report")) or "report", "reason": _sanitize(legacy.get("reason", "Legacy Reason Guard finding")), "obligation_id": obligation_id, "obligation": _sanitize( migrated_obligations[0].get("claim", "") if default_obligation_id else "" ), "occurrences": 1, "first_seen": at, "last_seen": at, "active": True, "resolved_at": "", "source_event_digests": [source_digest] if source_digest else [], "source_event_digest": source_digest, "at": at, } findings_by_key[key] = finding migrated_findings.append(finding) state["violations"] = migrated_findings[-MAX_RECORDS:] state["processed_summary_digests"] = [] state["reasoning_summary_ingestion"] = _empty_ingestion( DEFAULT_BACKFILL_MODE, DEFAULT_BACKFILL_LIMIT ) state.setdefault("diagnostics", []).append( { "code": "state_migrated", "detail": f"Reason Guard state migrated from {LEGACY_SCHEMA_VERSION} to {SCHEMA_VERSION}", "at": _utc_now(), } ) state["diagnostics"] = state["diagnostics"][-MAX_RECORDS:] return state def _migrate_previous_state( value: dict[str, Any], backfill_mode: str, backfill_limit: int ) -> dict[str, Any]: """Add the REQ-175 ingestion ledger without discarding schema-2 evidence.""" state = json.loads(json.dumps(value, ensure_ascii=True)) state["schema_version"] = SCHEMA_VERSION state["processed_summary_digests"] = [] state["reasoning_summary_ingestion"] = _empty_ingestion( backfill_mode, backfill_limit ) classification = state.get("metrics", {}).get("classification") if isinstance(classification, dict): classification.setdefault("uncertain", 0) state.setdefault("diagnostics", []).append( { "code": "state_migrated", "detail": ( "Reason Guard state migrated from " f"{PREVIOUS_SCHEMA_VERSION} to {SCHEMA_VERSION}" ), "at": _utc_now(), } ) state["diagnostics"] = state["diagnostics"][-MAX_RECORDS:] return state @contextmanager def _state_lock(root: Path): lock_path = root / LOCK_RELATIVE_PATH lock_path.parent.mkdir(parents=True, exist_ok=True) deadline = time.monotonic() + 3.0 while True: try: lock_path.mkdir() break except (FileExistsError, PermissionError): if not lock_path.exists(): continue try: if time.time() - lock_path.stat().st_mtime > 30: lock_path.rmdir() continue except (FileNotFoundError, OSError): pass if time.monotonic() >= deadline: raise TimeoutError("reason guard state lock remained busy") time.sleep(0.02) try: yield finally: try: lock_path.rmdir() except FileNotFoundError: pass def _normalized_host(host: str) -> str: value = _sanitize(host).lower() return value if value in HOST_REPORT_FILENAMES else "unknown" def _host_report_relative_path(host: str) -> Path: normalized = _normalized_host(host) filename = HOST_REPORT_FILENAMES.get( normalized, f"{normalized}_reasoning_guard.json" ) return Path(".prd_plugin/local") / filename def _empty_host_ledger(host: str) -> dict[str, Any]: return { "schema_version": LEDGER_SCHEMA_VERSION, "kind": "reason_guard_host_ledger", "host": _normalized_host(host), "active_session_id": "", "updated_at": _utc_now(), "sessions": [], } def _valid_host_ledger(value: Any, host: str) -> bool: if ( not isinstance(value, dict) or value.get("schema_version") != LEDGER_SCHEMA_VERSION or value.get("kind") != "reason_guard_host_ledger" or value.get("host") != _normalized_host(host) or not isinstance(value.get("active_session_id"), str) or not isinstance(value.get("updated_at"), str) or not isinstance(value.get("sessions"), list) ): return False sessions = value["sessions"] if len(sessions) > MAX_SESSION_HISTORY: return False session_ids: set[str] = set() for state in sessions: if not _valid_state(state): return False session_id = state.get("session_id") if not isinstance(session_id, str) or session_id in session_ids: return False session_ids.add(session_id) return True def _state_host(state: dict[str, Any]) -> str: coverage = state.get("coverage") if not isinstance(coverage, dict): return "unknown" return _normalized_host(str(coverage.get("host", ""))) def _load_state( root: Path, mode: str, host: str, payload: dict[str, Any], backfill_mode: str, backfill_limit: int, ) -> tuple[dict[str, Any], bool]: path = root / REPORT_RELATIVE_PATH try: value = json.loads(path.read_text(encoding="utf-8")) except FileNotFoundError: return _empty_state( mode, host, payload, backfill_mode, backfill_limit ), False except (OSError, json.JSONDecodeError) as exc: state = _empty_state( mode, host, payload, backfill_mode, backfill_limit ) state["diagnostics"].append( { "code": "malformed_state_recovered", "detail": _sanitize(type(exc).__name__), "at": _utc_now(), } ) return state, True if value.get("schema_version") == LEGACY_SCHEMA_VERSION: if _valid_legacy_state(value): migrated = _migrate_legacy_state(value) if _valid_state(migrated): return migrated, False if value.get("schema_version") == PREVIOUS_SCHEMA_VERSION: migrated = _migrate_previous_state( value, backfill_mode, backfill_limit ) if _valid_state(migrated): return migrated, False if value.get("schema_version") == SCHEMA_VERSION: classification = value.get("metrics", {}).get("classification") if isinstance(classification, dict): classification.setdefault("uncertain", 0) if not _valid_state(value): state = _empty_state( mode, host, payload, backfill_mode, backfill_limit ) state["diagnostics"].append( { "code": "invalid_state_recovered", "detail": f"Stored state did not match schema {SCHEMA_VERSION}", "at": _utc_now(), } ) return state, True return value, False def _atomic_write_path(path: Path, value: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) handle = tempfile.NamedTemporaryFile( "w", encoding="utf-8", newline="\n", dir=path.parent, prefix=f".{path.stem}-", suffix=".tmp", delete=False, ) temp_path = Path(handle.name) try: with handle: json.dump(value, handle, indent=2, sort_keys=True, ensure_ascii=True) handle.write("\n") handle.flush() os.fsync(handle.fileno()) os.replace(temp_path, path) finally: try: temp_path.unlink() except FileNotFoundError: pass def _atomic_write(root: Path, state: dict[str, Any]) -> None: _atomic_write_path(root / REPORT_RELATIVE_PATH, state) def _load_host_state( root: Path, mode: str, host: str, payload: dict[str, Any], backfill_mode: str, backfill_limit: int, ) -> tuple[dict[str, Any], dict[str, Any], bool]: """Load one host ledger and select the current session without overwrites.""" normalized_host = _normalized_host(host) host_path = root / _host_report_relative_path(normalized_host) malformed = False try: ledger_value = json.loads(host_path.read_text(encoding="utf-8")) except FileNotFoundError: ledger = _empty_host_ledger(normalized_host) except (OSError, json.JSONDecodeError) as exc: ledger = _empty_host_ledger(normalized_host) malformed = True ledger["diagnostic"] = { "code": "malformed_host_ledger_recovered", "detail": _sanitize(type(exc).__name__), "at": _utc_now(), } else: if _valid_host_ledger(ledger_value, normalized_host): ledger = ledger_value else: ledger = _empty_host_ledger(normalized_host) malformed = True ledger["diagnostic"] = { "code": "invalid_host_ledger_recovered", "detail": ( "Stored host ledger did not match schema " f"{LEDGER_SCHEMA_VERSION}" ), "at": _utc_now(), } legacy_state, legacy_malformed = _load_state( root, mode, normalized_host, payload, backfill_mode, backfill_limit ) malformed = malformed or legacy_malformed current_session_id = _session_id(payload) legacy_host_matches = _state_host(legacy_state) == normalized_host legacy_session_id = str(legacy_state.get("session_id", "")) if ( legacy_host_matches and not any( item.get("session_id", "") == legacy_session_id for item in ledger["sessions"] ) ): ledger["sessions"].append(legacy_state) selected = next( ( item for item in ledger["sessions"] if item.get("session_id", "") == current_session_id ), None, ) legacy_matches = ( legacy_host_matches and legacy_state.get("session_id", "") == current_session_id ) if legacy_matches and ( selected is None or str(legacy_state.get("updated_at", "")) >= str(selected.get("updated_at", "")) ): selected = legacy_state elif selected is None: selected = _empty_state( mode, normalized_host, payload, backfill_mode, backfill_limit ) selected.setdefault("started_at", selected.get("updated_at") or _utc_now()) if malformed: selected.setdefault("diagnostics", []).append( { "code": "host_ledger_recovered", "detail": "Reason Guard host/session state required safe recovery", "at": _utc_now(), } ) selected["diagnostics"] = selected["diagnostics"][-MAX_RECORDS:] return ledger, selected, malformed def _write_host_state( root: Path, host: str, ledger: dict[str, Any], state: dict[str, Any], ) -> Path: """Commit the host ledger first, then refresh the legacy current view.""" session_id = state.get("session_id", "") sessions = [ item for item in ledger.get("sessions", []) if item.get("session_id", "") != session_id ] sessions.append(state) ledger = { "schema_version": LEDGER_SCHEMA_VERSION, "kind": "reason_guard_host_ledger", "host": _normalized_host(host), "active_session_id": session_id, "updated_at": state.get("updated_at") or _utc_now(), "sessions": sessions[-MAX_SESSION_HISTORY:], } host_path = root / _host_report_relative_path(host) _atomic_write_path(host_path, ledger) _atomic_write(root, state) return host_path def _event_digest(event: str, payload: dict[str, Any]) -> str: stable_id = payload.get("event_id") or payload.get("tool_use_id") if stable_id: return _digest( { "event": event, "id": _sanitize(stable_id), "session_id": _session_id(payload), } ) return _digest({"event": event, "payload": payload}) def _session_id(payload: dict[str, Any]) -> str: return _sanitize(payload.get("session_id") or payload.get("sessionId") or "") def _same_session(obligation: dict[str, Any], session_id: str) -> bool: return obligation.get("session_id", "") == session_id def _latest_open( state: dict[str, Any], session_id: str ) -> dict[str, Any] | None: for obligation in reversed(state["obligations"]): if ( obligation.get("state") in OPEN_STATES and _same_session(obligation, session_id) ): return obligation return None def _consequential_tokens(text: str) -> set[str]: return { match.group(1).lower() for match in CONSEQUENTIAL_RE.finditer(text or "") if match.group(1) } def _check_actions(text: str) -> set[str]: actions = set() for match in CHECK_RE.finditer(text or ""): value = match.group(1).lower() if value.startswith("compar"): actions.add("compare") elif value.startswith("benchmark"): actions.add("benchmark") elif value.startswith("test"): actions.add("test") elif value.startswith("verif"): actions.add("verify") elif value.startswith("measur"): actions.add("measure") elif value.startswith("inspect"): actions.add("inspect") elif value.startswith("reproduc"): actions.add("reproduce") elif value.startswith("check"): actions.add("check") else: actions.add(value) return actions def _evidence_subject_tokens(text: str) -> set[str]: return { word.lower() for word in re.findall(r"[A-Za-z][A-Za-z0-9_-]{3,}", text or "") if word.lower() not in EVIDENCE_SUBJECT_STOP_WORDS } def _heuristic_evidence_affinity(claim: str, tool_text: str) -> bool: return bool( _check_actions(claim) & _check_actions(tool_text) and _evidence_subject_tokens(claim) & _evidence_subject_tokens(tool_text) ) def _expected_result(text: str) -> str: match = re.search(r"\bexpected:\s*(.+?)(?:[.;]|$)", text, re.IGNORECASE) return _sanitize(match.group(1)) if match else "" def _has_uncertain_active_summary(text: str) -> bool: """Detect active-looking visible summaries without treating them as facts.""" segments = [ match.group(1) for match in HEADING_SEGMENT_RE.finditer(text or "") ] if not segments: segments = re.split(r"[.;]\s*", text or "") return any( UNCERTAIN_ACTIVITY_RE.search(segment) and not ROUTINE_PRESENTATION_RE.search(segment) for segment in segments ) def _matching_obligation( state: dict[str, Any], text: str, session_id: str ) -> dict[str, Any] | None: words = { word.lower() for word in re.findall(r"[A-Za-z][A-Za-z0-9_-]{4,}", text) if word.lower() not in { "before", "would", "could", "should", "compare", "comparing", "possible", "causes", "either", "changing", } } for obligation in reversed(state["obligations"]): if ( obligation.get("state") not in OPEN_STATES or not _same_session(obligation, session_id) ): continue prior = { word.lower() for word in re.findall( r"[A-Za-z][A-Za-z0-9_-]{4,}", obligation.get("summary", "") ) } if len(words & prior) >= 2: return obligation return None def _record_engine_diagnostic( state: dict[str, Any], code: str, detail: str ) -> None: sanitized = _sanitize(detail) if any( item.get("code") == code and item.get("detail") == sanitized for item in state["diagnostics"] if isinstance(item, dict) ): return state["diagnostics"].append( {"code": code, "detail": sanitized, "at": _utc_now()} ) state["diagnostics"] = state["diagnostics"][-MAX_RECORDS:] def _record_explicit_obligation( state: dict[str, Any], payload: dict[str, Any], event_digest: str, session_id: str, turn_id: str, ) -> bool: present, candidate, error = _directive_candidate(payload) if not present: return False try: directive = _normalize_directive(candidate) if not error else None except ValueError as exc: directive = None error = str(exc) if directive is None: _record_engine_diagnostic( state, "invalid_evidence_check_directive", error or "Evidence-Check directive is invalid", ) return True claim = directive["claim"] existing = _matching_obligation(state, claim, session_id) if existing and existing.get("source", {}).get("type") == "explicit_directive": return True state["obligations"].append( { "id": f"obl-{event_digest[:12]}", "kind": directive["kind"], "claim": claim, "summary": claim, "required_evidence": directive["required_evidence"], "expected_result": directive["expected_result"], "blocking": directive["blocking"], "observed_evidence": [], "source": {"type": "explicit_directive", "confidence": "high"}, "state": "PLANNED", "history": ["PLANNED"], "created_at": _utc_now(), "source_event_digest": event_digest, "session_id": session_id, "turn_id": turn_id, "tool_use_id": "", "evidence_digest": "", "observed_result": "", } ) state["obligations"] = state["obligations"][-MAX_RECORDS:] return True def _classify_obligation(text: str) -> str: """Classify one bounded visible text without allocating state (REQ-174).""" active_check = bool(ACTIVE_CHECK_RE.search(text or "")) if not text or not CONSEQUENTIAL_RE.search(text): return "" segments = [ segment.strip() for segment in re.split(r"(?<=[.!?])\s+", text) if segment.strip() ] for segment in segments: if ( CONSEQUENTIAL_RE.search(segment) and COMMITMENT_RE.search(segment) and CHECK_RE.search(segment) and DECISION_RE.search(segment) ): return "PLANNED" if CONCLUSION_PREFIX_RE.search(text): return "" for segment in segments: if not CONSEQUENTIAL_RE.search(segment): continue if ( ALTERNATIVES_RE.search(segment) and CHECK_RE.search(segment) and DECISION_RE.search(segment) ): return "IDENTIFIED" if active_check and ACTIVE_DECISION_RE.search(text): return "PLANNED" return "" def _record_obligation( state: dict[str, Any], text: str, event_digest: str, session_id: str, turn_id: str, ) -> str: target_state = _classify_obligation(text) if not target_state: return "uncertain" if _has_uncertain_active_summary(text) else "ignored" existing = _matching_obligation(state, text, session_id) if existing: if target_state == "PLANNED" and existing.get("state") == "IDENTIFIED": existing["state"] = "PLANNED" existing.setdefault("history", ["IDENTIFIED"]).append("PLANNED") existing["summary"] = _sanitize(text) existing["expected_result"] = ( _expected_result(text) or existing.get("expected_result", "") ) return "reconciled" state["obligations"].append( { "id": f"obl-{event_digest[:12]}", "kind": "evidence-check", "claim": _sanitize(text), "summary": _sanitize(text), "required_evidence": [], "blocking": False, "observed_evidence": [], "source": {"type": "heuristic", "confidence": "low"}, "state": target_state, "history": [target_state], "created_at": _utc_now(), "source_event_digest": event_digest, "session_id": session_id, "turn_id": turn_id, "tool_use_id": "", "evidence_digest": "", "expected_result": _expected_result(text), "observed_result": "", } ) state["obligations"] = state["obligations"][-MAX_RECORDS:] return "created" def _record_classification_metric(state: dict[str, Any], outcome: str) -> None: metrics = state.setdefault("metrics", {}) classification = metrics.setdefault( "classification", { "summaries_seen": 0, "candidates": 0, "created": 0, "reconciled": 0, "ignored": 0, "uncertain": 0, }, ) classification["summaries_seen"] += 1 if outcome in {"created", "reconciled"}: classification["candidates"] += 1 classification[outcome] += 1 elif outcome == "uncertain": classification["uncertain"] += 1 else: classification["ignored"] += 1 def _record_deferral( state: dict[str, Any], text: str, session_id: str ) -> bool: match = re.search( r"Deferred:\s*(?P.+?);\s*reason:\s*(?P.+?);\s*" r"decision impact:\s*(?P.+?);\s*reopen when:\s*(?P.+?)(?:\.|$)", text, re.IGNORECASE, ) if not match: return False obligation = _latest_open(state, session_id) if not obligation or not all(value.strip() for value in match.groupdict().values()): return True obligation["state"] = "DEFERRED" obligation.setdefault("history", []).append("DEFERRED") values = {key: _sanitize(value) for key, value in match.groupdict().items()} obligation["deferral"] = { "item": values["item"], "reason": values["reason"], "decision_impact": values["impact"], "reopen_when": values["reopen"], } return True def _mark_execution( state: dict[str, Any], payload: dict[str, Any], session_id: str ) -> None: tool_name = _tool_name(payload) explicit_candidates = [ obligation for obligation in state["obligations"] if obligation.get("state") in OPEN_STATES and _same_session(obligation, session_id) and obligation.get("source", {}).get("type") == "explicit_directive" and any( str(evidence.get("tool", "")).casefold() == tool_name for evidence in obligation.get("required_evidence", []) if isinstance(evidence, dict) ) ] if explicit_candidates: obligation = explicit_candidates[-1] obligation["state"] = "EXECUTED" if not obligation.get("history") or obligation["history"][-1] != "EXECUTED": obligation.setdefault("history", []).append("EXECUTED") obligation["tool_use_id"] = _sanitize( payload.get("tool_use_id") or payload.get("tool_call_id") or "" ) return tool_text = _tool_text(payload) candidates = [ obligation for obligation in state["obligations"] if obligation.get("state") in OPEN_STATES and _same_session(obligation, session_id) and _heuristic_evidence_affinity( obligation.get("summary", ""), tool_text ) ] if not candidates: return obligation = candidates[-1] obligation["state"] = "EXECUTED" if not obligation.get("history") or obligation["history"][-1] != "EXECUTED": obligation.setdefault("history", []).append("EXECUTED") obligation["tool_use_id"] = _sanitize( payload.get("tool_use_id") or payload.get("tool_call_id") or "" ) def _mark_evidence( state: dict[str, Any], payload: dict[str, Any], session_id: str ) -> None: result = _tool_result_value(payload) output = _tool_output(payload) if result is None: return tool_use_id = _sanitize( payload.get("tool_use_id") or payload.get("tool_call_id") or "" ) candidates = [ item for item in state["obligations"] if item.get("state") == "EXECUTED" and _same_session(item, session_id) and ( item.get("tool_use_id") == tool_use_id if tool_use_id else _heuristic_evidence_affinity( item.get("summary", ""), _tool_text(payload) ) ) ] if not candidates: return obligation = candidates[-1] if obligation.get("source", {}).get("type") == "explicit_directive": name = _tool_name(payload) outcome = _tool_outcome(payload, result) observed = obligation.setdefault("observed_evidence", []) for index, required in enumerate(obligation.get("required_evidence", [])): if str(required.get("tool", "")).casefold() != name: continue predicate_results = [ { "predicate": predicate["predicate"], "passed": _evaluate_acceptance( predicate, payload, result, outcome ), } for predicate in required.get("acceptance", []) ] observed.append( { "requirement_index": index, "tool_use_id": tool_use_id, "tool_name": _sanitize(payload.get("tool_name", "")), "outcome": outcome, "evidence_digest": _digest(result), "acceptance_passed": bool(predicate_results) and all(item["passed"] for item in predicate_results), "predicate_results": predicate_results, "observed_at": _utc_now(), } ) obligation["observed_evidence"] = observed[-MAX_OBSERVED_EVIDENCE:] satisfied = { item.get("requirement_index") for item in obligation["observed_evidence"] if item.get("acceptance_passed") is True } required_count = len(obligation.get("required_evidence", [])) if required_count and satisfied.issuperset(range(required_count)): obligation["state"] = "EVIDENCED" if ( not obligation.get("history") or obligation["history"][-1] != "EVIDENCED" ): obligation.setdefault("history", []).append("EVIDENCED") accepted_digests = [ item["evidence_digest"] for item in obligation["observed_evidence"] if item.get("acceptance_passed") is True ] obligation["evidence_digest"] = _digest(accepted_digests) obligation["evidence_summary"] = ( f"Accepted structured evidence from {_sanitize(payload.get('tool_name', 'tool'))}" ) obligation["observed_result"] = "accepted" else: obligation["observed_result"] = "acceptance predicates not satisfied" return if not output: return obligation["state"] = "EVIDENCED" obligation.setdefault("history", []).append("EVIDENCED") obligation["evidence_digest"] = _digest(output) obligation["evidence_summary"] = output obligation["observed_result"] = output def _mark_resolution( state: dict[str, Any], text: str, session_id: str ) -> bool: match = re.search( r"Resolved:\s*(?P.+?);\s*evidence:\s*(?P.+?);\s*" r"conclusion:\s*(?P.+?)(?:\.|$)", text, re.IGNORECASE, ) if not match: return False evidence = _sanitize(match.group("evidence")) for obligation in reversed(state["obligations"]): if ( obligation.get("state") != "EVIDENCED" or not _same_session(obligation, session_id) ): continue if evidence not in { obligation.get("tool_use_id"), obligation.get("evidence_digest"), }: continue obligation["state"] = "RESOLVED" obligation.setdefault("history", []).append("RESOLVED") obligation["resolution"] = { "evidence": evidence, "conclusion": _sanitize(match.group("conclusion")), "resolved_at": _utc_now(), } _resolve_obligation_findings(state, obligation.get("id", "")) return True return True def _resolve_obligation_findings( state: dict[str, Any], obligation_id: str ) -> None: if not obligation_id: return resolved_at = _utc_now() for finding in state["violations"]: if ( finding.get("obligation_id") == obligation_id and finding.get("active", True) ): finding["active"] = False finding["resolved_at"] = resolved_at def _record_violation( state: dict[str, Any], category: str, mode: str, event_digest: str, reason: str, obligation: dict[str, Any], ) -> None: if mode == "off": return code = VIOLATION_CODES[category] obligation_id = _sanitize(obligation.get("id", "")) now = _utc_now() obligation_state = _sanitize(obligation.get("state", "")) for item in state["violations"]: if ( item.get("code") == code and item.get("obligation_id", "") == obligation_id and item.get("active", True) ): item["mode"] = mode item["reason"] = _sanitize(reason) prior_state = _sanitize(item.get("last_obligation_state", "")) if not prior_state: item["last_obligation_state"] = obligation_state elif prior_state != obligation_state: item["occurrences"] = int(item.get("occurrences", 1)) + 1 item["last_obligation_state"] = obligation_state item["last_seen"] = now digests = item.setdefault("source_event_digests", []) if event_digest not in digests: digests.append(event_digest) del digests[:-MAX_OBSERVED_EVIDENCE] return state["violations"].append( { "code": code, "category": category, "mode": mode, "reason": _sanitize(reason), "obligation_id": obligation_id, "obligation": _sanitize( obligation.get("claim") or obligation.get("summary", "") ), "occurrences": 1, "first_seen": now, "last_seen": now, "active": True, "resolved_at": "", "last_obligation_state": obligation_state, "source_event_digests": [event_digest], "source_event_digest": event_digest, "at": now, } ) state["violations"] = state["violations"][-MAX_RECORDS:] def _is_temporary_mitigation(tool_text: str) -> bool: return bool( re.search( r"\b(temporary|temporarily|mitigation|fail-closed)\b", tool_text, re.IGNORECASE, ) and re.search( r"\b(root cause remains unresolved|unresolved)\b", tool_text, re.IGNORECASE, ) and not re.search(r"\bpermanent(?:ly)?\b", tool_text, re.IGNORECASE) ) def _boundary_category(event: str, text: str, tool_text: str) -> str: if ( event in {"PreToolUse", "PostToolUse"} and PERMANENT_MUTATION_RE.search(tool_text) and MUTATION_TARGET_RE.search(tool_text) and not _is_temporary_mitigation(tool_text) ): return "permanent_mutations" if event == "Stop" and (CAUSAL_RE.search(text) or BRANCH_PRUNE_RE.search(text)): return "causal_claims" if event == "Stop" and COMPLETION_RE.search(text): return "completion_claims" if event == "Stop": return "evidence_follow_through" return "" def _guard_boundary( state: dict[str, Any], event: str, text: str, tool_text: str, event_digest: str, global_mode: str, categories: dict[str, str], session_id: str, ) -> tuple[bool, str]: obligation = _latest_open(state, session_id) if not obligation: return False, "" category = "" reason = "" temporary_mitigation = _is_temporary_mitigation(tool_text) if ( event in {"PreToolUse", "PostToolUse"} and PERMANENT_MUTATION_RE.search(tool_text) and MUTATION_TARGET_RE.search(tool_text) and not temporary_mitigation ): category = "permanent_mutations" reason = ( "This permanent mutation already ran without the promised evidence; " "label it as mitigation and complete, defer, or reverse it before continuing." if event == "PostToolUse" else "Complete or explicitly defer the promised evidence check before this permanent mutation." ) elif event == "Stop" and CAUSAL_RE.search(text): category = "causal_claims" reason = "The causal claim is ahead of the promised evidence check." elif event == "Stop" and COMPLETION_RE.search(text): category = "completion_claims" reason = "The completion claim is ahead of the promised evidence check." elif event == "Stop" and BRANCH_PRUNE_RE.search(text): category = "causal_claims" reason = "A plausible branch was pruned before the promised discriminating check." elif event == "Stop": category = "evidence_follow_through" reason = "A promised evidence check remains open at the completion boundary." if not category: return False, "" if obligation.get("state") == "EVIDENCED": reason = ( "Evidence was collected, but the obligation has not been explicitly " "resolved." ) mode = _effective_mode(global_mode, categories, category) if not obligation.get("blocking", False) and mode == "on": mode = "report" _record_violation( state, category, mode, event_digest, reason, obligation ) if mode == "on": state["guard_actions"].append( { "action": "block", "category": category, "source_event_digest": event_digest, "at": _utc_now(), } ) state["guard_actions"] = state["guard_actions"][-MAX_RECORDS:] return True, reason return False, "" def _update_summary(state: dict[str, Any], elapsed_ms: float) -> None: samples = state.setdefault("metrics", {}).setdefault("samples_ms", []) samples.append(round(elapsed_ms, 3)) del samples[:-128] ordered = sorted(samples) p50_index = max(0, round(0.50 * (len(ordered) - 1))) p95_index = max(0, round(0.95 * (len(ordered) - 1))) state["metrics"]["p50_ms"] = ordered[p50_index] state["metrics"]["p95_ms"] = ordered[p95_index] state["summary"] = { "open": sum(item.get("state") in OPEN_STATES for item in state["obligations"]), "deferred": sum(item.get("state") == "DEFERRED" for item in state["obligations"]), "violations": sum( item.get("active", True) for item in state["violations"] ), "blocked": sum( item.get("action") == "block" for item in state.get("guard_actions", []) ), } def _clearance_report_digest(state: dict[str, Any]) -> str: """Digest the bounded diagnostic facts that determine clearance (REQ-173).""" return _digest( { "schema_version": state.get("schema_version"), "mode": state.get("mode"), "session_id": state.get("session_id"), "coverage": state.get("coverage"), "obligations": [ { "id": item.get("id"), "claim": item.get("claim"), "state": item.get("state"), "blocking": item.get("blocking"), "evidence_digest": item.get("evidence_digest"), } for item in state.get("obligations", []) ], "findings": [ { "code": item.get("code"), "obligation_id": item.get("obligation_id"), "active": item.get("active", True), "occurrences": item.get("occurrences", 1), } for item in state.get("violations", []) ], "uncertain_summaries": ( state.get("metrics", {}) .get("classification", {}) .get("uncertain", 0) ), } ) def _update_clearance(state: dict[str, Any]) -> None: """Derive a human-readable, conservative clearance receipt.""" obligations = state.get("obligations", []) blocking_by_id = { item.get("id"): item for item in obligations if item.get("blocking") is True and item.get("id") } open_blocking = [ {"name": item.get("claim") or "Unnamed evidence check", "id": item.get("id")} for item in obligations if item.get("blocking") is True and item.get("state") in OPEN_STATES ] deferred_blocking = [ {"name": item.get("claim") or "Unnamed evidence check", "id": item.get("id")} for item in obligations if item.get("blocking") is True and item.get("state") == "DEFERRED" ] active_blocking_diagnostics = [] for finding in state.get("violations", []): obligation = blocking_by_id.get(finding.get("obligation_id")) if finding.get("active", True) is not True or not obligation: continue active_blocking_diagnostics.append( { "message": finding.get("reason") or "Evidence check remains unresolved.", "code": finding.get("code"), "obligation_name": ( obligation.get("claim") or "Unnamed evidence check" ), "obligation_id": obligation.get("id"), } ) coverage = state.get("coverage") or {} limitations = [ _sanitize(item) for item in coverage.get("missing", []) if isinstance(item, str) and item.strip() ] inferred_open = sum( item.get("source", {}).get("type") != "explicit_directive" and item.get("state") in OPEN_STATES for item in obligations ) if inferred_open: limitations.append( f"{inferred_open} low-confidence inferred observation" f"{'s remain' if inferred_open != 1 else ' remains'} non-blocking." ) uncertain = ( state.get("metrics", {}) .get("classification", {}) .get("uncertain", 0) ) if uncertain: limitations.append( f"{uncertain} visible active summar" f"{'ies' if uncertain != 1 else 'y'} could not be classified confidently." ) blocked = bool( open_blocking or deferred_blocking or active_blocking_diagnostics ) coverage_level = coverage.get("level", "reduced") status = "blocked" if blocked else ( "clear" if ( coverage_level == "historically_complete" and not inferred_open and not uncertain ) else "conditional" ) state["clearance"] = { "status": status, "mode": state.get("mode", "report"), "coverage": { "host": coverage.get("host", "unknown"), "level": coverage_level, }, "limitations": limitations, "open_blocking_obligations": open_blocking, "deferred_blocking_obligations": deferred_blocking, "active_blocking_diagnostics": active_blocking_diagnostics, "deferred_count": len(deferred_blocking), "report_digest": _clearance_report_digest(state), "issued_at": _utc_now(), } def process_event( root: str | Path, event: str, host: str, payload: dict[str, Any] ) -> dict[str, Any]: """Process one visible hook event and return a host-neutral decision.""" started = time.perf_counter() root = Path(root).resolve() payload = payload if isinstance(payload, dict) else {} mode, categories, backfill_mode, backfill_limit = _configured_modes(root) if mode == "off": return {"action": "off", "mode": "off"} blocked = False block_reason = "" malformed = False report_path = root / _host_report_relative_path(host) with _state_lock(root): ledger, state, malformed = _load_host_state( root, mode, host, payload, backfill_mode, backfill_limit, ) previous_session_id = state.get("session_id", "") current_session_id = _session_id(payload) previous_coverage = state.get("coverage") same_session = bool( current_session_id and current_session_id == previous_session_id and isinstance(previous_coverage, dict) and previous_coverage.get("host") == (host or "unknown") ) ingestion = state.get("reasoning_summary_ingestion") if ( not same_session or not isinstance(ingestion, dict) ): ingestion = _empty_ingestion(backfill_mode, backfill_limit) state["processed_summary_digests"] = [] elif ( ingestion.get("mode") != backfill_mode or ingestion.get("limit") != backfill_limit ): ingestion = _reset_ingestion_mode( ingestion, backfill_mode, backfill_limit ) payload, ingestion = _with_visible_reasoning_summaries( host, payload, ingestion, backfill_mode, backfill_limit, ) state["reasoning_summary_ingestion"] = ingestion semantic_payload = dict(payload) summary_items = semantic_payload.pop("_reasoning_summaries", []) semantic_payload.pop("reasoning_summary", None) event_digest = _event_digest(event, semantic_payload) state["mode"] = mode state["session_id"] = current_session_id or previous_session_id state["surface"] = _session_surface( host, payload, ingestion, state.get("surface", "unknown") if same_session else "unknown", ) state["model"], state["effort"] = _session_model_effort( payload, ingestion, state.get("model", "unknown") if same_session else "unknown", state.get("effort", "unknown") if same_session else "unknown", ) state["turn_id"] = _sanitize( payload.get("turn_id") or payload.get("turnId") or state.get("turn_id", "") ) state["effective_modes"] = { category: _effective_mode(mode, categories, category) for category in CATEGORIES } state["coverage"] = _coverage( host, payload, ingestion, previous_coverage if same_session else None, ) session_id = _session_id(payload) turn_id = _sanitize( payload.get("turn_id") or payload.get("turnId") or "" ) for summary_item in summary_items: if not isinstance(summary_item, dict): continue summary_digest = summary_item.get("digest") summary_text = summary_item.get("text") if ( not isinstance(summary_digest, str) or not isinstance(summary_text, str) or summary_digest in state["processed_summary_digests"] ): continue summary_payload = {"reasoning_summary": summary_text} classification_outcome = "ignored" terminal_directive = _record_deferral( state, summary_text, session_id ) terminal_directive = ( _mark_resolution(state, summary_text, session_id) or terminal_directive ) if not terminal_directive: explicit_directive = _record_explicit_obligation( state, summary_payload, summary_digest, session_id, turn_id, ) if not explicit_directive: classification_outcome = _record_obligation( state, summary_text, summary_digest, session_id, turn_id, ) _record_classification_metric(state, classification_outcome) state["processed_summary_digests"].append(summary_digest) del state["processed_summary_digests"][:-MAX_EVENTS] ingestion["summaries_processed"] = int( ingestion.get("summaries_processed", 0) ) + 1 if event_digest not in state["processed_event_digests"]: text = _visible_text(semantic_payload) tool_text = _tool_text(semantic_payload) terminal_directive = _record_deferral(state, text, session_id) terminal_directive = ( _mark_resolution(state, text, session_id) or terminal_directive ) if not terminal_directive: explicit_directive = _record_explicit_obligation( state, semantic_payload, event_digest, session_id, turn_id, ) if not explicit_directive: _record_obligation( state, text, event_digest, session_id, turn_id ) if event == "PostToolUse": _mark_evidence(state, semantic_payload, session_id) blocked, block_reason = _guard_boundary( state, event, text, tool_text, event_digest, mode, categories, session_id, ) if event == "PreToolUse" and not blocked: _mark_execution(state, semantic_payload, session_id) state["processed_event_digests"].append(event_digest) del state["processed_event_digests"][:-MAX_EVENTS] if malformed: malformed_category = _boundary_category( event, _visible_text(semantic_payload), _tool_text(semantic_payload), ) if malformed_category and _effective_mode( mode, categories, malformed_category ) == "on": blocked = True block_reason = ( "Reasoning Guard state was malformed and has been safely recovered; " "retry after reviewing the recovered report." ) state["updated_at"] = _utc_now() _update_summary(state, (time.perf_counter() - started) * 1000) _update_clearance(state) report_path = _write_host_state(root, host, ledger, state) if blocked: return { "decision": "block", "reason": block_reason, "mode": mode, "report_path": str(report_path), } return { "action": "allow", "mode": mode, "report_path": str(report_path), } def main() -> int: try: payload = json.load(sys.stdin) except (json.JSONDecodeError, OSError): payload = {} root = Path( os.environ.get("PRD_PROJECT_ROOT") or payload.get("cwd") or payload.get("directory") or os.getcwd() ) event = os.environ.get("PRD_HOOK_EVENT") or payload.get("hook_event_name") or "" host = os.environ.get("PRD_HOOK_HOST", "unknown") result = process_event(root, event, host, payload) print(json.dumps(result, sort_keys=True)) return 2 if result.get("decision") == "block" and event == "PreToolUse" else 0 if __name__ == "__main__": raise SystemExit(main())