From fc4087ca4bc5d961e59c2ddf96c13c57943ee9e2 Mon Sep 17 00:00:00 2001 From: twentyOne2x Date: Wed, 15 Jul 2026 23:03:00 +0200 Subject: [PATCH 1/3] Support private local source reconstruction --- scripts/compile_kb_source_packet.py | 22 +- .../prepare_kb_context_from_canonical_dump.py | 320 ++++++++++++++++++ scripts/prepare_kb_source_manifest.py | 144 +++++++- tests/test_compile_kb_source_packet.py | 11 +- ..._prepare_kb_context_from_canonical_dump.py | 128 +++++++ tests/test_prepare_kb_source_manifest.py | 135 ++++++++ 6 files changed, 741 insertions(+), 19 deletions(-) create mode 100644 scripts/prepare_kb_context_from_canonical_dump.py create mode 100644 tests/test_prepare_kb_context_from_canonical_dump.py diff --git a/scripts/compile_kb_source_packet.py b/scripts/compile_kb_source_packet.py index e7b3efe..88d09b4 100644 --- a/scripts/compile_kb_source_packet.py +++ b/scripts/compile_kb_source_packet.py @@ -13,8 +13,10 @@ from __future__ import annotations import argparse import hashlib import json +import os import re import sys +import tempfile import uuid from pathlib import Path from typing import Any @@ -574,6 +576,23 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace: return parser.parse_args(argv) +def _write_private_output(path: Path, rendered: str) -> None: + path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) + fd, raw_temp = tempfile.mkstemp(prefix=f".{path.name}.", suffix=".tmp", dir=path.parent) + temp = Path(raw_temp) + try: + os.fchmod(fd, 0o600) + with os.fdopen(fd, "w", encoding="utf-8") as handle: + handle.write(rendered) + handle.flush() + os.fsync(handle.fileno()) + os.replace(temp, path) + os.chmod(path, 0o600) + finally: + if temp.exists(): + temp.unlink() + + def main(argv: list[str] | None = None) -> int: args = parse_args(argv) try: @@ -600,8 +619,7 @@ def main(argv: list[str] | None = None) -> int: rendered = json.dumps(bundle, indent=2, sort_keys=True, ensure_ascii=True) + "\n" if args.output: try: - args.output.parent.mkdir(parents=True, exist_ok=True) - args.output.write_text(rendered, encoding="utf-8") + _write_private_output(args.output, rendered) except OSError as exc: print( json.dumps( diff --git a/scripts/prepare_kb_context_from_canonical_dump.py b/scripts/prepare_kb_context_from_canonical_dump.py new file mode 100644 index 0000000..d5e26ce --- /dev/null +++ b/scripts/prepare_kb_context_from_canonical_dump.py @@ -0,0 +1,320 @@ +#!/usr/bin/env python3 +"""Build source-specific canonical claim context from a Postgres dump without a database write.""" + +from __future__ import annotations + +import argparse +import hashlib +import importlib.util +import json +import os +import re +import shutil +import subprocess +import tempfile +from collections import Counter +from decimal import Decimal, InvalidOperation +from pathlib import Path +from types import ModuleType +from typing import Any + +SCHEMA = "livingip.kbSourceDumpContextReceipt.v1" +DEFAULT_DOCKER_IMAGE = "postgres:16-alpine" +DEFAULT_LIMIT = 12 +MAX_PREVIEW_BYTES = 16_000 +TABLES = ("claims", "claim_evidence", "claim_edges") + + +class ContextError(RuntimeError): + """Raised when a dump cannot safely produce canonical retrieval context.""" + + +def sha256_bytes(value: bytes) -> str: + return hashlib.sha256(value).hexdigest() + + +def _read_file(path: Path, label: str) -> bytes: + try: + if not path.is_file(): + raise ContextError(f"{label} is not a regular file: {path}") + return path.read_bytes() + except OSError as exc: + raise ContextError(f"could not read {label}: {exc}") from exc + + +def _load_kb_tool(repo_root: Path) -> ModuleType: + path = repo_root / "hermes-agent" / "leoclean-bin" / "kb_tool.py" + spec = importlib.util.spec_from_file_location("teleo_kb_tool_context", path) + if spec is None or spec.loader is None: + raise ContextError(f"could not load retrieval semantics from {path}") + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def _decode_copy_field(value: str) -> str | None: + if value == r"\N": + return None + result: list[str] = [] + index = 0 + escapes = {"b": "\b", "f": "\f", "n": "\n", "r": "\r", "t": "\t", "v": "\v", "\\": "\\"} + while index < len(value): + char = value[index] + if char != "\\" or index + 1 >= len(value): + result.append(char) + index += 1 + continue + index += 1 + escaped = value[index] + if escaped in escapes: + result.append(escapes[escaped]) + index += 1 + continue + if escaped == "x": + match = re.match(r"[0-9A-Fa-f]{1,2}", value[index + 1 :]) + if match: + result.append(chr(int(match.group(0), 16))) + index += 1 + len(match.group(0)) + continue + if escaped in "01234567": + match = re.match(r"[0-7]{1,3}", value[index:]) + if match: + result.append(chr(int(match.group(0), 8))) + index += len(match.group(0)) + continue + result.append(escaped) + index += 1 + return "".join(result) + + +def parse_copy_table(sql: str, table: str) -> list[dict[str, str | None]]: + prefix = f"COPY public.{table} (" + lines = iter(sql.splitlines()) + for line in lines: + if line.startswith(prefix) and line.endswith(") FROM stdin;"): + columns = [item.strip() for item in line[len(prefix) : -len(") FROM stdin;")].split(",")] + break + else: + raise ContextError(f"pg_restore output has no COPY block for public.{table}") + rows: list[dict[str, str | None]] = [] + for line in lines: + if line == r"\.": + return rows + fields = line.split("\t") + if len(fields) != len(columns): + raise ContextError(f"public.{table} COPY row has {len(fields)} fields; expected {len(columns)}") + rows.append(dict(zip(columns, (_decode_copy_field(field) for field in fields), strict=True))) + raise ContextError(f"public.{table} COPY block is unterminated") + + +def restore_table(dump_path: Path, table: str, docker_image: str) -> str: + if table not in TABLES: + raise ContextError(f"unsupported canonical table: {table}") + if shutil.which("docker") is None: + raise ContextError("docker is not installed or not available on PATH") + command = [ + "docker", + "run", + "--rm", + "--network=none", + "--mount", + f"type=bind,src={dump_path.resolve()},dst=/canonical.dump,readonly", + docker_image, + "pg_restore", + "--data-only", + "-t", + table, + "-f", + "-", + "/canonical.dump", + ] + try: + completed = subprocess.run(command, capture_output=True, text=True, timeout=180, check=False) + except (OSError, subprocess.TimeoutExpired) as exc: + raise ContextError(f"pg_restore failed ({type(exc).__name__}); diagnostic detail redacted") from exc + if completed.returncode != 0: + raise ContextError(f"pg_restore returned exit {completed.returncode}; diagnostic detail redacted") + return completed.stdout + + +def _decimal(value: str | None) -> Decimal: + if value is None: + return Decimal(0) + try: + return Decimal(value) + except InvalidOperation as exc: + raise ContextError("canonical claim confidence is not numeric") from exc + + +def build_context( + *, dump_path: Path, artifact_path: Path, title: str, limit: int, docker_image: str, repo_root: Path +) -> tuple[dict[str, Any], dict[str, Any]]: + if limit < 1 or limit > 20: + raise ContextError("limit must be between 1 and 20") + dump_bytes = _read_file(dump_path, "canonical dump") + artifact_bytes = _read_file(artifact_path, "source artifact") + try: + preview = artifact_bytes[:MAX_PREVIEW_BYTES].decode("utf-8", errors="strict") + except UnicodeDecodeError as exc: + raise ContextError(f"source artifact preview must be strict UTF-8: {exc}") from exc + kb_tool = _load_kb_tool(repo_root) + query = " ".join(kb_tool.query_terms(f"{title} {preview}")) + terms = kb_tool.query_terms(query) + min_score = 2 if len(terms) >= 3 else 1 + + tables = {table: parse_copy_table(restore_table(dump_path, table, docker_image), table) for table in TABLES} + evidence_counts = Counter(row["claim_id"] for row in tables["claim_evidence"] if row["claim_id"]) + edge_counts: Counter[str] = Counter() + for row in tables["claim_edges"]: + if row["from_claim"]: + edge_counts[row["from_claim"]] += 1 + if row["to_claim"]: + edge_counts[row["to_claim"]] += 1 + + ranked: list[dict[str, Any]] = [] + for row in tables["claims"]: + if row["status"] != "open" or not row["id"] or not row["text"]: + continue + text = row["text"] + searchable = f"{text}\n{row['tags'] or ''}".lower() + score = sum(term in searchable for term in terms) + if score < min_score: + continue + ranked.append( + { + "id": row["id"], + "text": text, + "score": score, + "evidence_count": evidence_counts[row["id"]], + "edge_count": edge_counts[row["id"]], + "_confidence": _decimal(row["confidence"]), + } + ) + ranked.sort( + key=lambda row: ( + -row["score"], + -row["evidence_count"], + -row["edge_count"], + -row["_confidence"], + len(row["text"]), + row["text"], + row["id"], + ) + ) + candidates = [{key: value for key, value in row.items() if key != "_confidence"} for row in ranked[:limit]] + context = {"database_search_query": query, "candidate_claims": candidates} + receipt = { + "schema": SCHEMA, + "status": "prepared", + "database_write_performed": False, + "model_call_performed": False, + "canonical_dump": { + "path": str(dump_path.resolve()), + "sha256": sha256_bytes(dump_bytes), + }, + "source": { + "path": str(artifact_path.resolve()), + "sha256": sha256_bytes(artifact_bytes), + }, + "retrieval": { + "query_sha256": sha256_bytes(query.encode("utf-8")), + "term_count": len(terms), + "minimum_score": min_score, + "canonical_open_claim_count": sum(row["status"] == "open" for row in tables["claims"]), + "candidate_count": len(candidates), + "candidate_ids": [row["id"] for row in candidates], + }, + "runtime": { + "docker_image": docker_image, + "network_disabled": True, + "pg_restore_output_retained": False, + }, + } + return context, receipt + + +def _private_json(path: Path, value: Any) -> None: + if path.exists(): + raise ContextError(f"refusing to overwrite existing output: {path}") + path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) + os.chmod(path.parent, 0o700) + fd, raw_temp = tempfile.mkstemp(prefix=f".{path.name}.", suffix=".tmp", dir=path.parent) + temp = Path(raw_temp) + try: + os.fchmod(fd, 0o600) + with os.fdopen(fd, "w", encoding="utf-8") as 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) + os.chmod(path, 0o600) + finally: + if temp.exists(): + temp.unlink() + + +def parse_args(argv: list[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--dump", required=True, type=Path) + parser.add_argument("--artifact", required=True, type=Path) + parser.add_argument("--title", required=True) + parser.add_argument("--context-output", required=True, type=Path) + parser.add_argument("--receipt-output", required=True, type=Path) + parser.add_argument("--limit", type=int, default=DEFAULT_LIMIT) + parser.add_argument("--docker-image", default=DEFAULT_DOCKER_IMAGE) + return parser.parse_args(argv) + + +def main(argv: list[str] | None = None) -> int: + args = parse_args(argv) + repo_root = Path(__file__).resolve().parents[1] + try: + if args.context_output.resolve() == args.receipt_output.resolve(): + raise ContextError("context and receipt outputs must be different files") + context, receipt = build_context( + dump_path=args.dump, + artifact_path=args.artifact, + title=args.title, + limit=args.limit, + docker_image=args.docker_image, + repo_root=repo_root, + ) + _private_json(args.context_output, context) + receipt["outputs"] = { + "context": str(args.context_output.resolve()), + "receipt": str(args.receipt_output.resolve()), + } + _private_json(args.receipt_output, receipt) + except (ContextError, OSError, ValueError) as exc: + print( + json.dumps( + { + "schema": SCHEMA, + "status": "rejected", + "database_write_performed": False, + "error": str(exc), + }, + sort_keys=True, + ) + ) + return 2 + print( + json.dumps( + { + "schema": SCHEMA, + "status": "prepared", + "database_write_performed": False, + "candidate_count": receipt["retrieval"]["candidate_count"], + "context_sha256": sha256_bytes(json.dumps(context, sort_keys=True).encode("utf-8")), + "outputs": receipt["outputs"], + }, + indent=2, + sort_keys=True, + ) + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/prepare_kb_source_manifest.py b/scripts/prepare_kb_source_manifest.py index 6823e1d..e9b8ade 100644 --- a/scripts/prepare_kb_source_manifest.py +++ b/scripts/prepare_kb_source_manifest.py @@ -15,6 +15,8 @@ import hashlib import json import os import re +import shutil +import subprocess import sys import tempfile import urllib.error @@ -34,6 +36,8 @@ DEFAULT_MODEL = "anthropic/claude-sonnet-4.5" EXTRACTOR_VERSION = "2" DEFAULT_OPENROUTER_URL = "https://openrouter.ai/api/v1/chat/completions" DEFAULT_KEY_FILE = Path("/opt/teleo-eval/secrets/openrouter-key") +DEFAULT_CLAUDE_MAX_BUDGET_USD = 0.10 +MODEL_PROVIDERS = {"openrouter", "claude-cli"} TEXT_SUFFIXES = {".csv", ".htm", ".html", ".json", ".jsonl", ".md", ".rst", ".txt", ".xml"} MAX_SOURCE_CHARS = 120_000 MAX_CANDIDATES = 20 @@ -248,10 +252,10 @@ def call_openrouter(prompt: str, *, model: str, key_file: Path, url: str, timeou with urllib.request.urlopen(request, timeout=timeout_seconds) as response: body = response.read() except urllib.error.HTTPError as exc: - detail = exc.read(600).decode("utf-8", errors="replace") - raise PreparationError(f"OpenRouter returned HTTP {exc.code}: {detail}") from exc + exc.read(600) + raise PreparationError(f"OpenRouter returned HTTP {exc.code}; provider detail redacted") from exc except (urllib.error.URLError, TimeoutError, OSError) as exc: - raise PreparationError(f"OpenRouter request failed: {exc}") from exc + raise PreparationError(f"OpenRouter request failed ({type(exc).__name__}); provider detail redacted") from exc try: envelope = json.loads(body.decode("utf-8", errors="strict")) content = envelope["choices"][0]["message"]["content"] @@ -263,6 +267,100 @@ def call_openrouter(prompt: str, *, model: str, key_file: Path, url: str, timeou return content, usage if isinstance(usage, dict) else {} +def call_claude_cli( + prompt: str, + *, + model: str, + timeout_seconds: int, + max_budget_usd: float, +) -> tuple[str, dict[str, Any]]: + """Run a non-persistent, tool-free Claude extraction with a cost cap. + + The private prompt is supplied on stdin rather than in the process argv. + Provider diagnostics are never copied into exceptions because they may + contain source fragments or credential-adjacent values. + """ + executable = shutil.which("claude") + if executable is None: + raise PreparationError("claude CLI is not installed or not available on PATH") + if isinstance(max_budget_usd, bool) or not isinstance(max_budget_usd, (int, float)) or max_budget_usd <= 0: + raise PreparationError("Claude CLI max budget must be a positive number") + command = [ + executable, + "-p", + "--no-session-persistence", + "--tools", + "", + "--model", + model, + "--output-format", + "json", + "--max-budget-usd", + str(max_budget_usd), + ] + try: + completed = subprocess.run( + command, + input=prompt, + capture_output=True, + text=True, + timeout=timeout_seconds, + check=False, + ) + except (OSError, subprocess.TimeoutExpired) as exc: + raise PreparationError(f"Claude CLI request failed ({type(exc).__name__}); provider detail redacted") from exc + if completed.returncode != 0: + raise PreparationError(f"Claude CLI returned exit {completed.returncode}; provider detail redacted") + try: + envelope = json.loads(completed.stdout) + except json.JSONDecodeError as exc: + raise PreparationError("Claude CLI returned an invalid JSON envelope; provider detail redacted") from exc + if not isinstance(envelope, dict) or envelope.get("is_error") is True or envelope.get("subtype") != "success": + raise PreparationError("Claude CLI returned an unsuccessful envelope; provider detail redacted") + content = envelope.get("result") + if not isinstance(content, str) or not content.strip(): + raise PreparationError("Claude CLI returned no extraction content") + usage = { + "total_cost_usd": envelope.get("total_cost_usd"), + "usage": envelope.get("usage") if isinstance(envelope.get("usage"), dict) else {}, + "model_usage": envelope.get("modelUsage") if isinstance(envelope.get("modelUsage"), dict) else {}, + "max_budget_usd": max_budget_usd, + "session_persistence": False, + "tools_enabled": False, + } + return content, usage + + +def call_model( + prompt: str, args: argparse.Namespace, *, max_budget_usd: float | None = None +) -> tuple[str, dict[str, Any]]: + provider = getattr(args, "provider", "openrouter") + if provider == "openrouter": + return call_openrouter( + prompt, + model=args.model, + key_file=args.key_file, + url=args.openrouter_url, + timeout_seconds=args.timeout_seconds, + ) + if provider == "claude-cli": + return call_claude_cli( + prompt, + model=args.model, + timeout_seconds=args.timeout_seconds, + max_budget_usd=( + max_budget_usd + if max_budget_usd is not None + else getattr(args, "max_budget_usd", DEFAULT_CLAUDE_MAX_BUDGET_USD) + ), + ) + raise PreparationError(f"model provider must be one of {sorted(MODEL_PROVIDERS)}") + + +def extractor_name(args: argparse.Namespace) -> str: + return f"{getattr(args, 'provider', 'openrouter')}:{args.model}" + + def parse_model_output(content: str) -> dict[str, Any]: stripped = content.strip() if stripped.startswith("```"): @@ -447,7 +545,7 @@ def build_manifest( "schema": compiler.MANIFEST_SCHEMA_V2, "artifact_sha256": artifact_sha256, "extracted_text_sha256": text_sha256, - "extractor": {"name": f"openrouter:{args.model}", "version": EXTRACTOR_VERSION}, + "extractor": {"name": extractor_name(args), "version": EXTRACTOR_VERSION}, "source": { "identity": args.identity, "source_key": args.source_key, @@ -526,14 +624,23 @@ def prepare(args: argparse.Namespace) -> dict[str, Any]: extraction: dict[str, Any] | None = None bundle: dict[str, Any] | None = None manifest: dict[str, Any] | None = None + reported_total_cost_usd = 0.0 for attempt in range(1, MAX_ATTEMPTS + 1): - raw, usage = call_openrouter( + provider = getattr(args, "provider", "openrouter") + max_total_budget_usd = getattr(args, "max_budget_usd", DEFAULT_CLAUDE_MAX_BUDGET_USD) + remaining_budget_usd = max_total_budget_usd - reported_total_cost_usd if provider == "claude-cli" else None + raw, usage = call_model( build_prompt(fragments, context, repair_error), - model=args.model, - key_file=args.key_file, - url=args.openrouter_url, - timeout_seconds=args.timeout_seconds, + args, + max_budget_usd=remaining_budget_usd, ) + if provider == "claude-cli": + reported_cost = usage.get("total_cost_usd") + if isinstance(reported_cost, bool) or not isinstance(reported_cost, (int, float)) or reported_cost < 0: + raise PreparationError("Claude CLI did not report a valid non-negative model cost") + reported_total_cost_usd += float(reported_cost) + if reported_total_cost_usd > max_total_budget_usd + 1e-9: + raise PreparationError("Claude CLI reported model cost above the configured total budget") attempts.append({"attempt": attempt, "response_sha256": sha256_bytes(raw.encode("utf-8")), "usage": usage}) try: selection = parse_model_output(raw) @@ -616,9 +723,17 @@ def prepare(args: argparse.Namespace) -> dict[str, Any]: "duplicate_judgments": extraction["duplicates"], }, "extraction": { - "provider": "openrouter", + "provider": getattr(args, "provider", "openrouter"), "model": args.model, "attempts": attempts, + "cost_control": { + "max_total_budget_usd": ( + getattr(args, "max_budget_usd", DEFAULT_CLAUDE_MAX_BUDGET_USD) + if getattr(args, "provider", "openrouter") == "claude-cli" + else None + ), + "reported_total_cost_usd": reported_total_cost_usd, + }, "claim_count": len(extraction["claims"]), "evidence_count": sum(len(claim["evidence"]) for claim in extraction["claims"]), "duplicate_judgment_count": len(extraction["duplicates"]), @@ -631,7 +746,7 @@ def prepare(args: argparse.Namespace) -> dict[str, Any]: "extracted_text_sha256": text_sha256, "kb_context_sha256": sha256_bytes(_json_bytes(context)), "fragment_index_sha256": sha256_bytes(_json_bytes(fragment_locations)), - "extractor": {"name": f"openrouter:{args.model}", "version": EXTRACTOR_VERSION}, + "extractor": {"name": extractor_name(args), "version": EXTRACTOR_VERSION}, "exact_selected_locations_validated": bundle is not None, }, "outputs": { @@ -668,6 +783,13 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser.add_argument("--locator", required=True) parser.add_argument("--output-dir", required=True, type=Path) parser.add_argument("--model", default=DEFAULT_MODEL) + parser.add_argument("--provider", choices=sorted(MODEL_PROVIDERS), default="openrouter") + parser.add_argument( + "--max-budget-usd", + type=float, + default=DEFAULT_CLAUDE_MAX_BUDGET_USD, + help="hard total Claude CLI budget cap across validation retries; ignored by the OpenRouter provider", + ) parser.add_argument("--openrouter-url", default=DEFAULT_OPENROUTER_URL) parser.add_argument("--timeout-seconds", type=int, default=180) parser.set_defaults(key_file=DEFAULT_KEY_FILE) diff --git a/tests/test_compile_kb_source_packet.py b/tests/test_compile_kb_source_packet.py index 49adf02..a220b3d 100644 --- a/tests/test_compile_kb_source_packet.py +++ b/tests/test_compile_kb_source_packet.py @@ -3,6 +3,7 @@ from __future__ import annotations import copy import hashlib import json +import stat import sys from pathlib import Path @@ -228,6 +229,7 @@ def test_cli_emits_structured_json_to_stdout_and_optional_output(tmp_path: Path, assert status == 0 assert output.read_text(encoding="utf-8") == stdout assert json.loads(stdout)["status"] == "pending_review" + assert stat.S_IMODE(output.stat().st_mode) == 0o600 @pytest.mark.parametrize( @@ -363,14 +365,11 @@ def test_cli_reports_output_write_failure_as_structured_rejection( ) -> None: artifact, text, manifest_path = _write_inputs(tmp_path) output = tmp_path / "packet.json" - original_write_text = Path.write_text - def guarded_write_text(path: Path, data: str, *args, **kwargs): - if path == output: - raise PermissionError("fixture output denied") - return original_write_text(path, data, *args, **kwargs) + def fail_write(_path: Path, _rendered: str) -> None: + raise PermissionError("fixture output denied") - monkeypatch.setattr(Path, "write_text", guarded_write_text) + monkeypatch.setattr(compiler, "_write_private_output", fail_write) status = compiler.main( [ "--artifact", diff --git a/tests/test_prepare_kb_context_from_canonical_dump.py b/tests/test_prepare_kb_context_from_canonical_dump.py new file mode 100644 index 0000000..5272afc --- /dev/null +++ b/tests/test_prepare_kb_context_from_canonical_dump.py @@ -0,0 +1,128 @@ +from __future__ import annotations + +import importlib.util +import json +import stat +import sys +from pathlib import Path + +import pytest + +REPO_ROOT = Path(__file__).resolve().parents[1] +SCRIPT_PATH = REPO_ROOT / "scripts" / "prepare_kb_context_from_canonical_dump.py" +SPEC = importlib.util.spec_from_file_location("prepare_kb_context_from_canonical_dump", SCRIPT_PATH) +assert SPEC is not None and SPEC.loader is not None +context_builder = importlib.util.module_from_spec(SPEC) +sys.modules[SPEC.name] = context_builder +SPEC.loader.exec_module(context_builder) + +CLAIM_A = "11111111-1111-4111-8111-111111111111" +CLAIM_B = "22222222-2222-4222-8222-222222222222" +SOURCE_ID = "33333333-3333-4333-8333-333333333333" + + +def _copy(table: str, columns: list[str], rows: list[list[str]]) -> str: + rendered = [f"COPY public.{table} ({', '.join(columns)}) FROM stdin;"] + rendered.extend("\t".join(row) for row in rows) + rendered.append(r"\.") + return "\n".join(rendered) + "\n" + + +def test_build_context_matches_readonly_claim_ranking_and_counts( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + dump = tmp_path / "canonical.dump" + dump.write_bytes(b"fixture dump") + artifact = tmp_path / "segment.txt" + artifact.write_text("Exact source quotes keep staged proposals reviewable.\n", encoding="utf-8") + sql = { + "claims": _copy( + "claims", + [ + "id", + "type", + "text", + "status", + "confidence", + "tags", + "created_by", + "superseded_by", + "created_at", + "updated_at", + ], + [ + [ + CLAIM_A, + "structural", + "Exact source quotes keep proposals reviewable.", + "open", + "0.8", + "{provenance}", + r"\N", + r"\N", + "2026-01-01", + "2026-01-01", + ], + [ + CLAIM_B, + "structural", + "Staged proposals need guarded apply.", + "open", + "0.9", + "{review}", + r"\N", + r"\N", + "2026-01-01", + "2026-01-01", + ], + ], + ), + "claim_evidence": _copy( + "claim_evidence", + ["claim_id", "source_id", "role", "weight", "created_by", "created_at"], + [[CLAIM_A, SOURCE_ID, "grounds", "1", r"\N", "2026-01-01"]], + ), + "claim_edges": _copy( + "claim_edges", + ["id", "from_claim", "to_claim", "edge_type", "weight", "created_by", "created_at"], + [["44444444-4444-4444-8444-444444444444", CLAIM_A, CLAIM_B, "supports", "1", r"\N", "2026-01-01"]], + ), + } + monkeypatch.setattr(context_builder, "restore_table", lambda _dump, table, _image: sql[table]) + + context, receipt = context_builder.build_context( + dump_path=dump, + artifact_path=artifact, + title="Telegram segment", + limit=12, + docker_image="fixture-postgres", + repo_root=REPO_ROOT, + ) + + assert context["candidate_claims"][0]["id"] == CLAIM_A + assert context["candidate_claims"][0]["evidence_count"] == 1 + assert context["candidate_claims"][0]["edge_count"] == 1 + assert receipt["database_write_performed"] is False + assert receipt["model_call_performed"] is False + assert receipt["runtime"]["network_disabled"] is True + assert receipt["runtime"]["pg_restore_output_retained"] is False + + +def test_copy_parser_preserves_escaped_source_text() -> None: + sql = _copy("claims", ["id", "text"], [[CLAIM_A, r"line one\nline two\tvalue\\tail"]]) + + rows = context_builder.parse_copy_table(sql, "claims") + + assert rows == [{"id": CLAIM_A, "text": "line one\nline two\tvalue\\tail"}] + + +def test_private_json_is_owner_only_and_refuses_overwrite(tmp_path: Path) -> None: + output = tmp_path / "private" / "context.json" + + context_builder._private_json(output, {"safe": True}) + + assert json.loads(output.read_text(encoding="utf-8")) == {"safe": True} + assert stat.S_IMODE(output.parent.stat().st_mode) == 0o700 + assert stat.S_IMODE(output.stat().st_mode) == 0o600 + with pytest.raises(context_builder.ContextError, match="refusing to overwrite"): + context_builder._private_json(output, {"safe": False}) diff --git a/tests/test_prepare_kb_source_manifest.py b/tests/test_prepare_kb_source_manifest.py index d85a0c2..f24335a 100644 --- a/tests/test_prepare_kb_source_manifest.py +++ b/tests/test_prepare_kb_source_manifest.py @@ -1,8 +1,10 @@ from __future__ import annotations +import io import json import stat import sys +import urllib.error from pathlib import Path from types import SimpleNamespace @@ -35,9 +37,11 @@ def _args(tmp_path: Path, artifact: Path, context: Path) -> SimpleNamespace: locator="artifact://source-preparation/test-v1", output_dir=tmp_path / "private-output", key_file=tmp_path / "openrouter-key", + provider="openrouter", model="fixture/model", openrouter_url="https://openrouter.invalid/v1/chat/completions", timeout_seconds=30, + max_budget_usd=0.10, ) @@ -274,3 +278,134 @@ def test_openrouter_key_cannot_be_redirected_to_an_unapproved_endpoint(tmp_path: url="https://example.invalid/collect", timeout_seconds=1, ) + + +def test_claude_cli_uses_stdin_without_tools_or_session_persistence_and_retains_cost( + monkeypatch: pytest.MonkeyPatch, +) -> None: + prompt = "private source prompt" + observed: dict[str, object] = {} + + def fake_run(command, **kwargs): + observed["command"] = command + observed.update(kwargs) + return SimpleNamespace( + returncode=0, + stderr="", + stdout=json.dumps( + { + "subtype": "success", + "is_error": False, + "result": _model_output(), + "total_cost_usd": 0.0123, + "usage": {"input_tokens": 321, "output_tokens": 123}, + "modelUsage": {"claude-sonnet-test": {"costUSD": 0.0123}}, + } + ), + ) + + monkeypatch.setattr(preparer.shutil, "which", lambda _name: "/usr/local/bin/claude") + monkeypatch.setattr(preparer.subprocess, "run", fake_run) + + content, usage = preparer.call_claude_cli( + prompt, + model="sonnet", + timeout_seconds=30, + max_budget_usd=0.10, + ) + + command = observed["command"] + assert prompt not in command + assert observed["input"] == prompt + assert "--no-session-persistence" in command + assert command[command.index("--tools") + 1] == "" + assert command[command.index("--max-budget-usd") + 1] == "0.1" + assert json.loads(content)["schema"] == preparer.MODEL_OUTPUT_SCHEMA + assert usage["total_cost_usd"] == 0.0123 + assert usage["session_persistence"] is False + assert usage["tools_enabled"] is False + + +def test_prepare_routes_claude_cli_and_records_provider_cost(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + artifact, context = _inputs(tmp_path) + args = _args(tmp_path, artifact, context) + args.provider = "claude-cli" + args.model = "sonnet" + monkeypatch.setattr( + preparer, + "call_openrouter", + lambda *_args, **_kwargs: pytest.fail("OpenRouter must not be called for the Claude CLI provider"), + ) + monkeypatch.setattr( + preparer, + "call_claude_cli", + lambda *_args, **_kwargs: (_model_output(), {"total_cost_usd": 0.0123}), + ) + + receipt = preparer.prepare(args) + + assert receipt["extraction"]["provider"] == "claude-cli" + assert receipt["extraction"]["attempts"][0]["usage"]["total_cost_usd"] == 0.0123 + assert receipt["extraction"]["cost_control"] == { + "max_total_budget_usd": 0.10, + "reported_total_cost_usd": 0.0123, + } + assert receipt["replay"]["extractor"]["name"] == "claude-cli:sonnet" + manifest = json.loads(Path(receipt["outputs"]["manifest"]).read_text(encoding="utf-8")) + assert manifest["extractor"]["name"] == "claude-cli:sonnet" + + +def test_claude_retry_budget_is_reduced_by_reported_first_attempt_cost( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + artifact, context = _inputs(tmp_path) + args = _args(tmp_path, artifact, context) + args.provider = "claude-cli" + args.model = "sonnet" + outputs = iter(((_model_output(bad_reference=True), 0.04), (_model_output(), 0.03))) + budgets: list[float] = [] + + def fake_claude(_prompt: str, *, model: str, timeout_seconds: int, max_budget_usd: float): + assert model == "sonnet" + assert timeout_seconds == 30 + budgets.append(max_budget_usd) + output, cost = next(outputs) + return output, {"total_cost_usd": cost} + + monkeypatch.setattr(preparer, "call_claude_cli", fake_claude) + + receipt = preparer.prepare(args) + + assert budgets == pytest.approx([0.10, 0.06]) + assert receipt["extraction"]["cost_control"]["reported_total_cost_usd"] == pytest.approx(0.07) + + +def test_provider_errors_redact_secret_and_private_detail(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + key_file = tmp_path / "key" + key_file.write_text("fixture-secret", encoding="utf-8") + error = urllib.error.HTTPError( + preparer.DEFAULT_OPENROUTER_URL, + 401, + "unauthorized", + {}, + io.BytesIO(b"fixture-secret and private source quote"), + ) + + def fail_urlopen(*_args, **_kwargs): + raise error + + monkeypatch.setattr(preparer.urllib.request, "urlopen", fail_urlopen) + + with pytest.raises(preparer.PreparationError) as caught: + preparer.call_openrouter( + "private source quote", + model="fixture/model", + key_file=key_file, + url=preparer.DEFAULT_OPENROUTER_URL, + timeout_seconds=1, + ) + + rendered = str(caught.value) + assert "fixture-secret" not in rendered + assert "private source quote" not in rendered + assert "provider detail redacted" in rendered From 9a9594e41b029bbdf6d64779d9388c1cd2e0b8af Mon Sep 17 00:00:00 2001 From: twentyOne2x Date: Thu, 16 Jul 2026 12:24:20 +0200 Subject: [PATCH 2/3] Harden private source reconstruction output --- hermes-agent/leoclean-bin/kb_tool.py | 104 ++++++++++----- scripts/compile_kb_source_packet.py | 62 ++++----- .../prepare_kb_context_from_canonical_dump.py | 126 ++++++++++++++---- scripts/prepare_kb_source_manifest.py | 21 ++- tests/test_compile_kb_source_packet.py | 48 +++++-- .../test_hermes_leoclean_kb_bridge_source.py | 31 ++++- ..._prepare_kb_context_from_canonical_dump.py | 106 ++++++++++++++- tests/test_prepare_kb_source_manifest.py | 85 ++++++++++++ 8 files changed, 463 insertions(+), 120 deletions(-) diff --git a/hermes-agent/leoclean-bin/kb_tool.py b/hermes-agent/leoclean-bin/kb_tool.py index a606d85..f01a3e1 100755 --- a/hermes-agent/leoclean-bin/kb_tool.py +++ b/hermes-agent/leoclean-bin/kb_tool.py @@ -33,7 +33,9 @@ SOURCE_INBOX_ROOT = Path("/home/teleo/.hermes/profiles/leoclean/state/kb-source- SOURCE_PREPARATION_ROOT = Path("/home/teleo/.hermes/profiles/leoclean/state/kb-source-preparation") SOURCE_PROPOSAL_RECEIPT_SCHEMA = "livingip.teleoKbSourceProposalReceipt.v1" SOURCE_PREPARATION_RECEIPT_SCHEMA = "livingip.kbSourcePreparationReceipt.v1" +SOURCE_PREPARATION_STATUS_SCHEMA = "livingip.kbSourcePreparationStatus.v1" SOURCE_COMPILER_BUNDLE_SCHEMA = "livingip.kbSourceProposalBundle.v1" +SOURCE_COMPILER_STATUS_SCHEMA = "livingip.kbSourceCompilerStatus.v1" RETRIEVAL_RECEIPT_SCHEMA = "livingip.teleoKbRetrievalReceipt.v1" CANONICAL_COUNT_LABELS = ("claims", "sources", "claim_edges", "claim_evidence") SHA256_RE = re.compile(r"^[0-9a-f]{64}$") @@ -2362,27 +2364,46 @@ def _validated_source_bundle(bundle: Any) -> tuple[dict[str, Any], dict[str, Any def compile_source_bundle(args: argparse.Namespace) -> dict[str, Any]: if not SOURCE_COMPILER_PATH.is_file(): raise SystemExit(f"source compiler is not deployed: {SOURCE_COMPILER_PATH}") - command = [ - sys.executable, - str(SOURCE_COMPILER_PATH), - "--artifact", - str(args.artifact), - "--text", - str(args.text), - "--manifest", - str(args.manifest), - ] - try: - completed = subprocess.run(command, capture_output=True, text=True, timeout=120, check=False) - except (OSError, subprocess.TimeoutExpired) as exc: - raise SystemExit(f"source compiler could not run: {exc}") from exc - if completed.returncode != 0: - detail = completed.stdout.strip() or completed.stderr.strip() or f"exit {completed.returncode}" - raise SystemExit(f"source compiler rejected the artifact: {detail}") - try: - bundle = json.loads(completed.stdout) - except json.JSONDecodeError as exc: - raise SystemExit(f"source compiler emitted invalid JSON: {exc}") from exc + with tempfile.TemporaryDirectory(prefix="teleo-source-compiler-") as raw_private_dir: + private_dir = Path(raw_private_dir) + os.chmod(private_dir, 0o700) + bundle_path = private_dir / "source-bundle.json" + command = [ + sys.executable, + str(SOURCE_COMPILER_PATH), + "--artifact", + str(args.artifact), + "--text", + str(args.text), + "--manifest", + str(args.manifest), + "--output", + str(bundle_path), + ] + try: + completed = subprocess.run(command, capture_output=True, text=True, timeout=120, check=False) + except (OSError, subprocess.TimeoutExpired) as exc: + raise SystemExit(f"source compiler could not run: {exc}") from exc + if completed.returncode != 0: + detail = completed.stdout.strip() or completed.stderr.strip() or f"exit {completed.returncode}" + raise SystemExit(f"source compiler rejected the artifact: {detail}") + try: + status = json.loads(completed.stdout) + except json.JSONDecodeError as exc: + raise SystemExit("source compiler emitted an invalid public status") from exc + if ( + not isinstance(status, dict) + or status.get("schema") != SOURCE_COMPILER_STATUS_SCHEMA + or status.get("status") != "pending_review" + or status.get("database_write_performed") is not False + or status.get("canonical_apply_performed") is not False + or status.get("private_bundle_written") is not True + ): + raise SystemExit("source compiler violated its private-output status contract") + try: + bundle = json.loads(bundle_path.read_text(encoding="utf-8", errors="strict")) + except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc: + raise SystemExit("source compiler did not write a valid private bundle") from exc _validated_source_bundle(bundle) return bundle @@ -2599,13 +2620,27 @@ def prepare_source(args: argparse.Namespace) -> dict[str, Any]: detail = completed.stdout.strip() or completed.stderr.strip() or f"exit {completed.returncode}" raise SystemExit(f"source preparer rejected the artifact: {detail}") try: - receipt = json.loads(completed.stdout) + status = json.loads(completed.stdout) except json.JSONDecodeError as exc: - raise SystemExit(f"source preparer emitted invalid JSON: {exc}") from exc + raise SystemExit("source preparer emitted an invalid public status") from exc + if ( + not isinstance(status, dict) + or status.get("schema") != SOURCE_PREPARATION_STATUS_SCHEMA + or status.get("status") not in {"prepared", "no_novel_claims"} + or status.get("database_write_performed") is not False + or status.get("canonical_apply_performed") is not False + or status.get("private_outputs_written") is not True + ): + raise SystemExit("source preparer violated its private-output status contract") + receipt_path = args.output_dir / "preparation-receipt.json" + try: + receipt = json.loads(receipt_path.read_text(encoding="utf-8", errors="strict")) + except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc: + raise SystemExit("source preparer did not write a valid private receipt") from exc if ( not isinstance(receipt, dict) or receipt.get("schema") != SOURCE_PREPARATION_RECEIPT_SCHEMA - or receipt.get("status") not in {"prepared", "no_novel_claims"} + or receipt.get("status") != status["status"] or receipt.get("database_write_performed") is not False or receipt.get("canonical_apply_performed") is not False ): @@ -2614,7 +2649,13 @@ def prepare_source(args: argparse.Namespace) -> dict[str, Any]: observed_ids = (receipt.get("retrieval") or {}).get("canonical_candidate_ids") if observed_ids != expected_ids: raise SystemExit("source preparer receipt does not match the canonical retrieval candidates") - return receipt + return { + "schema": SOURCE_PREPARATION_STATUS_SCHEMA, + "status": receipt["status"], + "database_write_performed": False, + "canonical_apply_performed": False, + "private_outputs_written": True, + } def propose_source(args: argparse.Namespace) -> dict[str, Any]: @@ -3537,19 +3578,10 @@ def print_source_proposal_receipt(data: dict[str, Any]) -> None: def print_source_preparation_receipt(data: dict[str, Any]) -> None: - retrieval = data["retrieval"] - extraction = data["extraction"] - outputs = data["outputs"] - print("# Source Preparation Receipt\n") + print("# Source Preparation Status\n") print(f"- status: `{data['status']}`") - print(f"- model: `{extraction['model']}`; claims prepared: `{extraction['claim_count']}`") - print(f"- canonical candidates checked: `{retrieval['canonical_candidate_count']}`") - print(f"- duplicate judgments: `{len(retrieval['duplicate_judgments'])}`") print("- database write performed: `false`; canonical apply performed: `false`") - print(f"- extracted text: `{outputs['extracted_text']}`") - print(f"- manifest: `{outputs['manifest'] or '-'}`") - print(f"- receipt: `{outputs['receipt']}`") - print(f"- next: {data['next_action']}") + print("- private outputs written: `true`; inspect the requested output directory locally") def main() -> int: diff --git a/scripts/compile_kb_source_packet.py b/scripts/compile_kb_source_packet.py index 88d09b4..103634c 100644 --- a/scripts/compile_kb_source_packet.py +++ b/scripts/compile_kb_source_packet.py @@ -32,6 +32,7 @@ MANIFEST_SCHEMA = "livingip.kbSourceExtractionManifest.v1" MANIFEST_SCHEMA_V2 = "livingip.kbSourceExtractionManifest.v2" MANIFEST_SCHEMAS = {MANIFEST_SCHEMA, MANIFEST_SCHEMA_V2} BUNDLE_SCHEMA = "livingip.kbSourceProposalBundle.v1" +STATUS_SCHEMA = "livingip.kbSourceCompilerStatus.v1" SHA256_RE = re.compile(r"^[0-9a-f]{64}$") KEY_RE = re.compile(r"^[a-z0-9][a-z0-9._:-]{0,127}$") URI_SCHEME_RE = re.compile(r"^[a-z][a-z0-9+.-]*:", re.IGNORECASE) @@ -572,7 +573,7 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser.add_argument("--artifact", required=True, type=Path, help="raw source artifact; any binary format") parser.add_argument("--text", required=True, type=Path, help="strict UTF-8 extraction of the artifact") parser.add_argument("--manifest", required=True, type=Path, help="extraction manifest JSON") - parser.add_argument("--output", type=Path, help="optional path for the same JSON emitted to stdout") + parser.add_argument("--output", required=True, type=Path, help="private path for the compiled proposal bundle") return parser.parse_args(argv) @@ -593,47 +594,38 @@ def _write_private_output(path: Path, rendered: str) -> None: temp.unlink() +def _print_status(status: str, *, reason: str | None = None) -> None: + payload: dict[str, Any] = { + "schema": STATUS_SCHEMA, + "status": status, + "database_write_performed": False, + "canonical_apply_performed": False, + "private_bundle_written": status == "pending_review", + } + if reason is not None: + payload["reason"] = reason + print(json.dumps(payload, sort_keys=True)) + + def main(argv: list[str] | None = None) -> int: args = parse_args(argv) try: - if args.output: - output_path = args.output.resolve() - input_paths = {args.artifact.resolve(), args.text.resolve(), args.manifest.resolve()} - if output_path in input_paths: - raise CompilerError("--output must not overwrite an input file") + output_path = args.output.resolve() + input_paths = {args.artifact.resolve(), args.text.resolve(), args.manifest.resolve()} + if output_path in input_paths: + raise CompilerError("--output must not overwrite an input file") bundle = compile_source_packet(args.artifact, args.text, args.manifest) - except (CompilerError, OSError, ValueError) as exc: - print( - json.dumps( - { - "schema": BUNDLE_SCHEMA, - "status": "rejected", - "database_write_performed": False, - "error": str(exc), - }, - sort_keys=True, - ) - ) + except (CompilerError, OSError, ValueError): + _print_status("rejected", reason="input_validation_failed") return 2 rendered = json.dumps(bundle, indent=2, sort_keys=True, ensure_ascii=True) + "\n" - if args.output: - try: - _write_private_output(args.output, rendered) - except OSError as exc: - print( - json.dumps( - { - "schema": BUNDLE_SCHEMA, - "status": "rejected", - "database_write_performed": False, - "error": f"could not write output: {exc}", - }, - sort_keys=True, - ) - ) - return 2 - print(rendered, end="") + try: + _write_private_output(args.output, rendered) + except OSError: + _print_status("rejected", reason="output_write_failed") + return 2 + _print_status("pending_review") return 0 diff --git a/scripts/prepare_kb_context_from_canonical_dump.py b/scripts/prepare_kb_context_from_canonical_dump.py index d5e26ce..667ca2f 100644 --- a/scripts/prepare_kb_context_from_canonical_dump.py +++ b/scripts/prepare_kb_context_from_canonical_dump.py @@ -4,6 +4,7 @@ from __future__ import annotations import argparse +import codecs import hashlib import importlib.util import json @@ -19,8 +20,10 @@ from types import ModuleType from typing import Any SCHEMA = "livingip.kbSourceDumpContextReceipt.v1" +STATUS_SCHEMA = "livingip.kbSourceDumpContextStatus.v1" DEFAULT_DOCKER_IMAGE = "postgres:16-alpine" DEFAULT_LIMIT = 12 +DEFAULT_COPY_ENCODING = "utf-8" MAX_PREVIEW_BYTES = 16_000 TABLES = ("claims", "claim_evidence", "claim_edges") @@ -52,18 +55,35 @@ def _load_kb_tool(repo_root: Path) -> ModuleType: return module -def _decode_copy_field(value: str) -> str | None: +def _copy_encoding(value: str) -> str: + try: + return codecs.lookup(value).name + except LookupError as exc: + raise ContextError("COPY encoding is not supported") from exc + + +def _decode_copy_field(value: str, *, encoding: str = DEFAULT_COPY_ENCODING) -> str | None: if value == r"\N": return None - result: list[str] = [] + encoding = _copy_encoding(encoding) + result = bytearray() index = 0 - escapes = {"b": "\b", "f": "\f", "n": "\n", "r": "\r", "t": "\t", "v": "\v", "\\": "\\"} + escapes = {"b": 8, "f": 12, "n": 10, "r": 13, "t": 9, "v": 11, "\\": 92} + + def append_text(text: str) -> None: + try: + result.extend(text.encode(encoding, errors="strict")) + except UnicodeEncodeError as exc: + raise ContextError(f"COPY field cannot be encoded as {encoding}") from exc + while index < len(value): char = value[index] - if char != "\\" or index + 1 >= len(value): - result.append(char) + if char != "\\": + append_text(char) index += 1 continue + if index + 1 >= len(value): + raise ContextError("COPY field contains a dangling backslash escape") index += 1 escaped = value[index] if escaped in escapes: @@ -72,22 +92,35 @@ def _decode_copy_field(value: str) -> str | None: continue if escaped == "x": match = re.match(r"[0-9A-Fa-f]{1,2}", value[index + 1 :]) - if match: - result.append(chr(int(match.group(0), 16))) - index += 1 + len(match.group(0)) - continue + if match is None: + raise ContextError("COPY field contains a malformed hexadecimal byte escape") + result.append(int(match.group(0), 16)) + index += 1 + len(match.group(0)) + continue if escaped in "01234567": match = re.match(r"[0-7]{1,3}", value[index:]) - if match: - result.append(chr(int(match.group(0), 8))) - index += len(match.group(0)) - continue - result.append(escaped) + if match is None: + raise ContextError("COPY field contains a malformed octal byte escape") + byte_value = int(match.group(0), 8) + if byte_value > 0xFF: + raise ContextError("COPY field contains an out-of-range octal byte escape") + result.append(byte_value) + index += len(match.group(0)) + continue + append_text(escaped) index += 1 - return "".join(result) + if 0 in result: + raise ContextError("COPY field contains a zero byte") + try: + return result.decode(encoding, errors="strict") + except UnicodeDecodeError as exc: + raise ContextError(f"COPY byte escapes do not form valid {encoding} text") from exc -def parse_copy_table(sql: str, table: str) -> list[dict[str, str | None]]: +def parse_copy_table( + sql: str, table: str, *, encoding: str = DEFAULT_COPY_ENCODING +) -> list[dict[str, str | None]]: + encoding = _copy_encoding(encoding) prefix = f"COPY public.{table} (" lines = iter(sql.splitlines()) for line in lines: @@ -103,11 +136,15 @@ def parse_copy_table(sql: str, table: str) -> list[dict[str, str | None]]: fields = line.split("\t") if len(fields) != len(columns): raise ContextError(f"public.{table} COPY row has {len(fields)} fields; expected {len(columns)}") - rows.append(dict(zip(columns, (_decode_copy_field(field) for field in fields), strict=True))) + rows.append( + dict(zip(columns, (_decode_copy_field(field, encoding=encoding) for field in fields), strict=True)) + ) raise ContextError(f"public.{table} COPY block is unterminated") -def restore_table(dump_path: Path, table: str, docker_image: str) -> str: +def restore_table( + dump_path: Path, table: str, docker_image: str, *, encoding: str = DEFAULT_COPY_ENCODING +) -> str: if table not in TABLES: raise ContextError(f"unsupported canonical table: {table}") if shutil.which("docker") is None: @@ -128,9 +165,18 @@ def restore_table(dump_path: Path, table: str, docker_image: str) -> str: "-", "/canonical.dump", ] + encoding = _copy_encoding(encoding) try: - completed = subprocess.run(command, capture_output=True, text=True, timeout=180, check=False) - except (OSError, subprocess.TimeoutExpired) as exc: + completed = subprocess.run( + command, + capture_output=True, + text=True, + encoding=encoding, + errors="strict", + timeout=180, + check=False, + ) + except (OSError, subprocess.TimeoutExpired, UnicodeError) as exc: raise ContextError(f"pg_restore failed ({type(exc).__name__}); diagnostic detail redacted") from exc if completed.returncode != 0: raise ContextError(f"pg_restore returned exit {completed.returncode}; diagnostic detail redacted") @@ -147,7 +193,14 @@ def _decimal(value: str | None) -> Decimal: def build_context( - *, dump_path: Path, artifact_path: Path, title: str, limit: int, docker_image: str, repo_root: Path + *, + dump_path: Path, + artifact_path: Path, + title: str, + limit: int, + docker_image: str, + repo_root: Path, + copy_encoding: str = DEFAULT_COPY_ENCODING, ) -> tuple[dict[str, Any], dict[str, Any]]: if limit < 1 or limit > 20: raise ContextError("limit must be between 1 and 20") @@ -162,7 +215,15 @@ def build_context( terms = kb_tool.query_terms(query) min_score = 2 if len(terms) >= 3 else 1 - tables = {table: parse_copy_table(restore_table(dump_path, table, docker_image), table) for table in TABLES} + copy_encoding = _copy_encoding(copy_encoding) + tables = { + table: parse_copy_table( + restore_table(dump_path, table, docker_image, encoding=copy_encoding), + table, + encoding=copy_encoding, + ) + for table in TABLES + } evidence_counts = Counter(row["claim_id"] for row in tables["claim_evidence"] if row["claim_id"]) edge_counts: Counter[str] = Counter() for row in tables["claim_edges"]: @@ -228,6 +289,8 @@ def build_context( "docker_image": docker_image, "network_disabled": True, "pg_restore_output_retained": False, + "copy_encoding": copy_encoding, + "copy_decode_errors": "strict", }, } return context, receipt @@ -263,6 +326,7 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser.add_argument("--receipt-output", required=True, type=Path) parser.add_argument("--limit", type=int, default=DEFAULT_LIMIT) parser.add_argument("--docker-image", default=DEFAULT_DOCKER_IMAGE) + parser.add_argument("--copy-encoding", default=DEFAULT_COPY_ENCODING) return parser.parse_args(argv) @@ -279,6 +343,7 @@ def main(argv: list[str] | None = None) -> int: limit=args.limit, docker_image=args.docker_image, repo_root=repo_root, + copy_encoding=args.copy_encoding, ) _private_json(args.context_output, context) receipt["outputs"] = { @@ -286,14 +351,17 @@ def main(argv: list[str] | None = None) -> int: "receipt": str(args.receipt_output.resolve()), } _private_json(args.receipt_output, receipt) - except (ContextError, OSError, ValueError) as exc: + except (ContextError, OSError, ValueError): print( json.dumps( { - "schema": SCHEMA, + "schema": STATUS_SCHEMA, "status": "rejected", "database_write_performed": False, - "error": str(exc), + "model_call_performed": False, + "private_context_written": False, + "private_receipt_written": False, + "reason": "context_preparation_failed", }, sort_keys=True, ) @@ -302,12 +370,12 @@ def main(argv: list[str] | None = None) -> int: print( json.dumps( { - "schema": SCHEMA, + "schema": STATUS_SCHEMA, "status": "prepared", "database_write_performed": False, - "candidate_count": receipt["retrieval"]["candidate_count"], - "context_sha256": sha256_bytes(json.dumps(context, sort_keys=True).encode("utf-8")), - "outputs": receipt["outputs"], + "model_call_performed": False, + "private_context_written": True, + "private_receipt_written": True, }, indent=2, sort_keys=True, diff --git a/scripts/prepare_kb_source_manifest.py b/scripts/prepare_kb_source_manifest.py index e9b8ade..073b237 100644 --- a/scripts/prepare_kb_source_manifest.py +++ b/scripts/prepare_kb_source_manifest.py @@ -30,6 +30,7 @@ sys.path.insert(0, str(HERE)) import compile_kb_source_packet as compiler # noqa: E402 RECEIPT_SCHEMA = "livingip.kbSourcePreparationReceipt.v1" +STATUS_SCHEMA = "livingip.kbSourcePreparationStatus.v1" MODEL_OUTPUT_SCHEMA = "livingip.kbSourceModelExtraction.v2" RESOLVED_OUTPUT_SCHEMA = "livingip.kbSourceResolvedExtraction.v1" DEFAULT_MODEL = "anthropic/claude-sonnet-4.5" @@ -800,21 +801,33 @@ def main(argv: list[str] | None = None) -> int: args = parse_args(argv) try: receipt = prepare(args) - except (PreparationError, compiler.CompilerError, OSError, ValueError) as exc: + except (PreparationError, compiler.CompilerError, OSError, ValueError): print( json.dumps( { - "schema": RECEIPT_SCHEMA, + "schema": STATUS_SCHEMA, "status": "rejected", "database_write_performed": False, "canonical_apply_performed": False, - "error": str(exc), + "private_outputs_written": False, + "reason": "source_preparation_failed", }, sort_keys=True, ) ) return 2 - print(json.dumps(receipt, indent=2, sort_keys=True, ensure_ascii=True)) + print( + json.dumps( + { + "schema": STATUS_SCHEMA, + "status": receipt["status"], + "database_write_performed": False, + "canonical_apply_performed": False, + "private_outputs_written": True, + }, + sort_keys=True, + ) + ) return 0 diff --git a/tests/test_compile_kb_source_packet.py b/tests/test_compile_kb_source_packet.py index a220b3d..6978699 100644 --- a/tests/test_compile_kb_source_packet.py +++ b/tests/test_compile_kb_source_packet.py @@ -4,6 +4,7 @@ import copy import hashlib import json import stat +import subprocess import sys from pathlib import Path @@ -208,12 +209,15 @@ def test_v2_manifest_rejects_unretrieved_or_unbound_duplicate_judgments(tmp_path _compile(tmp_path / "unbound", manifest) -def test_cli_emits_structured_json_to_stdout_and_optional_output(tmp_path: Path, capsys: pytest.CaptureFixture) -> None: - artifact, text, manifest_path = _write_inputs(tmp_path) - output = tmp_path / "compiled" / "packet.json" +def test_cli_writes_private_bundle_and_emits_only_public_status_bytes(tmp_path: Path) -> None: + private_root = tmp_path / "private-source-sentinel" + artifact, text, manifest_path = _write_inputs(private_root) + output = private_root / "compiled" / "packet.json" - status = compiler.main( + completed = subprocess.run( [ + sys.executable, + str(REPO_ROOT / "scripts" / "compile_kb_source_packet.py"), "--artifact", str(artifact), "--text", @@ -222,13 +226,33 @@ def test_cli_emits_structured_json_to_stdout_and_optional_output(tmp_path: Path, str(manifest_path), "--output", str(output), - ] + ], + capture_output=True, + check=False, ) - stdout = capsys.readouterr().out - assert status == 0 - assert output.read_text(encoding="utf-8") == stdout - assert json.loads(stdout)["status"] == "pending_review" + stdout = completed.stdout + public_status = json.loads(stdout.decode("utf-8", errors="strict")) + bundle = json.loads(output.read_text(encoding="utf-8")) + assert completed.returncode == 0 + assert completed.stderr == b"" + assert public_status == { + "schema": compiler.STATUS_SCHEMA, + "status": "pending_review", + "database_write_performed": False, + "canonical_apply_performed": False, + "private_bundle_written": True, + } + assert bundle["status"] == "pending_review" + assert output.read_bytes() != stdout + for private_value in ( + str(private_root), + str(artifact), + str(output), + EXTRACTED_TEXT, + bundle["validated_manifest"]["source"]["title"], + ): + assert private_value.encode("utf-8") not in stdout assert stat.S_IMODE(output.stat().st_mode) == 0o600 @@ -357,7 +381,8 @@ def test_rejects_path_gaps_invalid_utf8_and_output_overwrite(tmp_path: Path, cap assert status == 2 assert rejection["status"] == "rejected" assert rejection["database_write_performed"] is False - assert "must not overwrite" in rejection["error"] + assert rejection["reason"] == "input_validation_failed" + assert str(artifact) not in json.dumps(rejection) def test_cli_reports_output_write_failure_as_structured_rejection( @@ -387,7 +412,8 @@ def test_cli_reports_output_write_failure_as_structured_rejection( assert status == 2 assert rejection["status"] == "rejected" assert rejection["database_write_performed"] is False - assert "could not write output" in rejection["error"] + assert rejection["reason"] == "output_write_failed" + assert "fixture output denied" not in json.dumps(rejection) def test_rejects_attempt_to_invent_source_from_chat_label(tmp_path: Path) -> None: diff --git a/tests/test_hermes_leoclean_kb_bridge_source.py b/tests/test_hermes_leoclean_kb_bridge_source.py index af9d3f7..4c6daaa 100644 --- a/tests/test_hermes_leoclean_kb_bridge_source.py +++ b/tests/test_hermes_leoclean_kb_bridge_source.py @@ -788,7 +788,9 @@ def test_vps_bridge_binds_claim_challenge_to_exact_retrieved_rows() -> None: assert "no database write was made" in unbound_response -def test_vps_prepare_source_retrieves_canonical_candidates_before_model_extraction(tmp_path: Path, monkeypatch) -> None: +def test_vps_prepare_source_retrieves_canonical_candidates_before_model_extraction( + tmp_path: Path, monkeypatch, capsys +) -> None: module = _load_module(BRIDGE_DIR / "kb_tool.py") preparer = tmp_path / "prepare_kb_source_manifest.py" preparer.write_text("# fixture\n", encoding="utf-8") @@ -823,16 +825,29 @@ def test_vps_prepare_source_retrieves_canonical_candidates_before_model_extracti context_path = Path(command[command.index("--kb-context") + 1]) captured["context"] = json.loads(context_path.read_text(encoding="utf-8")) captured["command"] = command - receipt = { + private_output_dir = Path(command[command.index("--output-dir") + 1]) + private_output_dir.mkdir(parents=True) + private_receipt = { "schema": module.SOURCE_PREPARATION_RECEIPT_SCHEMA, "status": "prepared", "database_write_performed": False, "canonical_apply_performed": False, "retrieval": {"canonical_candidate_ids": [candidate_id]}, "extraction": {"model": "fixture/model", "claim_count": 1}, - "outputs": {}, + "source": {"title": "PRIVATE-SOURCE-TITLE-SENTINEL"}, + "outputs": {"receipt": str(private_output_dir / "preparation-receipt.json")}, } - return subprocess.CompletedProcess(command, 0, stdout=json.dumps(receipt), stderr="") + (private_output_dir / "preparation-receipt.json").write_text( + json.dumps(private_receipt), encoding="utf-8" + ) + public_status = { + "schema": module.SOURCE_PREPARATION_STATUS_SCHEMA, + "status": "prepared", + "database_write_performed": False, + "canonical_apply_performed": False, + "private_outputs_written": True, + } + return subprocess.CompletedProcess(command, 0, stdout=json.dumps(public_status), stderr="") monkeypatch.setattr(module.subprocess, "run", fake_run) args = SimpleNamespace( @@ -855,6 +870,14 @@ def test_vps_prepare_source_retrieves_canonical_candidates_before_model_extracti assert captured["context"]["database_search_query"] == args.query assert captured["context"]["candidate_claims"][0]["id"] == candidate_id assert "--text" not in captured["command"] + assert "source" not in receipt + assert "outputs" not in receipt + + module.print_source_preparation_receipt(receipt) + stdout = capsys.readouterr().out + assert "PRIVATE-SOURCE-TITLE-SENTINEL" not in stdout + assert str(output_dir) not in stdout + assert candidate_id not in stdout def test_vps_prepare_source_rejects_files_and_outputs_outside_private_roots(tmp_path: Path) -> None: diff --git a/tests/test_prepare_kb_context_from_canonical_dump.py b/tests/test_prepare_kb_context_from_canonical_dump.py index 5272afc..09c1ab5 100644 --- a/tests/test_prepare_kb_context_from_canonical_dump.py +++ b/tests/test_prepare_kb_context_from_canonical_dump.py @@ -88,7 +88,7 @@ def test_build_context_matches_readonly_claim_ranking_and_counts( [["44444444-4444-4444-8444-444444444444", CLAIM_A, CLAIM_B, "supports", "1", r"\N", "2026-01-01"]], ), } - monkeypatch.setattr(context_builder, "restore_table", lambda _dump, table, _image: sql[table]) + monkeypatch.setattr(context_builder, "restore_table", lambda _dump, table, _image, **_kwargs: sql[table]) context, receipt = context_builder.build_context( dump_path=dump, @@ -106,6 +106,8 @@ def test_build_context_matches_readonly_claim_ranking_and_counts( assert receipt["model_call_performed"] is False assert receipt["runtime"]["network_disabled"] is True assert receipt["runtime"]["pg_restore_output_retained"] is False + assert receipt["runtime"]["copy_encoding"] == "utf-8" + assert receipt["runtime"]["copy_decode_errors"] == "strict" def test_copy_parser_preserves_escaped_source_text() -> None: @@ -116,6 +118,31 @@ def test_copy_parser_preserves_escaped_source_text() -> None: assert rows == [{"id": CLAIM_A, "text": "line one\nline two\tvalue\\tail"}] +@pytest.mark.parametrize( + ("encoded", "decoded"), + [ + (r"caf\303\251", "caf\N{LATIN SMALL LETTER E WITH ACUTE}"), + (r"caf\xc3\xa9", "caf\N{LATIN SMALL LETTER E WITH ACUTE}"), + (r"price \342\202\2545", "price \N{EURO SIGN}5"), + (r"literal\qescape", "literalqescape"), + ], +) +def test_copy_parser_decodes_octal_and_hex_as_encoded_bytes(encoded: str, decoded: str) -> None: + assert context_builder._decode_copy_field(encoded) == decoded + + +def test_copy_parser_uses_explicit_encoding() -> None: + assert context_builder._decode_copy_field(r"caf\351", encoding="latin-1") == "caf\N{LATIN SMALL LETTER E WITH ACUTE}" + with pytest.raises(context_builder.ContextError, match="valid utf-8"): + context_builder._decode_copy_field(r"caf\351", encoding="utf-8") + + +@pytest.mark.parametrize("encoded", [r"\xff", r"\303", r"\400", r"\000", r"\x", "dangling\\"]) +def test_copy_parser_fails_closed_on_invalid_byte_escapes(encoded: str) -> None: + with pytest.raises(context_builder.ContextError): + context_builder._decode_copy_field(encoded) + + def test_private_json_is_owner_only_and_refuses_overwrite(tmp_path: Path) -> None: output = tmp_path / "private" / "context.json" @@ -126,3 +153,80 @@ def test_private_json_is_owner_only_and_refuses_overwrite(tmp_path: Path) -> Non assert stat.S_IMODE(output.stat().st_mode) == 0o600 with pytest.raises(context_builder.ContextError, match="refusing to overwrite"): context_builder._private_json(output, {"safe": False}) + + +def test_cli_keeps_private_context_receipt_paths_and_details_off_stdout( + tmp_path: Path, capsys: pytest.CaptureFixture, monkeypatch: pytest.MonkeyPatch +) -> None: + sentinel = "PRIVATE-CANONICAL-DETAIL-SENTINEL" + dump = tmp_path / "private-dump-sentinel.dump" + artifact = tmp_path / "private-source-sentinel.txt" + context_output = tmp_path / "private-output" / "context.json" + receipt_output = tmp_path / "private-output" / "receipt.json" + context = {"database_search_query": sentinel, "candidate_claims": [{"id": CLAIM_A, "text": sentinel}]} + receipt = {"source": {"path": str(artifact), "private_detail": sentinel}} + monkeypatch.setattr(context_builder, "build_context", lambda **_kwargs: (context, receipt)) + + status = context_builder.main( + [ + "--dump", + str(dump), + "--artifact", + str(artifact), + "--title", + sentinel, + "--context-output", + str(context_output), + "--receipt-output", + str(receipt_output), + ] + ) + + stdout = capsys.readouterr().out + public_status = json.loads(stdout) + assert status == 0 + assert public_status == { + "schema": context_builder.STATUS_SCHEMA, + "status": "prepared", + "database_write_performed": False, + "model_call_performed": False, + "private_context_written": True, + "private_receipt_written": True, + } + assert sentinel in context_output.read_text(encoding="utf-8") + assert sentinel in receipt_output.read_text(encoding="utf-8") + for private_value in (sentinel, str(dump), str(artifact), str(context_output), str(receipt_output)): + assert private_value not in stdout + + +def test_cli_redacts_private_failure_details_from_stdout( + tmp_path: Path, capsys: pytest.CaptureFixture, monkeypatch: pytest.MonkeyPatch +) -> None: + sentinel = "PRIVATE-CONTEXT-FAILURE-SENTINEL" + private_path = tmp_path / sentinel / "source.txt" + + def fail(**_kwargs): + raise context_builder.ContextError(f"{sentinel}: {private_path}") + + monkeypatch.setattr(context_builder, "build_context", fail) + status = context_builder.main( + [ + "--dump", + str(tmp_path / "dump"), + "--artifact", + str(private_path), + "--title", + sentinel, + "--context-output", + str(tmp_path / "context.json"), + "--receipt-output", + str(tmp_path / "receipt.json"), + ] + ) + + stdout = capsys.readouterr().out + public_status = json.loads(stdout) + assert status == 2 + assert public_status["reason"] == "context_preparation_failed" + assert sentinel not in stdout + assert str(private_path) not in stdout diff --git a/tests/test_prepare_kb_source_manifest.py b/tests/test_prepare_kb_source_manifest.py index f24335a..3a24148 100644 --- a/tests/test_prepare_kb_source_manifest.py +++ b/tests/test_prepare_kb_source_manifest.py @@ -409,3 +409,88 @@ def test_provider_errors_redact_secret_and_private_detail(tmp_path: Path, monkey assert "fixture-secret" not in rendered assert "private source quote" not in rendered assert "provider detail redacted" in rendered + + +def test_cli_emits_only_public_status_for_private_preparation_receipt( + tmp_path: Path, capsys: pytest.CaptureFixture, monkeypatch: pytest.MonkeyPatch +) -> None: + sentinel = "PRIVATE-SOURCE-RECEIPT-SENTINEL" + private_path = tmp_path / sentinel / "preparation-receipt.json" + private_receipt = { + "status": "prepared", + "source": {"title": sentinel, "artifact_path": str(private_path)}, + "retrieval": {"database_search_query": sentinel}, + "outputs": {"receipt": str(private_path)}, + } + monkeypatch.setattr(preparer, "prepare", lambda _args: private_receipt) + + status = preparer.main( + [ + "--artifact", + str(tmp_path / sentinel / "source.md"), + "--kb-context", + str(tmp_path / sentinel / "context.json"), + "--identity", + f"document:{sentinel}", + "--source-key", + "private_source", + "--source-type", + "article", + "--title", + sentinel, + "--locator", + f"artifact://{sentinel}", + "--output-dir", + str(private_path.parent), + ] + ) + + stdout = capsys.readouterr().out + assert status == 0 + assert json.loads(stdout) == { + "schema": preparer.STATUS_SCHEMA, + "status": "prepared", + "database_write_performed": False, + "canonical_apply_performed": False, + "private_outputs_written": True, + } + assert sentinel not in stdout + assert str(private_path) not in stdout + + +def test_cli_redacts_private_preparation_failure_from_stdout( + tmp_path: Path, capsys: pytest.CaptureFixture, monkeypatch: pytest.MonkeyPatch +) -> None: + sentinel = "PRIVATE-PREPARATION-FAILURE-SENTINEL" + + def fail(_args): + raise preparer.PreparationError(f"{sentinel}: {tmp_path / sentinel}") + + monkeypatch.setattr(preparer, "prepare", fail) + status = preparer.main( + [ + "--artifact", + str(tmp_path / sentinel / "source.md"), + "--kb-context", + str(tmp_path / sentinel / "context.json"), + "--identity", + f"document:{sentinel}", + "--source-key", + "private_source", + "--source-type", + "article", + "--title", + sentinel, + "--locator", + f"artifact://{sentinel}", + "--output-dir", + str(tmp_path / sentinel / "output"), + ] + ) + + stdout = capsys.readouterr().out + public_status = json.loads(stdout) + assert status == 2 + assert public_status["reason"] == "source_preparation_failed" + assert sentinel not in stdout + assert str(tmp_path) not in stdout From d3b151feb75dc00b9076ed25aa5b3029dec09b89 Mon Sep 17 00:00:00 2001 From: twentyOne2x Date: Thu, 16 Jul 2026 12:47:52 +0200 Subject: [PATCH 3/3] Harden private source reconstruction failures --- hermes-agent/leoclean-bin/kb_tool.py | 18 +-- .../prepare_kb_context_from_canonical_dump.py | 58 ++++++-- scripts/prepare_kb_source_manifest.py | 61 +++++++- .../test_hermes_leoclean_kb_bridge_source.py | 63 ++++++++ tests/test_kb_tool_propose_source.py | 46 ++++++ ..._prepare_kb_context_from_canonical_dump.py | 72 +++++++++- tests/test_prepare_kb_source_manifest.py | 136 ++++++++++++++++++ 7 files changed, 425 insertions(+), 29 deletions(-) diff --git a/hermes-agent/leoclean-bin/kb_tool.py b/hermes-agent/leoclean-bin/kb_tool.py index f01a3e1..0557d1d 100755 --- a/hermes-agent/leoclean-bin/kb_tool.py +++ b/hermes-agent/leoclean-bin/kb_tool.py @@ -2382,11 +2382,12 @@ def compile_source_bundle(args: argparse.Namespace) -> dict[str, Any]: ] try: completed = subprocess.run(command, capture_output=True, text=True, timeout=120, check=False) - except (OSError, subprocess.TimeoutExpired) as exc: - raise SystemExit(f"source compiler could not run: {exc}") from exc + except subprocess.TimeoutExpired: + raise SystemExit("source compiler timed out; private diagnostic detail redacted") from None + except OSError: + raise SystemExit("source compiler could not start; private diagnostic detail redacted") from None if completed.returncode != 0: - detail = completed.stdout.strip() or completed.stderr.strip() or f"exit {completed.returncode}" - raise SystemExit(f"source compiler rejected the artifact: {detail}") + raise SystemExit("source compiler failed; private diagnostic detail redacted") try: status = json.loads(completed.stdout) except json.JSONDecodeError as exc: @@ -2610,15 +2611,16 @@ def prepare_source(args: argparse.Namespace) -> dict[str, Any]: command.extend(("--text", str(args.text))) try: completed = subprocess.run(command, capture_output=True, text=True, timeout=420, check=False) - except (OSError, subprocess.TimeoutExpired) as exc: - raise SystemExit(f"source preparer could not run: {exc}") from exc + except subprocess.TimeoutExpired: + raise SystemExit("source preparer timed out; private diagnostic detail redacted") from None + except OSError: + raise SystemExit("source preparer could not start; private diagnostic detail redacted") from None finally: if context_path.exists(): context_path.unlink() if completed.returncode != 0: - detail = completed.stdout.strip() or completed.stderr.strip() or f"exit {completed.returncode}" - raise SystemExit(f"source preparer rejected the artifact: {detail}") + raise SystemExit("source preparer failed; private diagnostic detail redacted") try: status = json.loads(completed.stdout) except json.JSONDecodeError as exc: diff --git a/scripts/prepare_kb_context_from_canonical_dump.py b/scripts/prepare_kb_context_from_canonical_dump.py index 667ca2f..833e5e0 100644 --- a/scripts/prepare_kb_context_from_canonical_dump.py +++ b/scripts/prepare_kb_context_from_canonical_dump.py @@ -92,11 +92,10 @@ def _decode_copy_field(value: str, *, encoding: str = DEFAULT_COPY_ENCODING) -> continue if escaped == "x": match = re.match(r"[0-9A-Fa-f]{1,2}", value[index + 1 :]) - if match is None: - raise ContextError("COPY field contains a malformed hexadecimal byte escape") - result.append(int(match.group(0), 16)) - index += 1 + len(match.group(0)) - continue + if match is not None: + result.append(int(match.group(0), 16)) + index += 1 + len(match.group(0)) + continue if escaped in "01234567": match = re.match(r"[0-7]{1,3}", value[index:]) if match is None: @@ -122,7 +121,7 @@ def parse_copy_table( ) -> list[dict[str, str | None]]: encoding = _copy_encoding(encoding) prefix = f"COPY public.{table} (" - lines = iter(sql.splitlines()) + lines = iter(line[:-1] if line.endswith("\r") else line for line in sql.split("\n")) for line in lines: if line.startswith(prefix) and line.endswith(") FROM stdin;"): columns = [item.strip() for item in line[len(prefix) : -len(") FROM stdin;")].split(",")] @@ -296,9 +295,7 @@ def build_context( return context, receipt -def _private_json(path: Path, value: Any) -> None: - if path.exists(): - raise ContextError(f"refusing to overwrite existing output: {path}") +def _stage_private_json(path: Path, value: Any) -> Path: path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) os.chmod(path.parent, 0o700) fd, raw_temp = tempfile.mkstemp(prefix=f".{path.name}.", suffix=".tmp", dir=path.parent) @@ -310,6 +307,18 @@ def _private_json(path: Path, value: Any) -> None: handle.write("\n") handle.flush() os.fsync(handle.fileno()) + return temp + except BaseException: + if temp.exists(): + temp.unlink() + raise + + +def _private_json(path: Path, value: Any) -> None: + if path.exists(): + raise ContextError(f"refusing to overwrite existing output: {path}") + temp = _stage_private_json(path, value) + try: os.replace(temp, path) os.chmod(path, 0o600) finally: @@ -317,6 +326,24 @@ def _private_json(path: Path, value: Any) -> None: temp.unlink() +def _publish_private_json_outputs(outputs: tuple[tuple[Path, Any], ...]) -> None: + for path, _value in outputs: + if path.exists(): + raise ContextError(f"refusing to overwrite existing output: {path}") + + staged: list[tuple[Path, Path]] = [] + try: + for path, value in outputs: + staged.append((path, _stage_private_json(path, value))) + for path, temp in staged: + os.replace(temp, path) + os.chmod(path, 0o600) + finally: + for _path, temp in staged: + if temp.exists(): + temp.unlink() + + def parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--dump", required=True, type=Path) @@ -333,6 +360,8 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace: def main(argv: list[str] | None = None) -> int: args = parse_args(argv) repo_root = Path(__file__).resolve().parents[1] + context_existed = args.context_output.is_file() + receipt_existed = args.receipt_output.is_file() try: if args.context_output.resolve() == args.receipt_output.resolve(): raise ContextError("context and receipt outputs must be different files") @@ -345,13 +374,16 @@ def main(argv: list[str] | None = None) -> int: repo_root=repo_root, copy_encoding=args.copy_encoding, ) - _private_json(args.context_output, context) receipt["outputs"] = { "context": str(args.context_output.resolve()), "receipt": str(args.receipt_output.resolve()), } - _private_json(args.receipt_output, receipt) + _publish_private_json_outputs( + ((args.context_output, context), (args.receipt_output, receipt)) + ) except (ContextError, OSError, ValueError): + context_written = not context_existed and args.context_output.is_file() + receipt_written = not receipt_existed and args.receipt_output.is_file() print( json.dumps( { @@ -359,8 +391,8 @@ def main(argv: list[str] | None = None) -> int: "status": "rejected", "database_write_performed": False, "model_call_performed": False, - "private_context_written": False, - "private_receipt_written": False, + "private_context_written": context_written, + "private_receipt_written": receipt_written, "reason": "context_preparation_failed", }, sort_keys=True, diff --git a/scripts/prepare_kb_source_manifest.py b/scripts/prepare_kb_source_manifest.py index 073b237..6beefda 100644 --- a/scripts/prepare_kb_source_manifest.py +++ b/scripts/prepare_kb_source_manifest.py @@ -594,7 +594,7 @@ def _validate_metadata(args: argparse.Namespace) -> None: raise PreparationError(f"--source-type must be one of {sorted(compiler.ap.SOURCE_TYPES)}") -def prepare(args: argparse.Namespace) -> dict[str, Any]: +def _prepare_staged(args: argparse.Namespace, published_output_dir: Path) -> dict[str, Any]: _validate_metadata(args) artifact_bytes = _read_nonempty(args.artifact, "artifact") text_bytes = extract_utf8_text(args.artifact, args.text) @@ -751,10 +751,10 @@ def prepare(args: argparse.Namespace) -> dict[str, Any]: "exact_selected_locations_validated": bundle is not None, }, "outputs": { - "output_dir": str(output_dir), - "extracted_text": str(text_path), - "extraction_result": str(extraction_path), - "manifest": str(manifest_path) if status == "prepared" else None, + "output_dir": str(published_output_dir), + "extracted_text": str(published_output_dir / text_path.name), + "extraction_result": str(published_output_dir / extraction_path.name), + "manifest": str(published_output_dir / manifest_path.name) if status == "prepared" else None, }, "compiler": { "validated": bundle is not None, @@ -767,11 +767,43 @@ def prepare(args: argparse.Namespace) -> dict[str, Any]: else "No novel claim was prepared; inspect duplicate judgments and do not stage an empty proposal." ), } - receipt["outputs"]["receipt"] = str(receipt_path) + receipt["outputs"]["receipt"] = str(published_output_dir / receipt_path.name) _write_private(receipt_path, _json_bytes(receipt)) return receipt +def _validate_output_target(path: Path) -> None: + if path.is_symlink(): + raise PreparationError("--output-dir must not be a symlink") + if path.exists() and (not path.is_dir() or any(path.iterdir())): + raise PreparationError("--output-dir must be absent or an empty dedicated directory") + + +def prepare(args: argparse.Namespace) -> dict[str, Any]: + requested_output_dir = args.output_dir.expanduser() + _validate_output_target(requested_output_dir) + output_dir = requested_output_dir.resolve() + output_dir.parent.mkdir(parents=True, exist_ok=True) + staging_dir = Path( + tempfile.mkdtemp( + prefix=f".{output_dir.name}.", + suffix=".staging", + dir=output_dir.parent, + ) + ) + os.chmod(staging_dir, 0o700) + staged_args = argparse.Namespace(**vars(args)) + staged_args.output_dir = staging_dir + try: + receipt = _prepare_staged(staged_args, output_dir) + _validate_output_target(output_dir) + os.replace(staging_dir, output_dir) + return receipt + finally: + if staging_dir.exists(): + shutil.rmtree(staging_dir) + + def parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--artifact", required=True, type=Path) @@ -797,11 +829,25 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace: return parser.parse_args(argv) +def _private_output_state(output_dir: Path) -> dict[str, bool]: + return { + "extracted_text": (output_dir / "extracted-text.txt").is_file(), + "extraction_result": (output_dir / "extraction-result.json").is_file(), + "manifest": (output_dir / "source-manifest.json").is_file(), + "receipt": (output_dir / "preparation-receipt.json").is_file(), + } + + def main(argv: list[str] | None = None) -> int: args = parse_args(argv) + initial_output_state = _private_output_state(args.output_dir) try: receipt = prepare(args) except (PreparationError, compiler.CompilerError, OSError, ValueError): + current_output_state = _private_output_state(args.output_dir) + written_output_state = { + name: present and not initial_output_state[name] for name, present in current_output_state.items() + } print( json.dumps( { @@ -809,7 +855,8 @@ def main(argv: list[str] | None = None) -> int: "status": "rejected", "database_write_performed": False, "canonical_apply_performed": False, - "private_outputs_written": False, + "private_outputs_written": written_output_state["receipt"], + "private_output_files": written_output_state, "reason": "source_preparation_failed", }, sort_keys=True, diff --git a/tests/test_hermes_leoclean_kb_bridge_source.py b/tests/test_hermes_leoclean_kb_bridge_source.py index 3f9a370..dbc0a5a 100644 --- a/tests/test_hermes_leoclean_kb_bridge_source.py +++ b/tests/test_hermes_leoclean_kb_bridge_source.py @@ -1715,6 +1715,69 @@ def test_vps_prepare_source_retrieves_canonical_candidates_before_model_extracti assert candidate_id not in stdout +@pytest.mark.parametrize( + ("failure_kind", "expected_error"), + [ + ("timeout", "source preparer timed out; private diagnostic detail redacted"), + ("launch", "source preparer could not start; private diagnostic detail redacted"), + ("crash", "source preparer failed; private diagnostic detail redacted"), + ], +) +def test_vps_prepare_source_wrapper_redacts_commands_and_child_output( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + failure_kind: str, + expected_error: str, +) -> None: + module = _load_module(BRIDGE_DIR / "kb_tool.py") + sentinel = "PREPARER-WRAPPER-PRIVATE-CANARY" + preparer = tmp_path / "prepare_kb_source_manifest.py" + preparer.write_text("# fixture\n", encoding="utf-8") + inbox = tmp_path / "source-inbox" + preparation_root = tmp_path / "source-preparation" + private_inbox = inbox / sentinel + private_inbox.mkdir(parents=True) + preparation_root.mkdir() + artifact = private_inbox / "source.md" + artifact.write_text("private", encoding="utf-8") + monkeypatch.setattr(module, "SOURCE_PREPARER_PATH", preparer) + monkeypatch.setattr(module, "SOURCE_INBOX_ROOT", inbox) + monkeypatch.setattr(module, "SOURCE_PREPARATION_ROOT", preparation_root) + monkeypatch.setattr(module, "find_claims", lambda *_args, **_kwargs: []) + + def fail(command, **_kwargs): + if failure_kind == "timeout": + raise subprocess.TimeoutExpired(command, 420, output=sentinel, stderr=sentinel) + if failure_kind == "launch": + raise OSError(f"could not launch {sentinel}") + return subprocess.CompletedProcess(command, 19, stdout="", stderr=sentinel) + + monkeypatch.setattr(module.subprocess, "run", fail) + args = SimpleNamespace( + local=True, + query=sentinel, + artifact=artifact, + text=None, + identity=f"document:{sentinel}", + source_key="private_source", + source_type="article", + title=sentinel, + locator=f"artifact://{sentinel}", + output_dir=preparation_root / "private-output", + model=sentinel, + ) + + with pytest.raises(SystemExit) as caught: + module.prepare_source(args) + + rendered = str(caught.value) + assert rendered == expected_error + assert sentinel not in rendered + assert str(artifact) not in rendered + assert args.identity not in rendered + assert args.locator not in rendered + + def test_vps_prepare_source_rejects_files_and_outputs_outside_private_roots(tmp_path: Path) -> None: module = _load_module(BRIDGE_DIR / "kb_tool.py") inbox = tmp_path / "source-inbox" diff --git a/tests/test_kb_tool_propose_source.py b/tests/test_kb_tool_propose_source.py index b2f83a3..33b51e6 100644 --- a/tests/test_kb_tool_propose_source.py +++ b/tests/test_kb_tool_propose_source.py @@ -5,6 +5,7 @@ import importlib.util import json import re import stat +import subprocess from pathlib import Path from types import SimpleNamespace @@ -219,6 +220,51 @@ def test_deployed_bridge_accepts_real_source_compiler_bundle(tmp_path: Path, mon assert validated_manifest["artifact_sha256"] == artifact_sha256 +@pytest.mark.parametrize( + ("failure_kind", "expected_error"), + [ + ("timeout", "source compiler timed out; private diagnostic detail redacted"), + ("launch", "source compiler could not start; private diagnostic detail redacted"), + ("crash", "source compiler failed; private diagnostic detail redacted"), + ], +) +def test_source_compiler_wrapper_redacts_commands_timeouts_and_child_output( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + failure_kind: str, + expected_error: str, +) -> None: + module = load_module() + sentinel = "COMPILER-WRAPPER-PRIVATE-CANARY" + compiler = tmp_path / "compiler.py" + compiler.write_text("# fixture\n", encoding="utf-8") + args = SimpleNamespace( + artifact=tmp_path / sentinel / "artifact.pdf", + text=tmp_path / sentinel / "extracted.txt", + manifest=tmp_path / sentinel / "manifest.json", + ) + monkeypatch.setattr(module, "SOURCE_COMPILER_PATH", compiler) + + def fail(command, **_kwargs): + if failure_kind == "timeout": + raise subprocess.TimeoutExpired(command, 120, output=sentinel, stderr=sentinel) + if failure_kind == "launch": + raise OSError(f"could not launch {sentinel}") + return subprocess.CompletedProcess(command, 17, stdout="", stderr=sentinel) + + monkeypatch.setattr(module.subprocess, "run", fail) + + with pytest.raises(SystemExit) as caught: + module.compile_source_bundle(args) + + rendered = str(caught.value) + assert rendered == expected_error + assert sentinel not in rendered + assert str(args.artifact) not in rendered + assert str(args.text) not in rendered + assert str(args.manifest) not in rendered + + def test_propose_source_refuses_nonlocal_database_route(tmp_path: Path) -> None: module = load_module() args = SimpleNamespace( diff --git a/tests/test_prepare_kb_context_from_canonical_dump.py b/tests/test_prepare_kb_context_from_canonical_dump.py index 09c1ab5..14cc0f9 100644 --- a/tests/test_prepare_kb_context_from_canonical_dump.py +++ b/tests/test_prepare_kb_context_from_canonical_dump.py @@ -125,6 +125,8 @@ def test_copy_parser_preserves_escaped_source_text() -> None: (r"caf\xc3\xa9", "caf\N{LATIN SMALL LETTER E WITH ACUTE}"), (r"price \342\202\2545", "price \N{EURO SIGN}5"), (r"literal\qescape", "literalqescape"), + (r"literal\x", "literalx"), + (r"literal\xg1", "literalxg1"), ], ) def test_copy_parser_decodes_octal_and_hex_as_encoded_bytes(encoded: str, decoded: str) -> None: @@ -137,12 +139,29 @@ def test_copy_parser_uses_explicit_encoding() -> None: context_builder._decode_copy_field(r"caf\351", encoding="utf-8") -@pytest.mark.parametrize("encoded", [r"\xff", r"\303", r"\400", r"\000", r"\x", "dangling\\"]) +@pytest.mark.parametrize("encoded", [r"\xff", r"\303", r"\400", r"\000", "dangling\\"]) def test_copy_parser_fails_closed_on_invalid_byte_escapes(encoded: str) -> None: with pytest.raises(context_builder.ContextError): context_builder._decode_copy_field(encoded) +@pytest.mark.parametrize("line_ending", ["\n", "\r\n"]) +def test_copy_parser_splits_only_copy_row_boundaries(line_ending: str) -> None: + source_text = "before\N{LINE SEPARATOR}middle\N{PARAGRAPH SEPARATOR}after" + sql = line_ending.join( + [ + "COPY public.claims (id, text) FROM stdin;", + f"{CLAIM_A}\t{source_text}", + r"\.", + "", + ] + ) + + rows = context_builder.parse_copy_table(sql, "claims") + + assert rows == [{"id": CLAIM_A, "text": source_text}] + + def test_private_json_is_owner_only_and_refuses_overwrite(tmp_path: Path) -> None: output = tmp_path / "private" / "context.json" @@ -230,3 +249,54 @@ def test_cli_redacts_private_failure_details_from_stdout( assert public_status["reason"] == "context_preparation_failed" assert sentinel not in stdout assert str(private_path) not in stdout + + +@pytest.mark.parametrize( + ("fail_on_replace", "expected_context", "expected_receipt"), + [(1, False, False), (2, True, False)], +) +def test_cli_reports_exact_state_after_each_context_publication_boundary( + tmp_path: Path, + capsys: pytest.CaptureFixture, + monkeypatch: pytest.MonkeyPatch, + fail_on_replace: int, + expected_context: bool, + expected_receipt: bool, +) -> None: + context_output = tmp_path / "private-output" / "context.json" + receipt_output = tmp_path / "private-output" / "receipt.json" + context = {"database_search_query": "private", "candidate_claims": []} + receipt = {"retrieval": {"candidate_count": 0}} + monkeypatch.setattr(context_builder, "build_context", lambda **_kwargs: (context, receipt)) + real_replace = context_builder.os.replace + replace_count = 0 + + def faulted_replace(source: Path, destination: Path) -> None: + nonlocal replace_count + replace_count += 1 + if replace_count == fail_on_replace: + raise OSError("publication fault detail must stay private") + real_replace(source, destination) + + monkeypatch.setattr(context_builder.os, "replace", faulted_replace) + status = context_builder.main( + [ + "--dump", + str(tmp_path / "dump"), + "--artifact", + str(tmp_path / "artifact"), + "--title", + "private", + "--context-output", + str(context_output), + "--receipt-output", + str(receipt_output), + ] + ) + + public_status = json.loads(capsys.readouterr().out) + assert status == 2 + assert public_status["private_context_written"] is expected_context + assert public_status["private_receipt_written"] is expected_receipt + assert context_output.is_file() is expected_context + assert receipt_output.is_file() is expected_receipt diff --git a/tests/test_prepare_kb_source_manifest.py b/tests/test_prepare_kb_source_manifest.py index 3a24148..4b2c7e4 100644 --- a/tests/test_prepare_kb_source_manifest.py +++ b/tests/test_prepare_kb_source_manifest.py @@ -494,3 +494,139 @@ def test_cli_redacts_private_preparation_failure_from_stdout( assert public_status["reason"] == "source_preparation_failed" assert sentinel not in stdout assert str(tmp_path) not in stdout + + +@pytest.mark.parametrize("fail_after_write", [1, 2, 3, 4]) +def test_prepare_never_publishes_partially_staged_output_set( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + fail_after_write: int, +) -> None: + artifact, context = _inputs(tmp_path) + args = _args(tmp_path, artifact, context) + monkeypatch.setattr( + preparer, + "call_openrouter", + lambda *_args, **_kwargs: (_model_output(), {"prompt_tokens": 100, "completion_tokens": 50}), + ) + real_write = preparer._write_private + write_count = 0 + + def faulted_write(path: Path, value: bytes) -> None: + nonlocal write_count + real_write(path, value) + write_count += 1 + if write_count == fail_after_write: + raise OSError("staging fault detail must stay private") + + monkeypatch.setattr(preparer, "_write_private", faulted_write) + + with pytest.raises(OSError): + preparer.prepare(args) + + assert not args.output_dir.exists() or not any(args.output_dir.iterdir()) + assert not list(args.output_dir.parent.glob(f".{args.output_dir.name}.*.staging")) + + +@pytest.mark.parametrize( + ("fault_after_publication", "expected_published"), + [(False, False), (True, True)], +) +def test_prepare_atomic_publication_boundary_is_all_or_nothing( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + fault_after_publication: bool, + expected_published: bool, +) -> None: + artifact, context = _inputs(tmp_path) + args = _args(tmp_path, artifact, context) + monkeypatch.setattr( + preparer, + "call_openrouter", + lambda *_args, **_kwargs: (_model_output(), {"prompt_tokens": 100, "completion_tokens": 50}), + ) + real_replace = preparer.os.replace + + def faulted_replace(source: Path, destination: Path) -> None: + if Path(destination) != args.output_dir: + real_replace(source, destination) + return + if fault_after_publication: + real_replace(source, destination) + raise OSError("publication fault detail must stay private") + + monkeypatch.setattr(preparer.os, "replace", faulted_replace) + + with pytest.raises(OSError): + preparer.prepare(args) + + expected_state = { + "extracted_text": expected_published, + "extraction_result": expected_published, + "manifest": expected_published, + "receipt": expected_published, + } + assert preparer._private_output_state(args.output_dir) == expected_state + assert not list(args.output_dir.parent.glob(f".{args.output_dir.name}.*.staging")) + + +@pytest.mark.parametrize( + "published_names", + [ + ("extracted-text.txt",), + ("extracted-text.txt", "extraction-result.json"), + ("extracted-text.txt", "extraction-result.json", "source-manifest.json"), + ( + "extracted-text.txt", + "extraction-result.json", + "source-manifest.json", + "preparation-receipt.json", + ), + ], +) +def test_cli_failure_reports_actual_state_after_each_private_publication_boundary( + tmp_path: Path, + capsys: pytest.CaptureFixture, + monkeypatch: pytest.MonkeyPatch, + published_names: tuple[str, ...], +) -> None: + output_dir = tmp_path / "private-output" + + def fail(args) -> None: + args.output_dir.mkdir(parents=True) + for name in published_names: + (args.output_dir / name).write_text("private", encoding="utf-8") + raise preparer.PreparationError("private publication fault") + + monkeypatch.setattr(preparer, "prepare", fail) + status = preparer.main( + [ + "--artifact", + str(tmp_path / "source.md"), + "--kb-context", + str(tmp_path / "context.json"), + "--identity", + "document:private", + "--source-key", + "private_source", + "--source-type", + "article", + "--title", + "private", + "--locator", + "artifact://private", + "--output-dir", + str(output_dir), + ] + ) + + public_status = json.loads(capsys.readouterr().out) + expected_state = { + "extracted_text": "extracted-text.txt" in published_names, + "extraction_result": "extraction-result.json" in published_names, + "manifest": "source-manifest.json" in published_names, + "receipt": "preparation-receipt.json" in published_names, + } + assert status == 2 + assert public_status["private_output_files"] == expected_state + assert public_status["private_outputs_written"] is expected_state["receipt"]