from __future__ import annotations import json import stat import sys from pathlib import Path from types import SimpleNamespace import pytest REPO_ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(REPO_ROOT / "scripts")) import prepare_kb_source_manifest as preparer # noqa: E402 SOURCE_TEXT = ( "Source preparation note.\n\n" "A staged proposal remains non-canonical until review and guarded apply both succeed.\n\n" "Every prepared claim must remain bound to an exact source quote.\n" ) CLAIM_QUOTE = "Every prepared claim must remain bound to an exact source quote." DUPLICATE_QUOTE = "A staged proposal remains non-canonical until review and guarded apply both succeed." CANDIDATE_ID = "11111111-1111-4111-8111-111111111111" def _args(tmp_path: Path, artifact: Path, context: Path) -> SimpleNamespace: return SimpleNamespace( artifact=artifact, text=None, kb_context=context, identity="document:source-preparation-test-v1", source_key="source_preparation_test_v1", source_type="article", title="Source preparation test", locator="artifact://source-preparation/test-v1", output_dir=tmp_path / "private-output", key_file=tmp_path / "openrouter-key", model="fixture/model", openrouter_url="https://openrouter.invalid/v1/chat/completions", timeout_seconds=30, ) def _inputs(tmp_path: Path) -> tuple[Path, Path]: artifact = tmp_path / "source.md" artifact.write_text(SOURCE_TEXT, encoding="utf-8") context = tmp_path / "context.json" context.write_text( json.dumps( { "database_search_query": "staged proposal exact source quote", "candidate_claims": [ { "id": CANDIDATE_ID, "text": "A staged proposal is not canonical before guarded apply.", "score": 3, "evidence_count": 2, "edge_count": 1, } ], } ), encoding="utf-8", ) (tmp_path / "openrouter-key").write_text("fixture-secret", encoding="utf-8") return artifact, context def _model_output(*, claims: bool = True, bad_reference: bool = False) -> str: claim_rows = [] if claims: quote_ref = "S99999" if bad_reference else "S00003" claim_rows = [ { "claim_key": "exact_quotes_bind_prepared_claims", "type": "structural", "text": "Prepared claims remain auditable when their evidence is bound to exact source quotes.", "quote_ref": quote_ref, "confidence": 0.7, "tags": ["provenance", "review"], "evidence": [{"quote_ref": quote_ref, "role": "grounds"}], } ] return json.dumps( { "schema": preparer.MODEL_OUTPUT_SCHEMA, "claims": claim_rows, "duplicates": [ { "claim_id": CANDIDATE_ID, "source_quote_ref": "S00002", "reason": "The canonical candidate already captures the review-before-apply argument.", } ], "extraction_notes": "One duplicate argument and one novel exact-quote provenance claim.", } ) def test_prepare_builds_private_compiler_validated_manifest_without_db_write( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: artifact, context = _inputs(tmp_path) monkeypatch.setattr( preparer, "call_openrouter", lambda *_args, **_kwargs: (_model_output(), {"prompt_tokens": 100, "completion_tokens": 50}), ) receipt = preparer.prepare(_args(tmp_path, artifact, context)) assert receipt["status"] == "prepared" assert receipt["database_write_performed"] is False assert receipt["canonical_apply_performed"] is False assert receipt["retrieval"]["canonical_candidate_ids"] == [CANDIDATE_ID] assert receipt["extraction"]["claim_count"] == 1 assert receipt["compiler"]["validated"] is True assert receipt["compiler"]["strict_child_proposal_id"] output_dir = Path(receipt["outputs"]["output_dir"]) manifest = json.loads(Path(receipt["outputs"]["manifest"]).read_text(encoding="utf-8")) assert manifest["schema"] == preparer.compiler.MANIFEST_SCHEMA_V2 assert manifest["dedupe"]["duplicates"][0]["claim_id"] == CANDIDATE_ID assert manifest["claims"][0]["quote"] == CLAIM_QUOTE assert manifest["dedupe"]["duplicates"][0]["source_quote"] == DUPLICATE_QUOTE assert receipt["extraction"]["selected_source_references"] == [ {"kind": "claim", "reference": "S00003"}, {"kind": "evidence", "reference": "S00003"}, {"kind": "duplicate", "reference": "S00002"}, ] assert stat.S_IMODE(output_dir.stat().st_mode) == 0o700 for path in output_dir.iterdir(): assert stat.S_IMODE(path.stat().st_mode) == 0o600 def test_prepare_retries_once_when_model_invents_a_source_reference( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: artifact, context = _inputs(tmp_path) outputs = iter((_model_output(bad_reference=True), _model_output())) prompts: list[str] = [] def fake_call(prompt: str, **_kwargs): prompts.append(prompt) return next(outputs), {} monkeypatch.setattr(preparer, "call_openrouter", fake_call) receipt = preparer.prepare(_args(tmp_path, artifact, context)) assert receipt["status"] == "prepared" assert len(receipt["extraction"]["attempts"]) == 2 assert "previous extraction failed deterministic validation" in prompts[1] assert "must select one of the bracketed SOURCE fragment IDs" in prompts[1] def test_source_fragments_preserve_exact_unicode_and_punctuation() -> None: text = "Heading\n\nIdentity pins use an exact version - never a floating alias.\nCurly: \u201creviewed\u201d.\n" fragments = preparer.build_source_fragments(text) assert fragments == { "S00001": "Heading", "S00002": "Identity pins use an exact version - never a floating alias.", "S00003": "Curly: \u201creviewed\u201d.", } assert all(value in text for value in fragments.values()) def test_source_fragment_ids_are_dense_across_blank_lines() -> None: fragments = preparer.build_source_fragments("first\n\n\nsecond\n") assert fragments == {"S00001": "first", "S00002": "second"} def test_prepare_returns_no_novel_claims_without_manifest_or_stageable_packet( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: artifact, context = _inputs(tmp_path) monkeypatch.setattr(preparer, "call_openrouter", lambda *_args, **_kwargs: (_model_output(claims=False), {})) receipt = preparer.prepare(_args(tmp_path, artifact, context)) assert receipt["status"] == "no_novel_claims" assert receipt["outputs"]["manifest"] is None assert receipt["compiler"]["validated"] is False assert "do not stage an empty proposal" in receipt["next_action"] def test_binary_artifact_requires_explicit_utf8_extraction(tmp_path: Path) -> None: artifact = tmp_path / "source.pdf" artifact.write_bytes(b"%PDF-1.7\x00binary") with pytest.raises(preparer.PreparationError, match=r"supply --text for \.pdf"): preparer.extract_utf8_text(artifact, None) def test_openrouter_key_cannot_be_redirected_to_an_unapproved_endpoint(tmp_path: Path) -> None: with pytest.raises(preparer.PreparationError, match=r"must equal https://openrouter\.ai"): preparer.call_openrouter( "prompt", model="fixture/model", key_file=tmp_path / "private-file", url="https://example.invalid/collect", timeout_seconds=1, )