from __future__ import annotations import copy import hashlib import json import os import stat import subprocess import sys import tarfile import uuid from pathlib import Path import pytest REPO_ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(REPO_ROOT / "scripts")) import compile_kb_source_packet as compiler # noqa: E402, I001 ARTIFACT_BYTES = b"%PDF-1.7\nsource compiler fixture\x00\xff\n%%EOF\n" EXTRACTED_TEXT = ( "Source packet compiler note.\n\n" "A staged proposal remains non-canonical until a human review and guarded apply both succeed.\n\n" "Every proposed claim must retain an exact quote from the extracted source text." ) CLAIM_QUOTE = "A staged proposal remains non-canonical until a human review and guarded apply both succeed." EVIDENCE_QUOTE = "Every proposed claim must retain an exact quote from the extracted source text." EXTRACTOR_SPEC_BYTES = ( b"livingip semantic extractor specification\n" b"name=fixture-semantic-extractor\n" b"version=3.1.0\n" b"quotes=exact-utf8-byte-locators\n" ) REPAIRED_WRITER_HEAD = "6fdaf19b96d027bc614e8ade0d475ba0954740b9" V1_RENDERED_BUNDLE_SHA256 = "64795816f6b28e61df34cd00a62ffae12c7cc40bb26c475a56d47b4507d1bf73" V2_RENDERED_BUNDLE_SHA256 = "e0a240371885a2c7fc714dd1b08099ef122d472e76ae1065542c1f305a27d816" EXACT_WRITER_FILES = ( "scripts/apply_proposal.py", "scripts/kb_apply_replay_receipt.py", ) EXACT_WRITER_PROBE = r""" import copy import hashlib import json import sys from pathlib import Path scripts_dir = Path(sys.argv[1]).resolve() child_path = Path(sys.argv[2]).resolve() resolved_commit = sys.argv[3] forbidden_checkout = Path(sys.argv[4]).resolve() def canonical_bytes(value): return json.dumps(value, ensure_ascii=True, separators=(",", ":"), sort_keys=True).encode("utf-8") def sha256_bytes(value): return hashlib.sha256(value).hexdigest() sys.path.insert(0, str(scripts_dir)) import apply_proposal as writer import kb_apply_replay_receipt as replay_receipt module_origins = { "apply_proposal": str(Path(writer.__file__).resolve()), "kb_apply_replay_receipt": str(Path(replay_receipt.__file__).resolve()), } expected_origins = { "apply_proposal": str(scripts_dir / "apply_proposal.py"), "kb_apply_replay_receipt": str(scripts_dir / "kb_apply_replay_receipt.py"), } assert module_origins == expected_origins for origin in module_origins.values(): origin_path = Path(origin) assert origin_path.is_relative_to(scripts_dir) assert not origin_path.is_relative_to(forbidden_checkout) def forbidden_process(*_args, **_kwargs): raise AssertionError("exact-writer compatibility probe attempted process or database execution") writer.subprocess.run = forbidden_process raw_child = child_path.read_bytes() child = json.loads(raw_child.decode("utf-8")) payload_before = copy.deepcopy(child["payload"]) apply_payload = child["payload"]["apply_payload"] apply_payload_before = canonical_bytes(apply_payload) normalized = writer._normalize_v3_approve_claim(apply_payload) roles = {"supports": "grounds", "challenges": "contradicts", "illustrates": "illustrates"} expected_evidence = [] for row in apply_payload["evidence"]: expected = copy.deepcopy(row) expected["role"] = roles[row["polarity"]] expected_evidence.append(expected) assert normalized == { "agent_id": apply_payload["agent_id"], "claims": apply_payload["claims"], "sources": apply_payload["sources"], "evidence": expected_evidence, "edges": apply_payload["edges"], "assessments": apply_payload["assessments"], } assert normalized["sources"][0]["captured_at"] == apply_payload["sources"][0]["captured_at"] assert apply_payload["sources"][0]["captured_at"].endswith(".000000Z") assert canonical_bytes(apply_payload) == apply_payload_before approved = copy.deepcopy(child) approved.update( status="approved", reviewed_by_handle="source-compiler-v3-exact-writer-test", reviewed_by_agent_id=None, reviewed_at="2000-01-01T00:00:00.000000Z", review_note="Build-only exact-writer compatibility validation.", ) sql = writer.build_apply_sql(approved, writer.SERVICE_AGENT_HANDLE) assert "begin;" in sql assert approved["payload"] == payload_before assert child["payload"] == payload_before assert child_path.read_bytes() == raw_child assert sha256_bytes(canonical_bytes(child["payload"])) == child["payload_sha256"] receipt = { "resolved_commit": resolved_commit, "module_origins": module_origins, "module_sha256": { name: sha256_bytes(Path(origin).read_bytes()) for name, origin in module_origins.items() }, "input_child_file_sha256": sha256_bytes(raw_child), "strict_child_payload_sha256": child["payload_sha256"], "apply_payload_sha256_before": sha256_bytes(apply_payload_before), "apply_payload_sha256_after": sha256_bytes(canonical_bytes(apply_payload)), "payload_file_bytes_preserved": child_path.read_bytes() == raw_child, "payload_object_preserved": child["payload"] == payload_before, "canonical_timestamp_equal": normalized["sources"][0]["captured_at"] == apply_payload["sources"][0]["captured_at"], "build_apply_sql_sha256": sha256_bytes(sql.encode("utf-8")), "sql_executed": False, } print(json.dumps(receipt, ensure_ascii=True, sort_keys=True)) """ def _sha256(value: bytes) -> str: return hashlib.sha256(value).hexdigest() def _manifest() -> dict: artifact_sha256 = _sha256(ARTIFACT_BYTES) text_bytes = EXTRACTED_TEXT.encode("utf-8") return { "schema": compiler.MANIFEST_SCHEMA, "artifact_sha256": artifact_sha256, "extracted_text_sha256": _sha256(text_bytes), "extractor": {"name": "fixture-text-extractor", "version": "1"}, "source": { "identity": f"document:sha256:{artifact_sha256}", "source_key": "source_compiler_note", "source_type": "article", "title": "Source packet compiler note", "locator": f"artifact://sha256/{artifact_sha256}", }, "claims": [ { "claim_key": "review_gate_precedes_canonical_apply", "type": "structural", "text": "A staged proposal is not canonical knowledge until review and guarded apply succeed.", "quote": CLAIM_QUOTE, "confidence": 0.99, "tags": ["review", "provenance"], "evidence": [{"quote": EVIDENCE_QUOTE, "role": "grounds"}], } ], } def _byte_locator(quote: str, text: str = EXTRACTED_TEXT) -> dict: text_bytes = text.encode("utf-8") quote_bytes = quote.encode("utf-8") start = text_bytes.index(quote_bytes) return {"unit": "utf8_bytes", "start": start, "end": start + len(quote_bytes)} def _manifest_v3( *, artifact_bytes: bytes = ARTIFACT_BYTES, extracted_text: str = EXTRACTED_TEXT, extractor_spec_bytes: bytes = EXTRACTOR_SPEC_BYTES, ) -> dict: artifact_sha256 = _sha256(artifact_bytes) return { "schema": compiler.MANIFEST_SCHEMA_V3, "artifact_sha256": artifact_sha256, "extracted_text_sha256": _sha256(extracted_text.encode("utf-8")), "extractor": { "name": "fixture-semantic-extractor", "version": "3.1.0", "spec_sha256": _sha256(extractor_spec_bytes), }, "agent_id": "11111111-1111-4111-8111-111111111111", "source": { "identity": f"document:sha256:{artifact_sha256}", "source_key": "source_compiler_note_v3", "source_type": "article", "title": "Source packet compiler note", "locator": f"artifact://sha256/{artifact_sha256}", "access_scope": "collective_internal", "ingestion_origin": "operator_upload", "provenance_status": "verified", "source_class": "primary_record", "auto_promotion_policy": "human_required", "captured_at": "2026-07-18T10:00:00+00:00", }, "claims": [ { "claim_key": "review_gate_precedes_canonical_apply_v3", "type": "empirical", "proposition": "A staged proposal remains non-canonical until review and guarded apply succeed.", "body_markdown": "The source explicitly separates staging from review and guarded canonical apply.", "scope": "The retained source packet compiler note and its exact cited sentence.", "access_scope": "collective_internal", "owner_agent_id": None, "status": "active", "revision_criteria": "Revise if the bound source sentence changes or no longer supports the proposition.", "revision_kind": "original", "supersedes_id": None, "tags": ["review", "v3", "provenance"], "evidence": [ { "evidence_key": "review_gate_exact_sentence", "quote": CLAIM_QUOTE, "polarity": "supports", "locator": _byte_locator(CLAIM_QUOTE, extracted_text), "rationale": "The exact sentence directly supports the bounded proposition.", } ], "assessment": { "support_tier": "strong", "challenge_tier": "none", "overall_state": "support_dominant", "rationale": "One exact verified primary-record sentence directly supports the claim.", "supersedes_assessment_id": None, }, } ], } def _write_inputs(tmp_path: Path, manifest: dict | None = None, prefix: str = "input") -> tuple[Path, Path, Path]: tmp_path.mkdir(parents=True, exist_ok=True) artifact = tmp_path / f"{prefix}.bin" text = tmp_path / f"{prefix}.txt" manifest_path = tmp_path / f"{prefix}.json" artifact.write_bytes(ARTIFACT_BYTES) text.write_text(EXTRACTED_TEXT, encoding="utf-8") manifest_path.write_text(json.dumps(manifest or _manifest(), indent=2), encoding="utf-8") return artifact, text, manifest_path def _compile(tmp_path: Path, manifest: dict | None = None) -> dict: paths = _write_inputs(tmp_path, manifest) return compiler.compile_source_packet(*paths) def _write_v3_inputs( tmp_path: Path, manifest: dict | None = None, *, artifact_bytes: bytes = ARTIFACT_BYTES, extracted_text: str = EXTRACTED_TEXT, extractor_spec_bytes: bytes = EXTRACTOR_SPEC_BYTES, prefix: str = "v3-input", ) -> tuple[Path, Path, Path, Path]: tmp_path.mkdir(parents=True, exist_ok=True) artifact = tmp_path / f"{prefix}.bin" text = tmp_path / f"{prefix}.txt" manifest_path = tmp_path / f"{prefix}.json" extractor_spec = tmp_path / f"{prefix}.extractor-spec" artifact.write_bytes(artifact_bytes) text.write_text(extracted_text, encoding="utf-8") manifest_path.write_text( json.dumps( manifest or _manifest_v3( artifact_bytes=artifact_bytes, extracted_text=extracted_text, extractor_spec_bytes=extractor_spec_bytes, ), indent=2, ), encoding="utf-8", ) extractor_spec.write_bytes(extractor_spec_bytes) return artifact, text, manifest_path, extractor_spec def _compile_v3( tmp_path: Path, manifest: dict | None = None, *, artifact_bytes: bytes = ARTIFACT_BYTES, extracted_text: str = EXTRACTED_TEXT, extractor_spec_bytes: bytes = EXTRACTOR_SPEC_BYTES, ) -> dict: return compiler.compile_source_packet( *_write_v3_inputs( tmp_path, manifest, artifact_bytes=artifact_bytes, extracted_text=extracted_text, extractor_spec_bytes=extractor_spec_bytes, ) ) def _rendered_bundle_sha256(bundle: dict) -> str: rendered = json.dumps(bundle, ensure_ascii=True, indent=2, sort_keys=True) + "\n" return _sha256(rendered.encode("utf-8")) def _resolve_exact_writer_commit(requested_head: str) -> str: assert requested_head == REPAIRED_WRITER_HEAD, ( "KB_SOURCE_COMPILER_WRITER_COMPAT_HEAD must name the exact reviewed writer commit " f"{REPAIRED_WRITER_HEAD}, got {requested_head!r}" ) completed = subprocess.run( ["git", "rev-parse", "--verify", f"{requested_head}^{{commit}}"], cwd=REPO_ROOT, capture_output=True, check=False, text=True, ) assert completed.returncode == 0, ( f"could not resolve exact writer commit {requested_head}: {completed.stderr.strip()}" ) resolved = completed.stdout.strip() assert resolved == requested_head, f"writer ref resolved to {resolved}, expected exact commit {requested_head}" return resolved def _git_file_sha256(commit: str, path: str) -> str: completed = subprocess.run( ["git", "show", f"{commit}:{path}"], cwd=REPO_ROOT, capture_output=True, check=False, ) assert completed.returncode == 0, f"could not read {path} from exact writer commit {commit}" return _sha256(completed.stdout) def _run_exact_writer_compatibility(child: dict, tmp_path: Path) -> dict: tmp_path.mkdir(parents=True, exist_ok=True) requested_head = os.environ.get("KB_SOURCE_COMPILER_WRITER_COMPAT_HEAD", REPAIRED_WRITER_HEAD) resolved = _resolve_exact_writer_commit(requested_head) writer_root = tmp_path / "exact-writer" writer_root.mkdir() archive_path = tmp_path / "exact-writer.tar" archived = subprocess.run( [ "git", "archive", "--format=tar", f"--output={archive_path}", resolved, "--", *EXACT_WRITER_FILES, ], cwd=REPO_ROOT, capture_output=True, check=False, text=True, ) assert archived.returncode == 0, f"could not materialize exact writer commit {resolved}: {archived.stderr}" with tarfile.open(archive_path, "r") as archive: for member in archive.getmembers(): member_path = Path(member.name) assert not member_path.is_absolute() and ".." not in member_path.parts archive.extractall(writer_root) writer_scripts = writer_root / "scripts" expected_origins = { "apply_proposal": str((writer_scripts / "apply_proposal.py").resolve()), "kb_apply_replay_receipt": str((writer_scripts / "kb_apply_replay_receipt.py").resolve()), } expected_module_sha256 = { "apply_proposal": _git_file_sha256(resolved, "scripts/apply_proposal.py"), "kb_apply_replay_receipt": _git_file_sha256(resolved, "scripts/kb_apply_replay_receipt.py"), } child_path = tmp_path / "strict-child.json" child_bytes = compiler.canonical_json_bytes(child) child_path.write_bytes(child_bytes) completed = subprocess.run( [ sys.executable, "-I", "-c", EXACT_WRITER_PROBE, str(writer_scripts), str(child_path), resolved, str(REPO_ROOT), ], cwd=tmp_path, capture_output=True, check=False, text=True, env={"PYTHONHASHSEED": "0"}, ) assert completed.returncode == 0, ( f"exact writer {resolved} compatibility execution failed\nstdout: {completed.stdout}\nstderr: {completed.stderr}" ) receipt = json.loads(completed.stdout) assert receipt["resolved_commit"] == resolved assert receipt["module_origins"] == expected_origins assert receipt["module_sha256"] == expected_module_sha256 assert receipt["input_child_file_sha256"] == _sha256(child_bytes) assert receipt["strict_child_payload_sha256"] == child["payload_sha256"] assert receipt["apply_payload_sha256_before"] == receipt["apply_payload_sha256_after"] assert receipt["payload_file_bytes_preserved"] is True assert receipt["payload_object_preserved"] is True assert receipt["canonical_timestamp_equal"] is True assert receipt["sql_executed"] is False assert len(receipt["build_apply_sql_sha256"]) == 64 receipt["requested_commit"] = requested_head receipt_path = tmp_path / "exact-writer-compatibility-receipt.json" receipt_path.write_text(json.dumps(receipt, ensure_ascii=True, indent=2, sort_keys=True) + "\n", encoding="utf-8") receipt["receipt_path"] = str(receipt_path) return receipt def test_compiles_deterministic_hash_bound_pending_review_bundle_without_db_write( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: artifact, text, manifest_path = _write_inputs(tmp_path, prefix="first") def unexpected_db_call(*_args, **_kwargs): raise AssertionError("compiler attempted a database write") monkeypatch.setattr(compiler.stage, "stage_normalized_child", unexpected_db_call) monkeypatch.setattr(compiler.stage.bound, "_psql_json", unexpected_db_call) first = compiler.compile_source_packet(artifact, text, manifest_path) copied = tmp_path / "copied" copied.mkdir() copied_paths = _write_inputs(copied, prefix="different-path") second = compiler.compile_source_packet(*copied_paths) assert first == second assert first["status"] == "pending_review" assert first["build_only"] is True assert first["database_write_performed"] is False assert first["canonical_apply_performed"] is False assert first["stage_preview"]["executed"] is False assert len(first["stage_preview"]["stage_sql_sha256"]) == 64 parent = first["parent_proposal"] child = first["strict_child_proposal"] apply_payload = child["payload"]["apply_payload"] assert parent["status"] == child["status"] == "pending_review" assert len(parent["payload"]["source_candidates"]) == 1 assert parent["payload"]["source_candidates"][0]["source_key"] == "source_compiler_note" assert child["proposal_type"] == "approve_claim" assert apply_payload["contract_version"] == 2 assert apply_payload["source_proposal_id"] == parent["id"] assert len(apply_payload["claims"]) == 1 assert apply_payload["claims"][0]["text"] == _manifest()["claims"][0]["text"] assert apply_payload["claims"][0]["text"] != CLAIM_QUOTE assert len(apply_payload["sources"]) == 2 assert len(apply_payload["evidence"]) == 2 assert first["hashes"]["artifact_sha256"] in {row["hash"] for row in apply_payload["sources"]} assert first["compiler_contract"]["source_from_chat_labels"] is False assert parent["payload"]["claim_candidates"][0]["headline"] == _manifest()["claims"][0]["text"] assert parent["payload"]["claim_candidates"][0]["body"] == CLAIM_QUOTE without_bundle_hash = copy.deepcopy(first) bundle_hash = without_bundle_hash["hashes"].pop("bundle_content_sha256") assert bundle_hash == compiler.canonical_sha256(without_bundle_hash) def test_v2_manifest_carries_canonical_dedupe_candidates_into_review_packet(tmp_path: Path) -> None: manifest = _manifest() manifest["schema"] = compiler.MANIFEST_SCHEMA_V2 candidate_id = "11111111-1111-4111-8111-111111111111" manifest["dedupe"] = { "database_search_query": "review guarded canonical apply", "candidate_claims": [ { "id": candidate_id, "text": "Canonical writes require a guarded apply step.", "score": 3, "evidence_count": 2, "edge_count": 1, } ], "duplicates": [ { "claim_id": candidate_id, "source_quote": CLAIM_QUOTE, "reason": "The retrieved row already captures the review-before-apply argument.", } ], } bundle = _compile(tmp_path, manifest) assessment = bundle["parent_proposal"]["payload"]["dedupe_conflict_assessment"] assert bundle["validated_manifest"]["schema"] == compiler.MANIFEST_SCHEMA_V2 assert assessment["database_search_query"] == "review guarded canonical apply" assert assessment["canonical_candidates"][0]["id"] == candidate_id assert assessment["potential_existing_claim_ids"] == [candidate_id] assert assessment["conflicts"][0]["source_quote"] == CLAIM_QUOTE staged_assessment = bundle["strict_child_proposal"]["payload"]["apply_payload"]["normalization_manifest"][ "dedupe_conflict_assessment" ] assert staged_assessment == assessment def test_v1_v2_rendered_bundle_bytes_match_pre_repair_baselines(tmp_path: Path) -> None: artifact, text, manifest_path = _write_inputs(tmp_path / "v1") v1 = compiler.compile_source_packet(artifact, text, manifest_path) v2_manifest = _manifest() v2_manifest["schema"] = compiler.MANIFEST_SCHEMA_V2 v2_manifest["dedupe"] = { "database_search_query": "test query", "candidate_claims": [], "duplicates": [], } artifact, text, manifest_path = _write_inputs(tmp_path / "v2", v2_manifest) v2 = compiler.compile_source_packet(artifact, text, manifest_path) assert _rendered_bundle_sha256(v1) == V1_RENDERED_BUNDLE_SHA256 assert _rendered_bundle_sha256(v2) == V2_RENDERED_BUNDLE_SHA256 def test_v3_compiles_exact_deterministic_pending_review_payload_without_db_write( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: first_paths = _write_v3_inputs(tmp_path / "first", prefix="original-source") def unexpected_db_call(*_args, **_kwargs): raise AssertionError("V3 compiler attempted a database write") monkeypatch.setattr(compiler.stage, "stage_normalized_child", unexpected_db_call) monkeypatch.setattr(compiler.stage.bound, "_psql_json", unexpected_db_call) first = compiler.compile_source_packet(*first_paths) second = compiler.compile_source_packet(*_write_v3_inputs(tmp_path / "copied", prefix="copied-source")) assert first == second assert first["schema"] == compiler.BUNDLE_SCHEMA_V3 assert first["status"] == "pending_review" assert first["build_only"] is True assert first["database_write_performed"] is False assert first["canonical_apply_performed"] is False assert first["stage_preview"] == { "target_table": "kb_stage.kb_proposals", "target_status": "pending_review", "executed": False, "stage_sql_sha256": first["stage_preview"]["stage_sql_sha256"], } assert len(first["stage_preview"]["stage_sql_sha256"]) == 64 parent = first["parent_proposal"] child = first["strict_child_proposal"] stage_sql = compiler.stage.build_stage_sql(child) assert stage_sql.count("insert into kb_stage.kb_proposals") == 1 assert "'pending_review'" in stage_sql assert "public." not in stage_sql assert first["stage_preview"]["stage_sql_sha256"] == _sha256(stage_sql.encode("utf-8")) apply_payload = child["payload"]["apply_payload"] assert parent["status"] == child["status"] == "pending_review" assert child["proposal_type"] == "approve_claim" assert child["payload_sha256"] == compiler.canonical_sha256(child["payload"]) assert apply_payload["contract_version"] == 3 assert apply_payload["source_proposal_id"] == parent["id"] assert apply_payload["agent_id"] == _manifest_v3()["agent_id"] assert apply_payload["edges"] == [] assert apply_payload["reasoning_tools"] == [] assert len(apply_payload["sources"]) == 1 assert len(apply_payload["claims"]) == 1 assert len(apply_payload["evidence"]) == 1 assert len(apply_payload["assessments"]) == 1 source = apply_payload["sources"][0] assert source == { "id": source["id"], "source_type": "article", "canonical_url": None, "storage_uri": _manifest_v3()["source"]["locator"], "excerpt": CLAIM_QUOTE, "content_hash": _sha256(ARTIFACT_BYTES), "access_scope": "collective_internal", "ingestion_origin": "operator_upload", "provenance_status": "verified", "source_class": "primary_record", "auto_promotion_policy": "human_required", "captured_at": "2026-07-18T10:00:00.000000Z", "created_by": _manifest_v3()["agent_id"], } claim = apply_payload["claims"][0] assert claim["proposition"] == _manifest_v3()["claims"][0]["proposition"] assert claim["body_markdown"] == _manifest_v3()["claims"][0]["body_markdown"] assert claim["scope"] == _manifest_v3()["claims"][0]["scope"] assert claim["revision_criteria"] == _manifest_v3()["claims"][0]["revision_criteria"] assert claim["revision_kind"] == "original" assert claim["supersedes_id"] is None assert claim["created_by"] == apply_payload["agent_id"] evidence = apply_payload["evidence"][0] assert evidence["claim_id"] == claim["id"] assert evidence["source_id"] == source["id"] assert evidence["excerpt"] == CLAIM_QUOTE assert evidence["polarity"] == "supports" assert evidence["source_content_hash"] == source["content_hash"] locator = evidence["locator_json"] assert locator["unit"] == "utf8_bytes" assert EXTRACTED_TEXT.encode("utf-8")[locator["start"] : locator["end"]] == CLAIM_QUOTE.encode("utf-8") assert locator["quote_sha256"] == _sha256(CLAIM_QUOTE.encode("utf-8")) assert locator["artifact_sha256"] == _sha256(ARTIFACT_BYTES) assert locator["extracted_text_sha256"] == _sha256(EXTRACTED_TEXT.encode("utf-8")) assert locator["extractor"] == _manifest_v3()["extractor"] assessment = apply_payload["assessments"][0] assert assessment["claim_id"] == claim["id"] assert assessment["evidence_set_hash"] == compiler.ap._v3_evidence_set_hash(claim["id"], [evidence]) assert assessment["supersedes_assessment_id"] is None all_ids = [ parent["id"], child["id"], source["id"], claim["id"], evidence["id"], assessment["id"], ] assert len(all_ids) == len(set(all_ids)) assert all(value == str(uuid.UUID(value)) for value in all_ids) identity = apply_payload["normalization_manifest"]["identity_material"] assert identity == { "artifact_sha256": _sha256(ARTIFACT_BYTES), "extracted_text_sha256": _sha256(EXTRACTED_TEXT.encode("utf-8")), "extractor": _manifest_v3()["extractor"], "manifest_file_sha256": first["hashes"]["manifest_file_sha256"], "source_identity": _manifest_v3()["source"]["identity"], "source_locator": _manifest_v3()["source"]["locator"], "normalized_candidate_content_sha256": first["hashes"]["normalized_candidate_content_sha256"], } assert apply_payload["normalization_manifest"]["semantic_extraction_is_retained_input"] is True assert apply_payload["normalization_manifest"]["model_training_claim"] is False assert "retained input" in apply_payload["normalization_manifest"]["semantic_extraction_statement"] assert first["hashes"]["extractor_spec_sha256"] == _sha256(EXTRACTOR_SPEC_BYTES) assert first["compiler_contract"]["writer_apply_invoked"] is False without_bundle_hash = copy.deepcopy(first) bundle_hash = without_bundle_hash["hashes"].pop("bundle_content_sha256") assert bundle_hash == compiler.canonical_sha256(without_bundle_hash) def test_v3_exact_writer_commit_preserves_payload_and_builds_apply_sql(tmp_path: Path) -> None: bundle = _compile_v3(tmp_path) apply_payload = bundle["strict_child_proposal"]["payload"]["apply_payload"] assert apply_payload["sources"][0]["captured_at"] == "2026-07-18T10:00:00.000000Z" receipt = _run_exact_writer_compatibility(bundle["strict_child_proposal"], tmp_path / "writer-proof") assert receipt["resolved_commit"] == REPAIRED_WRITER_HEAD assert receipt["apply_payload_sha256_before"] == receipt["apply_payload_sha256_after"] assert Path(receipt["receipt_path"]).is_file() def test_v3_exact_writer_gate_cannot_fall_back_to_branch_local_apply_module( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: bundle = _compile_v3(tmp_path / "payload") branch_local_origin = str(Path(compiler.ap.__file__).resolve()) def forbidden_branch_local_writer(*_args, **_kwargs): raise AssertionError("exact writer gate used branch-local compiler.ap") monkeypatch.setattr(compiler.ap, "_normalize_v3_approve_claim", forbidden_branch_local_writer) monkeypatch.setattr(compiler.ap, "build_apply_sql", forbidden_branch_local_writer) receipt = _run_exact_writer_compatibility(bundle["strict_child_proposal"], tmp_path / "writer-proof") assert branch_local_origin not in receipt["module_origins"].values() assert all(Path(origin).is_relative_to(tmp_path) for origin in receipt["module_origins"].values()) def test_v3_cli_rerun_is_byte_identical_and_emits_only_public_status(tmp_path: Path) -> None: private_root = tmp_path / "private-v3-source" artifact, text, manifest_path, extractor_spec = _write_v3_inputs(private_root) outputs = [private_root / "compiled" / "first.json", private_root / "compiled" / "second.json"] completed = [] for output in outputs: completed.append( subprocess.run( [ sys.executable, str(REPO_ROOT / "scripts" / "compile_kb_source_packet.py"), "--artifact", str(artifact), "--text", str(text), "--manifest", str(manifest_path), "--extractor-spec", str(extractor_spec), "--output", str(output), ], capture_output=True, check=False, ) ) assert [item.returncode for item in completed] == [0, 0] assert [item.stderr for item in completed] == [b"", b""] assert completed[0].stdout == completed[1].stdout assert outputs[0].read_bytes() == outputs[1].read_bytes() assert json.loads(outputs[0].read_text(encoding="utf-8"))["schema"] == compiler.BUNDLE_SCHEMA_V3 public_status = json.loads(completed[0].stdout.decode("utf-8")) assert public_status["status"] == "pending_review" assert public_status["database_write_performed"] is False assert public_status["canonical_apply_performed"] is False assert EXTRACTED_TEXT.encode("utf-8") not in completed[0].stdout assert str(private_root).encode("utf-8") not in completed[0].stdout assert all(stat.S_IMODE(output.stat().st_mode) == 0o600 for output in outputs) def test_v3_stale_artifact_fails_closed_and_rehashed_artifact_changes_identity(tmp_path: Path) -> None: original = _compile_v3(tmp_path / "original") mutated_bytes = ARTIFACT_BYTES + b"mutated" stale_manifest = _manifest_v3() with pytest.raises(compiler.CompilerError, match="artifact SHA-256 does not match"): _compile_v3(tmp_path / "stale", stale_manifest, artifact_bytes=mutated_bytes) rehashed_manifest = _manifest_v3(artifact_bytes=mutated_bytes) stale_identity = copy.deepcopy(rehashed_manifest) stale_identity["source"]["identity"] = _manifest_v3()["source"]["identity"] with pytest.raises(compiler.CompilerError, match="content-addressed identity"): _compile_v3(tmp_path / "stale-identity", stale_identity, artifact_bytes=mutated_bytes) stale_locator = copy.deepcopy(rehashed_manifest) stale_locator["source"]["locator"] = _manifest_v3()["source"]["locator"] with pytest.raises(compiler.CompilerError, match="actual artifact SHA-256"): _compile_v3(tmp_path / "stale-locator", stale_locator, artifact_bytes=mutated_bytes) rehashed = _compile_v3(tmp_path / "rehashed", rehashed_manifest, artifact_bytes=mutated_bytes) assert rehashed["strict_child_proposal"]["id"] != original["strict_child_proposal"]["id"] assert rehashed["strict_child_proposal"]["payload_sha256"] != original["strict_child_proposal"]["payload_sha256"] assert rehashed["hashes"]["packet_identity_sha256"] != original["hashes"]["packet_identity_sha256"] @pytest.mark.parametrize("target", ["extractor_name", "extractor_version", "claim"]) def test_v3_extraction_identity_and_normalized_candidate_mutations_change_proposal(tmp_path: Path, target: str) -> None: original = _compile_v3(tmp_path / "original") manifest = _manifest_v3() if target == "extractor_name": manifest["extractor"]["name"] = "fixture-semantic-extractor-revised" elif target == "extractor_version": manifest["extractor"]["version"] = "3.1.1" else: manifest["claims"][0]["proposition"] += " The proposal remains reviewable." mutated = _compile_v3(tmp_path / target, manifest) assert mutated["strict_child_proposal"]["id"] != original["strict_child_proposal"]["id"] assert mutated["strict_child_proposal"]["payload_sha256"] != original["strict_child_proposal"]["payload_sha256"] assert mutated["hashes"]["packet_identity_sha256"] != original["hashes"]["packet_identity_sha256"] def test_v3_requires_exact_independent_extractor_spec_and_binds_spec_mutations(tmp_path: Path) -> None: artifact, text, manifest_path, _extractor_spec = _write_v3_inputs(tmp_path / "missing") with pytest.raises(compiler.CompilerError, match="independently supplied extractor specification"): compiler.compile_source_packet(artifact, text, manifest_path) mismatched_spec = EXTRACTOR_SPEC_BYTES + b"mismatch\n" with pytest.raises(compiler.CompilerError, match="specification SHA-256 does not match"): _compile_v3( tmp_path / "mismatched", _manifest_v3(), extractor_spec_bytes=mismatched_spec, ) original = _compile_v3(tmp_path / "original") revised_spec = EXTRACTOR_SPEC_BYTES + b"normalization=revision-2\n" revised_manifest = _manifest_v3(extractor_spec_bytes=revised_spec) revised = _compile_v3( tmp_path / "revised", revised_manifest, extractor_spec_bytes=revised_spec, ) assert revised["hashes"]["extractor_spec_sha256"] == _sha256(revised_spec) assert revised["strict_child_proposal"]["id"] != original["strict_child_proposal"]["id"] assert revised["strict_child_proposal"]["payload_sha256"] != original["strict_child_proposal"]["payload_sha256"] assert revised["hashes"]["packet_identity_sha256"] != original["hashes"]["packet_identity_sha256"] @pytest.mark.parametrize( ("field", "value", "expected"), [ ("identity", "document:alternate:source", "content-addressed identity"), ("locator", "artifact://mirror/source", "actual artifact SHA-256"), ], ) def test_v3_rejects_unbound_source_identity_or_locator(tmp_path: Path, field: str, value: str, expected: str) -> None: manifest = _manifest_v3() manifest["source"][field] = value with pytest.raises(compiler.CompilerError, match=expected): _compile_v3(tmp_path, manifest) def test_v3_exact_manifest_bytes_are_identity_bound_but_paths_are_not(tmp_path: Path) -> None: direct_paths = _write_v3_inputs(tmp_path / "direct", prefix="source") copied_paths = _write_v3_inputs(tmp_path / "copied", prefix="renamed") direct = compiler.compile_source_packet(*direct_paths) copied = compiler.compile_source_packet(*copied_paths) assert direct == copied compact_paths = _write_v3_inputs(tmp_path / "compact", prefix="source") compact_manifest = compact_paths[2] manifest_object = json.loads(compact_manifest.read_text(encoding="utf-8")) compact_manifest.write_text( json.dumps(manifest_object, separators=(",", ":"), sort_keys=True), encoding="utf-8", ) compact = compiler.compile_source_packet(*compact_paths) assert compact["validated_manifest"] == direct["validated_manifest"] assert compact["hashes"]["manifest_canonical_sha256"] == direct["hashes"]["manifest_canonical_sha256"] assert compact["hashes"]["manifest_file_sha256"] != direct["hashes"]["manifest_file_sha256"] assert compact["hashes"]["packet_identity_sha256"] != direct["hashes"]["packet_identity_sha256"] assert compact["strict_child_proposal"]["id"] != direct["strict_child_proposal"]["id"] assert compact["strict_child_proposal"]["payload_sha256"] != direct["strict_child_proposal"]["payload_sha256"] def test_v3_valid_quote_binding_mutation_changes_identity_and_mismatches_fail_closed(tmp_path: Path) -> None: original = _compile_v3(tmp_path / "original") valid_mutation = _manifest_v3() evidence = valid_mutation["claims"][0]["evidence"][0] evidence["quote"] = EVIDENCE_QUOTE evidence["locator"] = _byte_locator(EVIDENCE_QUOTE) mutated = _compile_v3(tmp_path / "valid-mutation", valid_mutation) assert mutated["strict_child_proposal"]["id"] != original["strict_child_proposal"]["id"] assert mutated["strict_child_proposal"]["payload_sha256"] != original["strict_child_proposal"]["payload_sha256"] quote_mismatch = _manifest_v3() quote_mismatch["claims"][0]["evidence"][0]["quote"] = "Invented source sentence." with pytest.raises(compiler.CompilerError, match="does not bind the exact quote bytes"): _compile_v3(tmp_path / "quote-mismatch", quote_mismatch) locator_mismatch = _manifest_v3() locator_mismatch["claims"][0]["evidence"][0]["locator"]["start"] += 1 with pytest.raises(compiler.CompilerError, match="does not bind the exact quote bytes"): _compile_v3(tmp_path / "locator-mismatch", locator_mismatch) def test_v3_rejects_missing_assessment_relevant_evidence_and_tampered_set_hash(tmp_path: Path) -> None: manifest = _manifest_v3() manifest["claims"][0]["evidence"][0]["polarity"] = "illustrates" manifest["claims"][0]["assessment"].update( support_tier="none", challenge_tier="none", overall_state="insufficient", ) with pytest.raises(compiler.CompilerError, match="requires at least one supports or challenges evidence"): _compile_v3(tmp_path / "irrelevant", manifest) bundle = _compile_v3(tmp_path / "valid") child = copy.deepcopy(bundle["strict_child_proposal"]) child["payload"]["apply_payload"]["assessments"][0]["evidence_set_hash"] = "f" * 64 child["payload_sha256"] = compiler.canonical_sha256(child["payload"]) with pytest.raises(compiler.CompilerError, match="evidence_set_hash does not match claim-local evidence"): compiler._validate_v3_compiled_child(child, bundle["validated_manifest"], bundle["quote_bindings"]) @pytest.mark.parametrize( ("mutation", "expected"), [ ( lambda manifest: manifest.__setitem__("agent_id", manifest["agent_id"].replace("-", "")), "canonical lowercase", ), ( lambda manifest: manifest["source"].__setitem__("captured_at", "2026-07-18T10:00:00.0000000Z"), "at most 6 fractional digits", ), (lambda manifest: manifest["claims"][0].__setitem__("type", "structural"), "type must be one of"), ( lambda manifest: manifest["claims"][0]["evidence"][0]["locator"].__setitem__("start", True), "non-negative integer", ), ( lambda manifest: manifest["claims"][0]["assessment"].__setitem__("support_tier", 1), "support_tier must be null or one of", ), ], ) def test_v3_rejects_noncanonical_typed_manifest_fields(tmp_path: Path, mutation, expected: str) -> None: manifest = _manifest_v3() mutation(manifest) with pytest.raises(compiler.CompilerError, match=expected): _compile_v3(tmp_path, manifest) @pytest.mark.parametrize( ("needle", "replacement"), [ ( f'"schema": "{compiler.MANIFEST_SCHEMA_V3}",', (f'"schema": "{compiler.MANIFEST_SCHEMA_V3}",\n "schema": "{compiler.MANIFEST_SCHEMA_V3}",'), ), ( f'"schema": "{compiler.MANIFEST_SCHEMA_V3}",', f'"schema": "{compiler.MANIFEST_SCHEMA_V3}",\n "schema": "{compiler.MANIFEST_SCHEMA}",', ), ( f'"spec_sha256": "{_sha256(EXTRACTOR_SPEC_BYTES)}"', ( f'"spec_sha256": "{_sha256(EXTRACTOR_SPEC_BYTES)}",\n' f' "spec_sha256": "{_sha256(EXTRACTOR_SPEC_BYTES)}"' ), ), ( '"overall_state": "support_dominant",', '"overall_state": "support_dominant",\n "overall_state": "insufficient",', ), ], ) def test_v3_rejects_identical_or_conflicting_duplicate_json_keys_at_any_depth( tmp_path: Path, needle: str, replacement: str ) -> None: paths = _write_v3_inputs(tmp_path) manifest_path = paths[2] raw_manifest = manifest_path.read_text(encoding="utf-8") assert raw_manifest.count(needle) == 1 manifest_path.write_text(raw_manifest.replace(needle, replacement, 1), encoding="utf-8") with pytest.raises(compiler.CompilerError, match="duplicate JSON object key"): compiler.compile_source_packet(*paths) def test_v3_duplicate_schema_key_fails_before_identity_or_output( tmp_path: Path, capsys: pytest.CaptureFixture, monkeypatch: pytest.MonkeyPatch, ) -> None: artifact, text, manifest_path, extractor_spec = _write_v3_inputs(tmp_path) schema_line = f'"schema": "{compiler.MANIFEST_SCHEMA_V3}",' raw_manifest = manifest_path.read_text(encoding="utf-8") manifest_path.write_text( raw_manifest.replace(schema_line, f'{schema_line}\n "schema": "{compiler.MANIFEST_SCHEMA_V3}",', 1), encoding="utf-8", ) def unexpected_identity(*_args, **_kwargs): raise AssertionError("duplicate-key manifest reached proposal identity") monkeypatch.setattr(compiler, "_v3_identity_material", unexpected_identity) output = tmp_path / "must-not-exist.json" status = compiler.main( [ "--artifact", str(artifact), "--text", str(text), "--manifest", str(manifest_path), "--extractor-spec", str(extractor_spec), "--output", str(output), ] ) assert status == 2 assert json.loads(capsys.readouterr().out)["reason"] == "input_validation_failed" assert not output.exists() def test_v3_rejects_duplicate_semantic_keys_and_duplicate_generated_ids( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: duplicate_claim = _manifest_v3() duplicate_claim["claims"].append(copy.deepcopy(duplicate_claim["claims"][0])) with pytest.raises(compiler.CompilerError, match="duplicate claim_key"): _compile_v3(tmp_path / "claim-key", duplicate_claim) duplicate_evidence = _manifest_v3() duplicate_evidence["claims"][0]["evidence"].append(copy.deepcopy(duplicate_evidence["claims"][0]["evidence"][0])) with pytest.raises(compiler.CompilerError, match="evidence_key duplicates"): _compile_v3(tmp_path / "evidence-key", duplicate_evidence) monkeypatch.setattr( compiler, "stable_uuid_v3", lambda _row_kind, _packet_identity_sha256, _semantic_key: "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa", ) with pytest.raises(compiler.CompilerError, match="duplicate deterministic ids"): _compile_v3(tmp_path / "generated-id") def test_rejects_normalized_child_that_replaces_atomic_claim_text_with_quote( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: original = compiler.stage.prepare_normalized_child def replace_claim_text(parent: dict) -> dict: child = original(parent) child["payload"]["apply_payload"]["claims"][0]["text"] = CLAIM_QUOTE return child monkeypatch.setattr(compiler.stage, "prepare_normalized_child", replace_claim_text) with pytest.raises(compiler.CompilerError, match="claim text differs from the extracted atomic propositions"): _compile(tmp_path) def test_v2_manifest_rejects_unretrieved_or_unbound_duplicate_judgments(tmp_path: Path) -> None: manifest = _manifest() manifest["schema"] = compiler.MANIFEST_SCHEMA_V2 manifest["dedupe"] = { "database_search_query": "review guarded canonical apply", "candidate_claims": [ { "id": "11111111-1111-4111-8111-111111111111", "text": "Canonical writes require a guarded apply step.", "score": 3, "evidence_count": 2, "edge_count": 1, } ], "duplicates": [ { "claim_id": "22222222-2222-4222-8222-222222222222", "source_quote": CLAIM_QUOTE, "reason": "Not retrieved.", } ], } with pytest.raises(compiler.CompilerError, match="was not present in the canonical retrieval candidates"): _compile(tmp_path / "unretrieved", manifest) manifest["dedupe"]["duplicates"][0]["claim_id"] = "11111111-1111-4111-8111-111111111111" manifest["dedupe"]["duplicates"][0]["source_quote"] = "Invented duplicate quote." with pytest.raises(compiler.CompilerError, match="source_quote is not an exact substring"): _compile(tmp_path / "unbound", manifest) def test_cli_writes_private_bundle_and_emits_only_public_status_bytes(tmp_path: Path) -> None: private_root = tmp_path / "private-source-sentinel" artifact, text, manifest_path = _write_inputs(private_root) output = private_root / "compiled" / "packet.json" completed = subprocess.run( [ sys.executable, str(REPO_ROOT / "scripts" / "compile_kb_source_packet.py"), "--artifact", str(artifact), "--text", str(text), "--manifest", str(manifest_path), "--output", str(output), ], capture_output=True, check=False, ) stdout = completed.stdout public_status = json.loads(stdout.decode("utf-8", errors="strict")) bundle = json.loads(output.read_text(encoding="utf-8")) 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( ("field", "expected"), [ ("artifact_sha256", "artifact SHA-256 does not match"), ("extracted_text_sha256", "extracted text SHA-256 does not match"), ], ) def test_rejects_hash_mismatches(tmp_path: Path, field: str, expected: str) -> None: manifest = _manifest() manifest[field] = "0" * 64 with pytest.raises(compiler.CompilerError, match=expected): _compile(tmp_path, manifest) @pytest.mark.parametrize( ("target", "expected"), [ ("claim", "claims\\[0\\]\\.quote is not an exact substring"), ("evidence", "evidence\\[0\\]\\.quote is not an exact substring"), ], ) def test_rejects_hallucinated_claim_or_evidence_quote(tmp_path: Path, target: str, expected: str) -> None: manifest = _manifest() if target == "claim": manifest["claims"][0]["quote"] = "This quote was invented by the extractor." else: manifest["claims"][0]["evidence"][0]["quote"] = "This evidence was invented by the extractor." with pytest.raises(compiler.CompilerError, match=expected): _compile(tmp_path, manifest) def test_rejects_duplicate_claim_and_derived_source_keys(tmp_path: Path) -> None: duplicate_claim = _manifest() duplicate_claim["claims"].append(copy.deepcopy(duplicate_claim["claims"][0])) with pytest.raises(compiler.CompilerError, match="duplicate claim_key"): _compile(tmp_path / "claim", duplicate_claim) duplicate_source = _manifest() duplicate_source["source"]["source_key"] = "review_gate_precedes_canonical_apply__proposal_body" with pytest.raises(compiler.CompilerError, match="duplicate source_key after normalization"): _compile(tmp_path / "source", duplicate_source) @pytest.mark.parametrize( ("target", "expected"), [ ("claim", "claims\\[0\\]\\.type must be one of"), ("source", "source\\.source_type must be one of"), ("evidence", "evidence\\[0\\]\\.role must be one of"), ], ) def test_rejects_types_outside_current_taxonomies(tmp_path: Path, target: str, expected: str) -> None: manifest = _manifest() if target == "claim": manifest["claims"][0]["type"] = "fact" elif target == "source": manifest["source"]["source_type"] = "document" else: manifest["claims"][0]["evidence"][0]["role"] = "proves" with pytest.raises(compiler.CompilerError, match=expected): _compile(tmp_path, manifest) @pytest.mark.parametrize("field", ["identity", "locator"]) def test_rejects_missing_or_unstable_source_provenance(tmp_path: Path, field: str) -> None: manifest = _manifest() manifest["source"][field] = "message from Alice" with pytest.raises(compiler.CompilerError, match="stable URI-like reference"): _compile(tmp_path, manifest) @pytest.mark.parametrize( ("reference", "expected"), [ ("javascript:alert(1)", "unsafe URI scheme"), ("https:/missing-host", "must include a host"), ("https://user:password@example.com/source", "must not embed credentials"), ("telegram:Alice", "stable chat or message identifier"), ], ) def test_rejects_unsafe_or_non_specific_source_references(tmp_path: Path, reference: str, expected: str) -> None: manifest = _manifest() manifest["source"]["locator"] = reference with pytest.raises(compiler.CompilerError, match=expected): _compile(tmp_path, manifest) def test_rejects_tags_with_ambiguous_whitespace(tmp_path: Path) -> None: manifest = _manifest() manifest["claims"][0]["tags"] = [" review"] with pytest.raises(compiler.CompilerError, match="tags must be a list of non-empty strings"): _compile(tmp_path, manifest) def test_rejects_path_gaps_invalid_utf8_and_output_overwrite(tmp_path: Path, capsys: pytest.CaptureFixture) -> None: artifact, text, manifest_path = _write_inputs(tmp_path) with pytest.raises(compiler.CompilerError, match="artifact path is not a regular file"): compiler.compile_source_packet(tmp_path / "missing.bin", text, manifest_path) text.write_bytes(b"\xff\xfe") with pytest.raises(compiler.CompilerError, match="valid UTF-8"): compiler.compile_source_packet(artifact, text, manifest_path) status = compiler.main( [ "--artifact", str(artifact), "--text", str(text), "--manifest", str(manifest_path), "--output", str(artifact), ] ) rejection = json.loads(capsys.readouterr().out) assert status == 2 assert rejection["status"] == "rejected" assert rejection["database_write_performed"] is False assert rejection["reason"] == "input_validation_failed" assert str(artifact) not in json.dumps(rejection) def test_cli_reports_output_write_failure_as_structured_rejection( tmp_path: Path, capsys: pytest.CaptureFixture, monkeypatch: pytest.MonkeyPatch ) -> None: artifact, text, manifest_path = _write_inputs(tmp_path) output = tmp_path / "packet.json" def fail_write(_path: Path, _rendered: str) -> None: raise PermissionError("fixture output denied") monkeypatch.setattr(compiler, "_write_private_output", fail_write) status = compiler.main( [ "--artifact", str(artifact), "--text", str(text), "--manifest", str(manifest_path), "--output", str(output), ] ) rejection = json.loads(capsys.readouterr().out) assert status == 2 assert rejection["status"] == "rejected" assert rejection["database_write_performed"] is False 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: manifest = _manifest() manifest["claims"][0]["evidence"][0]["source_label"] = "Alice in Telegram" with pytest.raises(compiler.CompilerError, match="unsupported keys: source_label"): _compile(tmp_path, manifest) def test_compiles_stable_telegram_message_as_review_candidate_without_learning_from_chat(tmp_path: Path) -> None: manifest = _manifest() locator = "telegram://chat/-5146042086/message/12345" manifest["source"].update( { "identity": locator, "source_key": "telegram_message_12345", "source_type": "transcript", "title": "Telegram discussion message 12345", "locator": locator, } ) bundle = _compile(tmp_path, manifest) source = bundle["parent_proposal"]["payload"]["source_candidates"][0] assert bundle["status"] == "pending_review" assert bundle["database_write_performed"] is False assert bundle["canonical_apply_performed"] is False assert bundle["compiler_contract"]["source_from_chat_labels"] is False assert source["source_type"] == "transcript" assert source["storage_path"] == locator assert source["content_sha256"] == manifest["artifact_sha256"] def test_rejects_quotes_that_cannot_fit_the_reviewable_canonical_excerpt(tmp_path: Path) -> None: long_quote = "x" * 8100 text = f"prefix {long_quote} suffix" artifact = tmp_path / "long.bin" text_path = tmp_path / "long.txt" manifest_path = tmp_path / "long.json" artifact.write_bytes(ARTIFACT_BYTES) text_path.write_text(text, encoding="utf-8") manifest = _manifest() manifest["extracted_text_sha256"] = _sha256(text.encode("utf-8")) manifest["claims"][0]["quote"] = long_quote manifest["claims"][0]["evidence"] = [{"quote": long_quote, "role": "grounds"}] manifest_path.write_text(json.dumps(manifest), encoding="utf-8") with pytest.raises(compiler.CompilerError, match=r"truncated|excerpt capacity"): compiler.compile_source_packet(artifact, text_path, manifest_path)