From d3b151feb75dc00b9076ed25aa5b3029dec09b89 Mon Sep 17 00:00:00 2001 From: twentyOne2x Date: Thu, 16 Jul 2026 12:47:52 +0200 Subject: [PATCH] Harden private source reconstruction failures --- hermes-agent/leoclean-bin/kb_tool.py | 18 +-- .../prepare_kb_context_from_canonical_dump.py | 58 ++++++-- scripts/prepare_kb_source_manifest.py | 61 +++++++- .../test_hermes_leoclean_kb_bridge_source.py | 63 ++++++++ tests/test_kb_tool_propose_source.py | 46 ++++++ ..._prepare_kb_context_from_canonical_dump.py | 72 +++++++++- tests/test_prepare_kb_source_manifest.py | 136 ++++++++++++++++++ 7 files changed, 425 insertions(+), 29 deletions(-) diff --git a/hermes-agent/leoclean-bin/kb_tool.py b/hermes-agent/leoclean-bin/kb_tool.py index f01a3e1..0557d1d 100755 --- a/hermes-agent/leoclean-bin/kb_tool.py +++ b/hermes-agent/leoclean-bin/kb_tool.py @@ -2382,11 +2382,12 @@ def compile_source_bundle(args: argparse.Namespace) -> dict[str, Any]: ] try: completed = subprocess.run(command, capture_output=True, text=True, timeout=120, check=False) - except (OSError, subprocess.TimeoutExpired) as exc: - raise SystemExit(f"source compiler could not run: {exc}") from exc + except subprocess.TimeoutExpired: + raise SystemExit("source compiler timed out; private diagnostic detail redacted") from None + except OSError: + raise SystemExit("source compiler could not start; private diagnostic detail redacted") from None if completed.returncode != 0: - detail = completed.stdout.strip() or completed.stderr.strip() or f"exit {completed.returncode}" - raise SystemExit(f"source compiler rejected the artifact: {detail}") + raise SystemExit("source compiler failed; private diagnostic detail redacted") try: status = json.loads(completed.stdout) except json.JSONDecodeError as exc: @@ -2610,15 +2611,16 @@ def prepare_source(args: argparse.Namespace) -> dict[str, Any]: command.extend(("--text", str(args.text))) try: completed = subprocess.run(command, capture_output=True, text=True, timeout=420, check=False) - except (OSError, subprocess.TimeoutExpired) as exc: - raise SystemExit(f"source preparer could not run: {exc}") from exc + except subprocess.TimeoutExpired: + raise SystemExit("source preparer timed out; private diagnostic detail redacted") from None + except OSError: + raise SystemExit("source preparer could not start; private diagnostic detail redacted") from None finally: if context_path.exists(): context_path.unlink() if completed.returncode != 0: - detail = completed.stdout.strip() or completed.stderr.strip() or f"exit {completed.returncode}" - raise SystemExit(f"source preparer rejected the artifact: {detail}") + raise SystemExit("source preparer failed; private diagnostic detail redacted") try: status = json.loads(completed.stdout) except json.JSONDecodeError as exc: diff --git a/scripts/prepare_kb_context_from_canonical_dump.py b/scripts/prepare_kb_context_from_canonical_dump.py index 667ca2f..833e5e0 100644 --- a/scripts/prepare_kb_context_from_canonical_dump.py +++ b/scripts/prepare_kb_context_from_canonical_dump.py @@ -92,11 +92,10 @@ def _decode_copy_field(value: str, *, encoding: str = DEFAULT_COPY_ENCODING) -> continue if escaped == "x": match = re.match(r"[0-9A-Fa-f]{1,2}", value[index + 1 :]) - if match is None: - raise ContextError("COPY field contains a malformed hexadecimal byte escape") - result.append(int(match.group(0), 16)) - index += 1 + len(match.group(0)) - continue + if match is not None: + result.append(int(match.group(0), 16)) + index += 1 + len(match.group(0)) + continue if escaped in "01234567": match = re.match(r"[0-7]{1,3}", value[index:]) if match is None: @@ -122,7 +121,7 @@ def parse_copy_table( ) -> list[dict[str, str | None]]: encoding = _copy_encoding(encoding) prefix = f"COPY public.{table} (" - lines = iter(sql.splitlines()) + lines = iter(line[:-1] if line.endswith("\r") else line for line in sql.split("\n")) for line in lines: if line.startswith(prefix) and line.endswith(") FROM stdin;"): columns = [item.strip() for item in line[len(prefix) : -len(") FROM stdin;")].split(",")] @@ -296,9 +295,7 @@ def build_context( return context, receipt -def _private_json(path: Path, value: Any) -> None: - if path.exists(): - raise ContextError(f"refusing to overwrite existing output: {path}") +def _stage_private_json(path: Path, value: Any) -> Path: path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) os.chmod(path.parent, 0o700) fd, raw_temp = tempfile.mkstemp(prefix=f".{path.name}.", suffix=".tmp", dir=path.parent) @@ -310,6 +307,18 @@ def _private_json(path: Path, value: Any) -> None: handle.write("\n") handle.flush() os.fsync(handle.fileno()) + return temp + except BaseException: + if temp.exists(): + temp.unlink() + raise + + +def _private_json(path: Path, value: Any) -> None: + if path.exists(): + raise ContextError(f"refusing to overwrite existing output: {path}") + temp = _stage_private_json(path, value) + try: os.replace(temp, path) os.chmod(path, 0o600) finally: @@ -317,6 +326,24 @@ def _private_json(path: Path, value: Any) -> None: temp.unlink() +def _publish_private_json_outputs(outputs: tuple[tuple[Path, Any], ...]) -> None: + for path, _value in outputs: + if path.exists(): + raise ContextError(f"refusing to overwrite existing output: {path}") + + staged: list[tuple[Path, Path]] = [] + try: + for path, value in outputs: + staged.append((path, _stage_private_json(path, value))) + for path, temp in staged: + os.replace(temp, path) + os.chmod(path, 0o600) + finally: + for _path, temp in staged: + if temp.exists(): + temp.unlink() + + def parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--dump", required=True, type=Path) @@ -333,6 +360,8 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace: def main(argv: list[str] | None = None) -> int: args = parse_args(argv) repo_root = Path(__file__).resolve().parents[1] + context_existed = args.context_output.is_file() + receipt_existed = args.receipt_output.is_file() try: if args.context_output.resolve() == args.receipt_output.resolve(): raise ContextError("context and receipt outputs must be different files") @@ -345,13 +374,16 @@ def main(argv: list[str] | None = None) -> int: repo_root=repo_root, copy_encoding=args.copy_encoding, ) - _private_json(args.context_output, context) receipt["outputs"] = { "context": str(args.context_output.resolve()), "receipt": str(args.receipt_output.resolve()), } - _private_json(args.receipt_output, receipt) + _publish_private_json_outputs( + ((args.context_output, context), (args.receipt_output, receipt)) + ) except (ContextError, OSError, ValueError): + context_written = not context_existed and args.context_output.is_file() + receipt_written = not receipt_existed and args.receipt_output.is_file() print( json.dumps( { @@ -359,8 +391,8 @@ def main(argv: list[str] | None = None) -> int: "status": "rejected", "database_write_performed": False, "model_call_performed": False, - "private_context_written": False, - "private_receipt_written": False, + "private_context_written": context_written, + "private_receipt_written": receipt_written, "reason": "context_preparation_failed", }, sort_keys=True, diff --git a/scripts/prepare_kb_source_manifest.py b/scripts/prepare_kb_source_manifest.py index 073b237..6beefda 100644 --- a/scripts/prepare_kb_source_manifest.py +++ b/scripts/prepare_kb_source_manifest.py @@ -594,7 +594,7 @@ def _validate_metadata(args: argparse.Namespace) -> None: raise PreparationError(f"--source-type must be one of {sorted(compiler.ap.SOURCE_TYPES)}") -def prepare(args: argparse.Namespace) -> dict[str, Any]: +def _prepare_staged(args: argparse.Namespace, published_output_dir: Path) -> dict[str, Any]: _validate_metadata(args) artifact_bytes = _read_nonempty(args.artifact, "artifact") text_bytes = extract_utf8_text(args.artifact, args.text) @@ -751,10 +751,10 @@ def prepare(args: argparse.Namespace) -> dict[str, Any]: "exact_selected_locations_validated": bundle is not None, }, "outputs": { - "output_dir": str(output_dir), - "extracted_text": str(text_path), - "extraction_result": str(extraction_path), - "manifest": str(manifest_path) if status == "prepared" else None, + "output_dir": str(published_output_dir), + "extracted_text": str(published_output_dir / text_path.name), + "extraction_result": str(published_output_dir / extraction_path.name), + "manifest": str(published_output_dir / manifest_path.name) if status == "prepared" else None, }, "compiler": { "validated": bundle is not None, @@ -767,11 +767,43 @@ def prepare(args: argparse.Namespace) -> dict[str, Any]: else "No novel claim was prepared; inspect duplicate judgments and do not stage an empty proposal." ), } - receipt["outputs"]["receipt"] = str(receipt_path) + receipt["outputs"]["receipt"] = str(published_output_dir / receipt_path.name) _write_private(receipt_path, _json_bytes(receipt)) return receipt +def _validate_output_target(path: Path) -> None: + if path.is_symlink(): + raise PreparationError("--output-dir must not be a symlink") + if path.exists() and (not path.is_dir() or any(path.iterdir())): + raise PreparationError("--output-dir must be absent or an empty dedicated directory") + + +def prepare(args: argparse.Namespace) -> dict[str, Any]: + requested_output_dir = args.output_dir.expanduser() + _validate_output_target(requested_output_dir) + output_dir = requested_output_dir.resolve() + output_dir.parent.mkdir(parents=True, exist_ok=True) + staging_dir = Path( + tempfile.mkdtemp( + prefix=f".{output_dir.name}.", + suffix=".staging", + dir=output_dir.parent, + ) + ) + os.chmod(staging_dir, 0o700) + staged_args = argparse.Namespace(**vars(args)) + staged_args.output_dir = staging_dir + try: + receipt = _prepare_staged(staged_args, output_dir) + _validate_output_target(output_dir) + os.replace(staging_dir, output_dir) + return receipt + finally: + if staging_dir.exists(): + shutil.rmtree(staging_dir) + + def parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--artifact", required=True, type=Path) @@ -797,11 +829,25 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace: return parser.parse_args(argv) +def _private_output_state(output_dir: Path) -> dict[str, bool]: + return { + "extracted_text": (output_dir / "extracted-text.txt").is_file(), + "extraction_result": (output_dir / "extraction-result.json").is_file(), + "manifest": (output_dir / "source-manifest.json").is_file(), + "receipt": (output_dir / "preparation-receipt.json").is_file(), + } + + def main(argv: list[str] | None = None) -> int: args = parse_args(argv) + initial_output_state = _private_output_state(args.output_dir) try: receipt = prepare(args) except (PreparationError, compiler.CompilerError, OSError, ValueError): + current_output_state = _private_output_state(args.output_dir) + written_output_state = { + name: present and not initial_output_state[name] for name, present in current_output_state.items() + } print( json.dumps( { @@ -809,7 +855,8 @@ def main(argv: list[str] | None = None) -> int: "status": "rejected", "database_write_performed": False, "canonical_apply_performed": False, - "private_outputs_written": False, + "private_outputs_written": written_output_state["receipt"], + "private_output_files": written_output_state, "reason": "source_preparation_failed", }, sort_keys=True, diff --git a/tests/test_hermes_leoclean_kb_bridge_source.py b/tests/test_hermes_leoclean_kb_bridge_source.py index 3f9a370..dbc0a5a 100644 --- a/tests/test_hermes_leoclean_kb_bridge_source.py +++ b/tests/test_hermes_leoclean_kb_bridge_source.py @@ -1715,6 +1715,69 @@ def test_vps_prepare_source_retrieves_canonical_candidates_before_model_extracti assert candidate_id not in stdout +@pytest.mark.parametrize( + ("failure_kind", "expected_error"), + [ + ("timeout", "source preparer timed out; private diagnostic detail redacted"), + ("launch", "source preparer could not start; private diagnostic detail redacted"), + ("crash", "source preparer failed; private diagnostic detail redacted"), + ], +) +def test_vps_prepare_source_wrapper_redacts_commands_and_child_output( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + failure_kind: str, + expected_error: str, +) -> None: + module = _load_module(BRIDGE_DIR / "kb_tool.py") + sentinel = "PREPARER-WRAPPER-PRIVATE-CANARY" + preparer = tmp_path / "prepare_kb_source_manifest.py" + preparer.write_text("# fixture\n", encoding="utf-8") + inbox = tmp_path / "source-inbox" + preparation_root = tmp_path / "source-preparation" + private_inbox = inbox / sentinel + private_inbox.mkdir(parents=True) + preparation_root.mkdir() + artifact = private_inbox / "source.md" + artifact.write_text("private", encoding="utf-8") + monkeypatch.setattr(module, "SOURCE_PREPARER_PATH", preparer) + monkeypatch.setattr(module, "SOURCE_INBOX_ROOT", inbox) + monkeypatch.setattr(module, "SOURCE_PREPARATION_ROOT", preparation_root) + monkeypatch.setattr(module, "find_claims", lambda *_args, **_kwargs: []) + + def fail(command, **_kwargs): + if failure_kind == "timeout": + raise subprocess.TimeoutExpired(command, 420, output=sentinel, stderr=sentinel) + if failure_kind == "launch": + raise OSError(f"could not launch {sentinel}") + return subprocess.CompletedProcess(command, 19, stdout="", stderr=sentinel) + + monkeypatch.setattr(module.subprocess, "run", fail) + args = SimpleNamespace( + local=True, + query=sentinel, + artifact=artifact, + text=None, + identity=f"document:{sentinel}", + source_key="private_source", + source_type="article", + title=sentinel, + locator=f"artifact://{sentinel}", + output_dir=preparation_root / "private-output", + model=sentinel, + ) + + with pytest.raises(SystemExit) as caught: + module.prepare_source(args) + + rendered = str(caught.value) + assert rendered == expected_error + assert sentinel not in rendered + assert str(artifact) not in rendered + assert args.identity not in rendered + assert args.locator not in rendered + + def test_vps_prepare_source_rejects_files_and_outputs_outside_private_roots(tmp_path: Path) -> None: module = _load_module(BRIDGE_DIR / "kb_tool.py") inbox = tmp_path / "source-inbox" diff --git a/tests/test_kb_tool_propose_source.py b/tests/test_kb_tool_propose_source.py index b2f83a3..33b51e6 100644 --- a/tests/test_kb_tool_propose_source.py +++ b/tests/test_kb_tool_propose_source.py @@ -5,6 +5,7 @@ import importlib.util import json import re import stat +import subprocess from pathlib import Path from types import SimpleNamespace @@ -219,6 +220,51 @@ def test_deployed_bridge_accepts_real_source_compiler_bundle(tmp_path: Path, mon assert validated_manifest["artifact_sha256"] == artifact_sha256 +@pytest.mark.parametrize( + ("failure_kind", "expected_error"), + [ + ("timeout", "source compiler timed out; private diagnostic detail redacted"), + ("launch", "source compiler could not start; private diagnostic detail redacted"), + ("crash", "source compiler failed; private diagnostic detail redacted"), + ], +) +def test_source_compiler_wrapper_redacts_commands_timeouts_and_child_output( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + failure_kind: str, + expected_error: str, +) -> None: + module = load_module() + sentinel = "COMPILER-WRAPPER-PRIVATE-CANARY" + compiler = tmp_path / "compiler.py" + compiler.write_text("# fixture\n", encoding="utf-8") + args = SimpleNamespace( + artifact=tmp_path / sentinel / "artifact.pdf", + text=tmp_path / sentinel / "extracted.txt", + manifest=tmp_path / sentinel / "manifest.json", + ) + monkeypatch.setattr(module, "SOURCE_COMPILER_PATH", compiler) + + def fail(command, **_kwargs): + if failure_kind == "timeout": + raise subprocess.TimeoutExpired(command, 120, output=sentinel, stderr=sentinel) + if failure_kind == "launch": + raise OSError(f"could not launch {sentinel}") + return subprocess.CompletedProcess(command, 17, stdout="", stderr=sentinel) + + monkeypatch.setattr(module.subprocess, "run", fail) + + with pytest.raises(SystemExit) as caught: + module.compile_source_bundle(args) + + rendered = str(caught.value) + assert rendered == expected_error + assert sentinel not in rendered + assert str(args.artifact) not in rendered + assert str(args.text) not in rendered + assert str(args.manifest) not in rendered + + def test_propose_source_refuses_nonlocal_database_route(tmp_path: Path) -> None: module = load_module() args = SimpleNamespace( diff --git a/tests/test_prepare_kb_context_from_canonical_dump.py b/tests/test_prepare_kb_context_from_canonical_dump.py index 09c1ab5..14cc0f9 100644 --- a/tests/test_prepare_kb_context_from_canonical_dump.py +++ b/tests/test_prepare_kb_context_from_canonical_dump.py @@ -125,6 +125,8 @@ def test_copy_parser_preserves_escaped_source_text() -> None: (r"caf\xc3\xa9", "caf\N{LATIN SMALL LETTER E WITH ACUTE}"), (r"price \342\202\2545", "price \N{EURO SIGN}5"), (r"literal\qescape", "literalqescape"), + (r"literal\x", "literalx"), + (r"literal\xg1", "literalxg1"), ], ) def test_copy_parser_decodes_octal_and_hex_as_encoded_bytes(encoded: str, decoded: str) -> None: @@ -137,12 +139,29 @@ def test_copy_parser_uses_explicit_encoding() -> None: context_builder._decode_copy_field(r"caf\351", encoding="utf-8") -@pytest.mark.parametrize("encoded", [r"\xff", r"\303", r"\400", r"\000", r"\x", "dangling\\"]) +@pytest.mark.parametrize("encoded", [r"\xff", r"\303", r"\400", r"\000", "dangling\\"]) def test_copy_parser_fails_closed_on_invalid_byte_escapes(encoded: str) -> None: with pytest.raises(context_builder.ContextError): context_builder._decode_copy_field(encoded) +@pytest.mark.parametrize("line_ending", ["\n", "\r\n"]) +def test_copy_parser_splits_only_copy_row_boundaries(line_ending: str) -> None: + source_text = "before\N{LINE SEPARATOR}middle\N{PARAGRAPH SEPARATOR}after" + sql = line_ending.join( + [ + "COPY public.claims (id, text) FROM stdin;", + f"{CLAIM_A}\t{source_text}", + r"\.", + "", + ] + ) + + rows = context_builder.parse_copy_table(sql, "claims") + + assert rows == [{"id": CLAIM_A, "text": source_text}] + + def test_private_json_is_owner_only_and_refuses_overwrite(tmp_path: Path) -> None: output = tmp_path / "private" / "context.json" @@ -230,3 +249,54 @@ def test_cli_redacts_private_failure_details_from_stdout( assert public_status["reason"] == "context_preparation_failed" assert sentinel not in stdout assert str(private_path) not in stdout + + +@pytest.mark.parametrize( + ("fail_on_replace", "expected_context", "expected_receipt"), + [(1, False, False), (2, True, False)], +) +def test_cli_reports_exact_state_after_each_context_publication_boundary( + tmp_path: Path, + capsys: pytest.CaptureFixture, + monkeypatch: pytest.MonkeyPatch, + fail_on_replace: int, + expected_context: bool, + expected_receipt: bool, +) -> None: + context_output = tmp_path / "private-output" / "context.json" + receipt_output = tmp_path / "private-output" / "receipt.json" + context = {"database_search_query": "private", "candidate_claims": []} + receipt = {"retrieval": {"candidate_count": 0}} + monkeypatch.setattr(context_builder, "build_context", lambda **_kwargs: (context, receipt)) + real_replace = context_builder.os.replace + replace_count = 0 + + def faulted_replace(source: Path, destination: Path) -> None: + nonlocal replace_count + replace_count += 1 + if replace_count == fail_on_replace: + raise OSError("publication fault detail must stay private") + real_replace(source, destination) + + monkeypatch.setattr(context_builder.os, "replace", faulted_replace) + status = context_builder.main( + [ + "--dump", + str(tmp_path / "dump"), + "--artifact", + str(tmp_path / "artifact"), + "--title", + "private", + "--context-output", + str(context_output), + "--receipt-output", + str(receipt_output), + ] + ) + + public_status = json.loads(capsys.readouterr().out) + assert status == 2 + assert public_status["private_context_written"] is expected_context + assert public_status["private_receipt_written"] is expected_receipt + assert context_output.is_file() is expected_context + assert receipt_output.is_file() is expected_receipt diff --git a/tests/test_prepare_kb_source_manifest.py b/tests/test_prepare_kb_source_manifest.py index 3a24148..4b2c7e4 100644 --- a/tests/test_prepare_kb_source_manifest.py +++ b/tests/test_prepare_kb_source_manifest.py @@ -494,3 +494,139 @@ def test_cli_redacts_private_preparation_failure_from_stdout( assert public_status["reason"] == "source_preparation_failed" assert sentinel not in stdout assert str(tmp_path) not in stdout + + +@pytest.mark.parametrize("fail_after_write", [1, 2, 3, 4]) +def test_prepare_never_publishes_partially_staged_output_set( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + fail_after_write: int, +) -> None: + artifact, context = _inputs(tmp_path) + args = _args(tmp_path, artifact, context) + monkeypatch.setattr( + preparer, + "call_openrouter", + lambda *_args, **_kwargs: (_model_output(), {"prompt_tokens": 100, "completion_tokens": 50}), + ) + real_write = preparer._write_private + write_count = 0 + + def faulted_write(path: Path, value: bytes) -> None: + nonlocal write_count + real_write(path, value) + write_count += 1 + if write_count == fail_after_write: + raise OSError("staging fault detail must stay private") + + monkeypatch.setattr(preparer, "_write_private", faulted_write) + + with pytest.raises(OSError): + preparer.prepare(args) + + assert not args.output_dir.exists() or not any(args.output_dir.iterdir()) + assert not list(args.output_dir.parent.glob(f".{args.output_dir.name}.*.staging")) + + +@pytest.mark.parametrize( + ("fault_after_publication", "expected_published"), + [(False, False), (True, True)], +) +def test_prepare_atomic_publication_boundary_is_all_or_nothing( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + fault_after_publication: bool, + expected_published: bool, +) -> None: + artifact, context = _inputs(tmp_path) + args = _args(tmp_path, artifact, context) + monkeypatch.setattr( + preparer, + "call_openrouter", + lambda *_args, **_kwargs: (_model_output(), {"prompt_tokens": 100, "completion_tokens": 50}), + ) + real_replace = preparer.os.replace + + def faulted_replace(source: Path, destination: Path) -> None: + if Path(destination) != args.output_dir: + real_replace(source, destination) + return + if fault_after_publication: + real_replace(source, destination) + raise OSError("publication fault detail must stay private") + + monkeypatch.setattr(preparer.os, "replace", faulted_replace) + + with pytest.raises(OSError): + preparer.prepare(args) + + expected_state = { + "extracted_text": expected_published, + "extraction_result": expected_published, + "manifest": expected_published, + "receipt": expected_published, + } + assert preparer._private_output_state(args.output_dir) == expected_state + assert not list(args.output_dir.parent.glob(f".{args.output_dir.name}.*.staging")) + + +@pytest.mark.parametrize( + "published_names", + [ + ("extracted-text.txt",), + ("extracted-text.txt", "extraction-result.json"), + ("extracted-text.txt", "extraction-result.json", "source-manifest.json"), + ( + "extracted-text.txt", + "extraction-result.json", + "source-manifest.json", + "preparation-receipt.json", + ), + ], +) +def test_cli_failure_reports_actual_state_after_each_private_publication_boundary( + tmp_path: Path, + capsys: pytest.CaptureFixture, + monkeypatch: pytest.MonkeyPatch, + published_names: tuple[str, ...], +) -> None: + output_dir = tmp_path / "private-output" + + def fail(args) -> None: + args.output_dir.mkdir(parents=True) + for name in published_names: + (args.output_dir / name).write_text("private", encoding="utf-8") + raise preparer.PreparationError("private publication fault") + + monkeypatch.setattr(preparer, "prepare", fail) + status = preparer.main( + [ + "--artifact", + str(tmp_path / "source.md"), + "--kb-context", + str(tmp_path / "context.json"), + "--identity", + "document:private", + "--source-key", + "private_source", + "--source-type", + "article", + "--title", + "private", + "--locator", + "artifact://private", + "--output-dir", + str(output_dir), + ] + ) + + public_status = json.loads(capsys.readouterr().out) + expected_state = { + "extracted_text": "extracted-text.txt" in published_names, + "extraction_result": "extraction-result.json" in published_names, + "manifest": "source-manifest.json" in published_names, + "receipt": "preparation-receipt.json" in published_names, + } + assert status == 2 + assert public_status["private_output_files"] == expected_state + assert public_status["private_outputs_written"] is expected_state["receipt"]