Harden private source reconstruction output

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,6 +2364,10 @@ 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}")
+    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),
@@ -2371,6 +2377,8 @@ def compile_source_bundle(args: argparse.Namespace) -> dict[str, Any]:
             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)
@@ -2380,9 +2388,22 @@ def compile_source_bundle(args: argparse.Namespace) -> dict[str, Any]:
             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)
+            status = json.loads(completed.stdout)
         except json.JSONDecodeError as exc:
-        raise SystemExit(f"source compiler emitted invalid JSON: {exc}") from 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"]
+    print("# Source Preparation Status\n")
-    extraction = data["extraction"]
-    outputs = data["outputs"]
-    print("# Source Preparation Receipt\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("- private outputs written: `true`; inspect the requested output directory locally")
-    print(f"- manifest: `{outputs['manifest'] or '-'}`")
-    print(f"- receipt: `{outputs['receipt']}`")
-    print(f"- next: {data['next_action']}")
 
 
 def main() -> int:

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")
         bundle = compile_source_packet(args.artifact, args.text, args.manifest)
-    except (CompilerError, OSError, ValueError) as exc:
+    except (CompilerError, OSError, ValueError):
-        print(
+        _print_status("rejected", reason="input_validation_failed")
-            json.dumps(
-                {
-                    "schema": BUNDLE_SCHEMA,
-                    "status": "rejected",
-                    "database_write_performed": False,
-                    "error": str(exc),
-                },
-                sort_keys=True,
-            )
-        )
         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:
+    except OSError:
-            print(
+        _print_status("rejected", reason="output_write_failed")
-                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="")
+    _print_status("pending_review")
     return 0
 
 

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):
+        if char != "\\":
-            result.append(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:
+            if match is None:
-                result.append(chr(int(match.group(0), 16)))
+                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:
+            if match is None:
-                result.append(chr(int(match.group(0), 8)))
+                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
-        result.append(escaped)
+        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)
+        completed = subprocess.run(
-    except (OSError, subprocess.TimeoutExpired) as exc:
+            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"],
+                "model_call_performed": False,
-                "context_sha256": sha256_bytes(json.dumps(context, sort_keys=True).encode("utf-8")),
+                "private_context_written": True,
-                "outputs": receipt["outputs"],
+                "private_receipt_written": True,
             },
             indent=2,
             sort_keys=True,

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
 
 

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:
+def test_cli_writes_private_bundle_and_emits_only_public_status_bytes(tmp_path: Path) -> None:
-    artifact, text, manifest_path = _write_inputs(tmp_path)
+    private_root = tmp_path / "private-source-sentinel"
-    output = tmp_path / "compiled" / "packet.json"
+    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
+    stdout = completed.stdout
-    assert status == 0
+    public_status = json.loads(stdout.decode("utf-8", errors="strict"))
-    assert output.read_text(encoding="utf-8") == stdout
+    bundle = json.loads(output.read_text(encoding="utf-8"))
-    assert json.loads(stdout)["status"] == "pending_review"
+    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:

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:

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

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