1308 lines
54 KiB
Python
1308 lines
54 KiB
Python
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)
|