Harden private source reconstruction failures

hermes-agent/leoclean-bin/kb_tool.py

@@ -2382,11 +2382,12 @@ def compile_source_bundle(args: argparse.Namespace) -> dict[str, Any]:
         ]
         try:
             completed = subprocess.run(command, capture_output=True, text=True, timeout=120, check=False)
-        except (OSError, subprocess.TimeoutExpired) as exc:
+        except subprocess.TimeoutExpired:
-            raise SystemExit(f"source compiler could not run: {exc}") from exc
+            raise SystemExit("source compiler timed out; private diagnostic detail redacted") from None
+        except OSError:
+            raise SystemExit("source compiler could not start; private diagnostic detail redacted") from None
         if completed.returncode != 0:
-            detail = completed.stdout.strip() or completed.stderr.strip() or f"exit {completed.returncode}"
+            raise SystemExit("source compiler failed; private diagnostic detail redacted")
-            raise SystemExit(f"source compiler rejected the artifact: {detail}")
         try:
             status = json.loads(completed.stdout)
         except json.JSONDecodeError as exc:
@@ -2610,15 +2611,16 @@ def prepare_source(args: argparse.Namespace) -> dict[str, Any]:
             command.extend(("--text", str(args.text)))
         try:
             completed = subprocess.run(command, capture_output=True, text=True, timeout=420, check=False)
-        except (OSError, subprocess.TimeoutExpired) as exc:
+        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:
         status = json.loads(completed.stdout)
     except json.JSONDecodeError as exc:

scripts/prepare_kb_context_from_canonical_dump.py

@@ -92,8 +92,7 @@ def _decode_copy_field(value: str, *, encoding: str = DEFAULT_COPY_ENCODING) ->
             continue
         if escaped == "x":
             match = re.match(r"[0-9A-Fa-f]{1,2}", value[index + 1 :])
-            if match is None:
+            if match is not None:
-                raise ContextError("COPY field contains a malformed hexadecimal byte escape")
                 result.append(int(match.group(0), 16))
                 index += 1 + len(match.group(0))
                 continue
@@ -122,7 +121,7 @@ def parse_copy_table(
 ) -> list[dict[str, str | None]]:
     encoding = _copy_encoding(encoding)
     prefix = f"COPY public.{table} ("
-    lines = iter(sql.splitlines())
+    lines = iter(line[:-1] if line.endswith("\r") else line for line in sql.split("\n"))
     for line in lines:
         if line.startswith(prefix) and line.endswith(") FROM stdin;"):
             columns = [item.strip() for item in line[len(prefix) : -len(") FROM stdin;")].split(",")]
@@ -296,9 +295,7 @@ def build_context(
     return context, receipt
 
 
-def _private_json(path: Path, value: Any) -> None:
+def _stage_private_json(path: Path, value: Any) -> Path:
-    if path.exists():
-        raise ContextError(f"refusing to overwrite existing output: {path}")
     path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
     os.chmod(path.parent, 0o700)
     fd, raw_temp = tempfile.mkstemp(prefix=f".{path.name}.", suffix=".tmp", dir=path.parent)
@@ -310,6 +307,18 @@ def _private_json(path: Path, value: Any) -> None:
             handle.write("\n")
             handle.flush()
             os.fsync(handle.fileno())
+        return temp
+    except BaseException:
+        if temp.exists():
+            temp.unlink()
+        raise
+
+
+def _private_json(path: Path, value: Any) -> None:
+    if path.exists():
+        raise ContextError(f"refusing to overwrite existing output: {path}")
+    temp = _stage_private_json(path, value)
+    try:
         os.replace(temp, path)
         os.chmod(path, 0o600)
     finally:
@@ -317,6 +326,24 @@ def _private_json(path: Path, value: Any) -> None:
             temp.unlink()
 
 
+def _publish_private_json_outputs(outputs: tuple[tuple[Path, Any], ...]) -> None:
+    for path, _value in outputs:
+        if path.exists():
+            raise ContextError(f"refusing to overwrite existing output: {path}")
+
+    staged: list[tuple[Path, Path]] = []
+    try:
+        for path, value in outputs:
+            staged.append((path, _stage_private_json(path, value)))
+        for path, temp in staged:
+            os.replace(temp, path)
+            os.chmod(path, 0o600)
+    finally:
+        for _path, temp in staged:
+            if temp.exists():
+                temp.unlink()
+
+
 def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument("--dump", required=True, type=Path)
@@ -333,6 +360,8 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
 def main(argv: list[str] | None = None) -> int:
     args = parse_args(argv)
     repo_root = Path(__file__).resolve().parents[1]
+    context_existed = args.context_output.is_file()
+    receipt_existed = args.receipt_output.is_file()
     try:
         if args.context_output.resolve() == args.receipt_output.resolve():
             raise ContextError("context and receipt outputs must be different files")
@@ -345,13 +374,16 @@ def main(argv: list[str] | None = None) -> int:
             repo_root=repo_root,
             copy_encoding=args.copy_encoding,
         )
-        _private_json(args.context_output, context)
         receipt["outputs"] = {
             "context": str(args.context_output.resolve()),
             "receipt": str(args.receipt_output.resolve()),
         }
-        _private_json(args.receipt_output, receipt)
+        _publish_private_json_outputs(
+            ((args.context_output, context), (args.receipt_output, receipt))
+        )
     except (ContextError, OSError, ValueError):
+        context_written = not context_existed and args.context_output.is_file()
+        receipt_written = not receipt_existed and args.receipt_output.is_file()
         print(
             json.dumps(
                 {
@@ -359,8 +391,8 @@ def main(argv: list[str] | None = None) -> int:
                     "status": "rejected",
                     "database_write_performed": False,
                     "model_call_performed": False,
-                    "private_context_written": False,
+                    "private_context_written": context_written,
-                    "private_receipt_written": False,
+                    "private_receipt_written": receipt_written,
                     "reason": "context_preparation_failed",
                 },
                 sort_keys=True,

scripts/prepare_kb_source_manifest.py

@@ -594,7 +594,7 @@ def _validate_metadata(args: argparse.Namespace) -> None:
         raise PreparationError(f"--source-type must be one of {sorted(compiler.ap.SOURCE_TYPES)}")
 
 
-def prepare(args: argparse.Namespace) -> dict[str, Any]:
+def _prepare_staged(args: argparse.Namespace, published_output_dir: Path) -> dict[str, Any]:
     _validate_metadata(args)
     artifact_bytes = _read_nonempty(args.artifact, "artifact")
     text_bytes = extract_utf8_text(args.artifact, args.text)
@@ -751,10 +751,10 @@ def prepare(args: argparse.Namespace) -> dict[str, Any]:
             "exact_selected_locations_validated": bundle is not None,
         },
         "outputs": {
-            "output_dir": str(output_dir),
+            "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,
@@ -767,11 +767,43 @@ def prepare(args: argparse.Namespace) -> dict[str, Any]:
             else "No novel claim was prepared; inspect duplicate judgments and do not stage an empty proposal."
         ),
     }
-    receipt["outputs"]["receipt"] = str(receipt_path)
+    receipt["outputs"]["receipt"] = str(published_output_dir / receipt_path.name)
     _write_private(receipt_path, _json_bytes(receipt))
     return receipt
 
 
+def _validate_output_target(path: Path) -> None:
+    if path.is_symlink():
+        raise PreparationError("--output-dir must not be a symlink")
+    if path.exists() and (not path.is_dir() or any(path.iterdir())):
+        raise PreparationError("--output-dir must be absent or an empty dedicated directory")
+
+
+def prepare(args: argparse.Namespace) -> dict[str, Any]:
+    requested_output_dir = args.output_dir.expanduser()
+    _validate_output_target(requested_output_dir)
+    output_dir = requested_output_dir.resolve()
+    output_dir.parent.mkdir(parents=True, exist_ok=True)
+    staging_dir = Path(
+        tempfile.mkdtemp(
+            prefix=f".{output_dir.name}.",
+            suffix=".staging",
+            dir=output_dir.parent,
+        )
+    )
+    os.chmod(staging_dir, 0o700)
+    staged_args = argparse.Namespace(**vars(args))
+    staged_args.output_dir = staging_dir
+    try:
+        receipt = _prepare_staged(staged_args, output_dir)
+        _validate_output_target(output_dir)
+        os.replace(staging_dir, output_dir)
+        return receipt
+    finally:
+        if staging_dir.exists():
+            shutil.rmtree(staging_dir)
+
+
 def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument("--artifact", required=True, type=Path)
@@ -797,11 +829,25 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
     return parser.parse_args(argv)
 
 
+def _private_output_state(output_dir: Path) -> dict[str, bool]:
+    return {
+        "extracted_text": (output_dir / "extracted-text.txt").is_file(),
+        "extraction_result": (output_dir / "extraction-result.json").is_file(),
+        "manifest": (output_dir / "source-manifest.json").is_file(),
+        "receipt": (output_dir / "preparation-receipt.json").is_file(),
+    }
+
+
 def main(argv: list[str] | None = None) -> int:
     args = parse_args(argv)
+    initial_output_state = _private_output_state(args.output_dir)
     try:
         receipt = prepare(args)
     except (PreparationError, compiler.CompilerError, OSError, ValueError):
+        current_output_state = _private_output_state(args.output_dir)
+        written_output_state = {
+            name: present and not initial_output_state[name] for name, present in current_output_state.items()
+        }
         print(
             json.dumps(
                 {
@@ -809,7 +855,8 @@ def main(argv: list[str] | None = None) -> int:
                     "status": "rejected",
                     "database_write_performed": False,
                     "canonical_apply_performed": False,
-                    "private_outputs_written": False,
+                    "private_outputs_written": written_output_state["receipt"],
+                    "private_output_files": written_output_state,
                     "reason": "source_preparation_failed",
                 },
                 sort_keys=True,

tests/test_hermes_leoclean_kb_bridge_source.py

@@ -1715,6 +1715,69 @@ def test_vps_prepare_source_retrieves_canonical_candidates_before_model_extracti
     assert candidate_id not in stdout
 
 
+@pytest.mark.parametrize(
+    ("failure_kind", "expected_error"),
+    [
+        ("timeout", "source preparer timed out; private diagnostic detail redacted"),
+        ("launch", "source preparer could not start; private diagnostic detail redacted"),
+        ("crash", "source preparer failed; private diagnostic detail redacted"),
+    ],
+)
+def test_vps_prepare_source_wrapper_redacts_commands_and_child_output(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    failure_kind: str,
+    expected_error: str,
+) -> None:
+    module = _load_module(BRIDGE_DIR / "kb_tool.py")
+    sentinel = "PREPARER-WRAPPER-PRIVATE-CANARY"
+    preparer = tmp_path / "prepare_kb_source_manifest.py"
+    preparer.write_text("# fixture\n", encoding="utf-8")
+    inbox = tmp_path / "source-inbox"
+    preparation_root = tmp_path / "source-preparation"
+    private_inbox = inbox / sentinel
+    private_inbox.mkdir(parents=True)
+    preparation_root.mkdir()
+    artifact = private_inbox / "source.md"
+    artifact.write_text("private", encoding="utf-8")
+    monkeypatch.setattr(module, "SOURCE_PREPARER_PATH", preparer)
+    monkeypatch.setattr(module, "SOURCE_INBOX_ROOT", inbox)
+    monkeypatch.setattr(module, "SOURCE_PREPARATION_ROOT", preparation_root)
+    monkeypatch.setattr(module, "find_claims", lambda *_args, **_kwargs: [])
+
+    def fail(command, **_kwargs):
+        if failure_kind == "timeout":
+            raise subprocess.TimeoutExpired(command, 420, output=sentinel, stderr=sentinel)
+        if failure_kind == "launch":
+            raise OSError(f"could not launch {sentinel}")
+        return subprocess.CompletedProcess(command, 19, stdout="", stderr=sentinel)
+
+    monkeypatch.setattr(module.subprocess, "run", fail)
+    args = SimpleNamespace(
+        local=True,
+        query=sentinel,
+        artifact=artifact,
+        text=None,
+        identity=f"document:{sentinel}",
+        source_key="private_source",
+        source_type="article",
+        title=sentinel,
+        locator=f"artifact://{sentinel}",
+        output_dir=preparation_root / "private-output",
+        model=sentinel,
+    )
+
+    with pytest.raises(SystemExit) as caught:
+        module.prepare_source(args)
+
+    rendered = str(caught.value)
+    assert rendered == expected_error
+    assert sentinel not in rendered
+    assert str(artifact) not in rendered
+    assert args.identity not in rendered
+    assert args.locator not in rendered
+
+
 def test_vps_prepare_source_rejects_files_and_outputs_outside_private_roots(tmp_path: Path) -> None:
     module = _load_module(BRIDGE_DIR / "kb_tool.py")
     inbox = tmp_path / "source-inbox"

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

@@ -125,6 +125,8 @@ def test_copy_parser_preserves_escaped_source_text() -> None:
         (r"caf\xc3\xa9", "caf\N{LATIN SMALL LETTER E WITH ACUTE}"),
         (r"price \342\202\2545", "price \N{EURO SIGN}5"),
         (r"literal\qescape", "literalqescape"),
+        (r"literal\x", "literalx"),
+        (r"literal\xg1", "literalxg1"),
     ],
 )
 def test_copy_parser_decodes_octal_and_hex_as_encoded_bytes(encoded: str, decoded: str) -> None:
@@ -137,12 +139,29 @@ def test_copy_parser_uses_explicit_encoding() -> None:
         context_builder._decode_copy_field(r"caf\351", encoding="utf-8")
 
 
-@pytest.mark.parametrize("encoded", [r"\xff", r"\303", r"\400", r"\000", r"\x", "dangling\\"])
+@pytest.mark.parametrize("encoded", [r"\xff", r"\303", r"\400", r"\000", "dangling\\"])
 def test_copy_parser_fails_closed_on_invalid_byte_escapes(encoded: str) -> None:
     with pytest.raises(context_builder.ContextError):
         context_builder._decode_copy_field(encoded)
 
 
+@pytest.mark.parametrize("line_ending", ["\n", "\r\n"])
+def test_copy_parser_splits_only_copy_row_boundaries(line_ending: str) -> None:
+    source_text = "before\N{LINE SEPARATOR}middle\N{PARAGRAPH SEPARATOR}after"
+    sql = line_ending.join(
+        [
+            "COPY public.claims (id, text) FROM stdin;",
+            f"{CLAIM_A}\t{source_text}",
+            r"\.",
+            "",
+        ]
+    )
+
+    rows = context_builder.parse_copy_table(sql, "claims")
+
+    assert rows == [{"id": CLAIM_A, "text": source_text}]
+
+
 def test_private_json_is_owner_only_and_refuses_overwrite(tmp_path: Path) -> None:
     output = tmp_path / "private" / "context.json"
 
@@ -230,3 +249,54 @@ def test_cli_redacts_private_failure_details_from_stdout(
     assert public_status["reason"] == "context_preparation_failed"
     assert sentinel not in stdout
     assert str(private_path) not in stdout
+
+
+@pytest.mark.parametrize(
+    ("fail_on_replace", "expected_context", "expected_receipt"),
+    [(1, False, False), (2, True, False)],
+)
+def test_cli_reports_exact_state_after_each_context_publication_boundary(
+    tmp_path: Path,
+    capsys: pytest.CaptureFixture,
+    monkeypatch: pytest.MonkeyPatch,
+    fail_on_replace: int,
+    expected_context: bool,
+    expected_receipt: bool,
+) -> None:
+    context_output = tmp_path / "private-output" / "context.json"
+    receipt_output = tmp_path / "private-output" / "receipt.json"
+    context = {"database_search_query": "private", "candidate_claims": []}
+    receipt = {"retrieval": {"candidate_count": 0}}
+    monkeypatch.setattr(context_builder, "build_context", lambda **_kwargs: (context, receipt))
+    real_replace = context_builder.os.replace
+    replace_count = 0
+
+    def faulted_replace(source: Path, destination: Path) -> None:
+        nonlocal replace_count
+        replace_count += 1
+        if replace_count == fail_on_replace:
+            raise OSError("publication fault detail must stay private")
+        real_replace(source, destination)
+
+    monkeypatch.setattr(context_builder.os, "replace", faulted_replace)
+    status = context_builder.main(
+        [
+            "--dump",
+            str(tmp_path / "dump"),
+            "--artifact",
+            str(tmp_path / "artifact"),
+            "--title",
+            "private",
+            "--context-output",
+            str(context_output),
+            "--receipt-output",
+            str(receipt_output),
+        ]
+    )
+
+    public_status = json.loads(capsys.readouterr().out)
+    assert status == 2
+    assert public_status["private_context_written"] is expected_context
+    assert public_status["private_receipt_written"] is expected_receipt
+    assert context_output.is_file() is expected_context
+    assert receipt_output.is_file() is expected_receipt

tests/test_prepare_kb_source_manifest.py

@@ -494,3 +494,139 @@ def test_cli_redacts_private_preparation_failure_from_stdout(
     assert public_status["reason"] == "source_preparation_failed"
     assert sentinel not in stdout
     assert str(tmp_path) not in stdout
+
+
+@pytest.mark.parametrize("fail_after_write", [1, 2, 3, 4])
+def test_prepare_never_publishes_partially_staged_output_set(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    fail_after_write: int,
+) -> None:
+    artifact, context = _inputs(tmp_path)
+    args = _args(tmp_path, artifact, context)
+    monkeypatch.setattr(
+        preparer,
+        "call_openrouter",
+        lambda *_args, **_kwargs: (_model_output(), {"prompt_tokens": 100, "completion_tokens": 50}),
+    )
+    real_write = preparer._write_private
+    write_count = 0
+
+    def faulted_write(path: Path, value: bytes) -> None:
+        nonlocal write_count
+        real_write(path, value)
+        write_count += 1
+        if write_count == fail_after_write:
+            raise OSError("staging fault detail must stay private")
+
+    monkeypatch.setattr(preparer, "_write_private", faulted_write)
+
+    with pytest.raises(OSError):
+        preparer.prepare(args)
+
+    assert not args.output_dir.exists() or not any(args.output_dir.iterdir())
+    assert not list(args.output_dir.parent.glob(f".{args.output_dir.name}.*.staging"))
+
+
+@pytest.mark.parametrize(
+    ("fault_after_publication", "expected_published"),
+    [(False, False), (True, True)],
+)
+def test_prepare_atomic_publication_boundary_is_all_or_nothing(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    fault_after_publication: bool,
+    expected_published: bool,
+) -> None:
+    artifact, context = _inputs(tmp_path)
+    args = _args(tmp_path, artifact, context)
+    monkeypatch.setattr(
+        preparer,
+        "call_openrouter",
+        lambda *_args, **_kwargs: (_model_output(), {"prompt_tokens": 100, "completion_tokens": 50}),
+    )
+    real_replace = preparer.os.replace
+
+    def faulted_replace(source: Path, destination: Path) -> None:
+        if Path(destination) != args.output_dir:
+            real_replace(source, destination)
+            return
+        if fault_after_publication:
+            real_replace(source, destination)
+        raise OSError("publication fault detail must stay private")
+
+    monkeypatch.setattr(preparer.os, "replace", faulted_replace)
+
+    with pytest.raises(OSError):
+        preparer.prepare(args)
+
+    expected_state = {
+        "extracted_text": expected_published,
+        "extraction_result": expected_published,
+        "manifest": expected_published,
+        "receipt": expected_published,
+    }
+    assert preparer._private_output_state(args.output_dir) == expected_state
+    assert not list(args.output_dir.parent.glob(f".{args.output_dir.name}.*.staging"))
+
+
+@pytest.mark.parametrize(
+    "published_names",
+    [
+        ("extracted-text.txt",),
+        ("extracted-text.txt", "extraction-result.json"),
+        ("extracted-text.txt", "extraction-result.json", "source-manifest.json"),
+        (
+            "extracted-text.txt",
+            "extraction-result.json",
+            "source-manifest.json",
+            "preparation-receipt.json",
+        ),
+    ],
+)
+def test_cli_failure_reports_actual_state_after_each_private_publication_boundary(
+    tmp_path: Path,
+    capsys: pytest.CaptureFixture,
+    monkeypatch: pytest.MonkeyPatch,
+    published_names: tuple[str, ...],
+) -> None:
+    output_dir = tmp_path / "private-output"
+
+    def fail(args) -> None:
+        args.output_dir.mkdir(parents=True)
+        for name in published_names:
+            (args.output_dir / name).write_text("private", encoding="utf-8")
+        raise preparer.PreparationError("private publication fault")
+
+    monkeypatch.setattr(preparer, "prepare", fail)
+    status = preparer.main(
+        [
+            "--artifact",
+            str(tmp_path / "source.md"),
+            "--kb-context",
+            str(tmp_path / "context.json"),
+            "--identity",
+            "document:private",
+            "--source-key",
+            "private_source",
+            "--source-type",
+            "article",
+            "--title",
+            "private",
+            "--locator",
+            "artifact://private",
+            "--output-dir",
+            str(output_dir),
+        ]
+    )
+
+    public_status = json.loads(capsys.readouterr().out)
+    expected_state = {
+        "extracted_text": "extracted-text.txt" in published_names,
+        "extraction_result": "extraction-result.json" in published_names,
+        "manifest": "source-manifest.json" in published_names,
+        "receipt": "preparation-receipt.json" in published_names,
+    }
+    assert status == 2
+    assert public_status["private_output_files"] == expected_state
+    assert public_status["private_outputs_written"] is expected_state["receipt"]