From 9a9594e41b029bbdf6d64779d9388c1cd2e0b8af Mon Sep 17 00:00:00 2001 From: twentyOne2x Date: Thu, 16 Jul 2026 12:24:20 +0200 Subject: [PATCH] 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