Merge remote-tracking branch 'origin/main' into codex/leo-vps-gcp-drift-w2-20260715

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,47 @@ 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 = [
+    with tempfile.TemporaryDirectory(prefix="teleo-source-compiler-") as raw_private_dir:
-        sys.executable,
+        private_dir = Path(raw_private_dir)
-        str(SOURCE_COMPILER_PATH),
+        os.chmod(private_dir, 0o700)
-        "--artifact",
+        bundle_path = private_dir / "source-bundle.json"
-        str(args.artifact),
+        command = [
-        "--text",
+            sys.executable,
-        str(args.text),
+            str(SOURCE_COMPILER_PATH),
-        "--manifest",
+            "--artifact",
-        str(args.manifest),
+            str(args.artifact),
-    ]
+            "--text",
-    try:
+            str(args.text),
-        completed = subprocess.run(command, capture_output=True, text=True, timeout=120, check=False)
+            "--manifest",
-    except (OSError, subprocess.TimeoutExpired) as exc:
+            str(args.manifest),
-        raise SystemExit(f"source compiler could not run: {exc}") from exc
+            "--output",
-    if completed.returncode != 0:
+            str(bundle_path),
-        detail = completed.stdout.strip() or completed.stderr.strip() or f"exit {completed.returncode}"
+        ]
-        raise SystemExit(f"source compiler rejected the artifact: {detail}")
+        try:
-    try:
+            completed = subprocess.run(command, capture_output=True, text=True, timeout=120, check=False)
-        bundle = json.loads(completed.stdout)
+        except subprocess.TimeoutExpired:
-    except json.JSONDecodeError as exc:
+            raise SystemExit("source compiler timed out; private diagnostic detail redacted") from None
-        raise SystemExit(f"source compiler emitted invalid JSON: {exc}") from exc
+        except OSError:
+            raise SystemExit("source compiler could not start; private diagnostic detail redacted") from None
+        if completed.returncode != 0:
+            raise SystemExit("source compiler failed; private diagnostic detail redacted")
+        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
 
@@ -2589,23 +2611,38 @@ 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:
+        except subprocess.TimeoutExpired:
-            raise SystemExit(f"source preparer could not run: {exc}") from exc
+            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("source preparer failed; private diagnostic detail redacted")
-        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 +2651,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 +3580,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

@@ -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
@@ -30,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)
@@ -570,52 +573,59 @@ 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)
 
 
+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 _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()
-            output_path = args.output.resolve()
+        input_paths = {args.artifact.resolve(), args.text.resolve(), args.manifest.resolve()}
-            input_paths = {args.artifact.resolve(), args.text.resolve(), args.manifest.resolve()}
+        if output_path in input_paths:
-            if output_path in input_paths:
+            raise CompilerError("--output must not overwrite an input file")
-                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:
-        try:
+        _write_private_output(args.output, rendered)
-            args.output.parent.mkdir(parents=True, exist_ok=True)
+    except OSError:
-            args.output.write_text(rendered, encoding="utf-8")
+        _print_status("rejected", reason="output_write_failed")
-        except OSError as exc:
+        return 2
-            print(
+    _print_status("pending_review")
-                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="")
     return 0
 
 

scripts/prepare_kb_context_from_canonical_dump.py

@@ -0,0 +1,420 @@
+#!/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 codecs
+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"
+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")
+
+
+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 _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
+    encoding = _copy_encoding(encoding)
+    result = bytearray()
+    index = 0
+    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 != "\\":
+            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:
+            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 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:
+                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
+    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, *, encoding: str = DEFAULT_COPY_ENCODING
+) -> list[dict[str, str | None]]:
+    encoding = _copy_encoding(encoding)
+    prefix = f"COPY public.{table} ("
+    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(",")]
+            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, 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, *, encoding: str = DEFAULT_COPY_ENCODING
+) -> 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",
+    ]
+    encoding = _copy_encoding(encoding)
+    try:
+        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")
+    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,
+    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")
+    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
+
+    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"]:
+        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,
+            "copy_encoding": copy_encoding,
+            "copy_decode_errors": "strict",
+        },
+    }
+    return context, receipt
+
+
+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)
+    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())
+        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:
+        if temp.exists():
+            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)
+    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)
+    parser.add_argument("--copy-encoding", default=DEFAULT_COPY_ENCODING)
+    return parser.parse_args(argv)
+
+
+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")
+        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,
+            copy_encoding=args.copy_encoding,
+        )
+        receipt["outputs"] = {
+            "context": str(args.context_output.resolve()),
+            "receipt": str(args.receipt_output.resolve()),
+        }
+        _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(
+                {
+                    "schema": STATUS_SCHEMA,
+                    "status": "rejected",
+                    "database_write_performed": False,
+                    "model_call_performed": False,
+                    "private_context_written": context_written,
+                    "private_receipt_written": receipt_written,
+                    "reason": "context_preparation_failed",
+                },
+                sort_keys=True,
+            )
+        )
+        return 2
+    print(
+        json.dumps(
+            {
+                "schema": STATUS_SCHEMA,
+                "status": "prepared",
+                "database_write_performed": False,
+                "model_call_performed": False,
+                "private_context_written": True,
+                "private_receipt_written": True,
+            },
+            indent=2,
+            sort_keys=True,
+        )
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

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
@@ -28,12 +30,15 @@ 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"
 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 +253,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")
+        exc.read(600)
-        raise PreparationError(f"OpenRouter returned HTTP {exc.code}: {detail}") from exc
+        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 +268,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 +546,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,
@@ -495,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)
@@ -526,14 +625,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,
+            args,
-            key_file=args.key_file,
+            max_budget_usd=remaining_budget_usd,
-            url=args.openrouter_url,
-            timeout_seconds=args.timeout_seconds,
         )
+        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 +724,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,14 +747,14 @@ 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": {
-            "output_dir": str(output_dir),
+            "output_dir": str(published_output_dir),
-            "extracted_text": str(text_path),
+            "extracted_text": str(published_output_dir / text_path.name),
-            "extraction_result": str(extraction_path),
+            "extraction_result": str(published_output_dir / extraction_path.name),
-            "manifest": str(manifest_path) if status == "prepared" else None,
+            "manifest": str(published_output_dir / manifest_path.name) if status == "prepared" else None,
         },
         "compiler": {
             "validated": bundle is not None,
@@ -651,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)
@@ -668,31 +816,65 @@ 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)
     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) as exc:
+    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(
                 {
-                    "schema": RECEIPT_SCHEMA,
+                    "schema": STATUS_SCHEMA,
                     "status": "rejected",
                     "database_write_performed": False,
                     "canonical_apply_performed": False,
-                    "error": str(exc),
+                    "private_outputs_written": written_output_state["receipt"],
+                    "private_output_files": written_output_state,
+                    "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

@@ -3,6 +3,8 @@ from __future__ import annotations
 import copy
 import hashlib
 import json
+import stat
+import subprocess
 import sys
 from pathlib import Path
 
@@ -207,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",
@@ -221,13 +226,34 @@ 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
 
 
 @pytest.mark.parametrize(
@@ -355,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(
@@ -363,14 +390,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):
+    def fail_write(_path: Path, _rendered: str) -> None:
-        if path == output:
+        raise PermissionError("fixture output denied")
-            raise PermissionError("fixture output denied")
-        return original_write_text(path, data, *args, **kwargs)
 
-    monkeypatch.setattr(Path, "write_text", guarded_write_text)
+    monkeypatch.setattr(compiler, "_write_private_output", fail_write)
     status = compiler.main(
         [
             "--artifact",
@@ -388,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

@@ -1623,7 +1623,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")
@@ -1658,16 +1660,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(
@@ -1690,6 +1705,77 @@ 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
+
+
+@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:

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(

tests/test_prepare_kb_context_from_canonical_dump.py

@@ -0,0 +1,302 @@
+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, **_kwargs: 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
+    assert receipt["runtime"]["copy_encoding"] == "utf-8"
+    assert receipt["runtime"]["copy_decode_errors"] == "strict"
+
+
+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"}]
+
+
+@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"),
+        (r"literal\x", "literalx"),
+        (r"literal\xg1", "literalxg1"),
+    ],
+)
+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", "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"
+
+    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})
+
+
+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
+
+
+@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

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,355 @@ 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
+
+
+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
+
+
+@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"]