782 lines
36 KiB
Python
782 lines
36 KiB
Python
from __future__ import annotations
|
|
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(REPO_ROOT / "scripts"))
|
|
|
|
import run_leo_local_ingestion_proposal_canary as canary # noqa: E402
|
|
|
|
|
|
def _loaded(path: Path) -> tuple[dict, dict, dict, dict, dict]:
|
|
fixture, input_receipt = canary.load_fixture(path)
|
|
row_ids = canary.build_row_ids(input_receipt, fixture)
|
|
parent = canary.build_parent_packet(fixture, input_receipt, row_ids)
|
|
child = canary.normalized_stage.prepare_normalized_child(parent)
|
|
return fixture, input_receipt, row_ids, parent, child
|
|
|
|
|
|
def _copy_scenario(tmp_path: Path, scenario_path: Path) -> Path:
|
|
scenario = json.loads(scenario_path.read_text(encoding="utf-8"))
|
|
artifact_source = scenario_path.parent / scenario["artifact"]["path"]
|
|
artifact_target = tmp_path / artifact_source.name
|
|
artifact_target.write_bytes(artifact_source.read_bytes())
|
|
scenario_target = tmp_path / scenario_path.name
|
|
scenario_target.write_text(json.dumps(scenario), encoding="utf-8")
|
|
return scenario_target
|
|
|
|
|
|
def _canonical_snapshot() -> dict:
|
|
return {
|
|
"claims": [{"id": "1"}],
|
|
"sources": [{"id": "2"}],
|
|
"claim_evidence": [{"claim_id": "1"}],
|
|
"claim_edges": [{"from_claim": "1"}],
|
|
}
|
|
|
|
|
|
def _row_id_values(row_ids: dict) -> set[str]:
|
|
values: set[str] = set()
|
|
for value in row_ids.values():
|
|
if isinstance(value, dict):
|
|
values.update(_row_id_values(value))
|
|
else:
|
|
values.add(value)
|
|
return values
|
|
|
|
|
|
def _planned_uuid_values(child: dict) -> set[str]:
|
|
apply_payload = child["payload"]["apply_payload"]
|
|
values = {child["id"]}
|
|
for row in apply_payload["claims"] + apply_payload["sources"]:
|
|
values.add(row["id"])
|
|
for row in apply_payload["evidence"]:
|
|
values.update((row["claim_id"], row["source_id"]))
|
|
for row in apply_payload["edges"]:
|
|
values.update((row["from_claim"], row["to_claim"]))
|
|
return values
|
|
|
|
|
|
def _synthetic_readback(fixture: dict, input_receipt: dict, row_ids: dict, child: dict) -> dict:
|
|
claim_rows = []
|
|
evidence_rows = []
|
|
segment_by_id = {segment["id"]: segment for segment in fixture["segments"]}
|
|
for claim in fixture["extraction"]["claims"]:
|
|
claim_id = row_ids["claim_extraction_ids"][claim["claim_key"]]
|
|
claim_rows.append(
|
|
{
|
|
"id": claim_id,
|
|
"source_capture_id": row_ids["source_capture_id"],
|
|
"claim_key": claim["claim_key"],
|
|
"body": claim["body"],
|
|
"metadata": {
|
|
**claim["metadata"],
|
|
"claim_type": claim["type"],
|
|
"confidence": claim["confidence"],
|
|
"text": claim["text"],
|
|
"extractor": input_receipt["extractor"],
|
|
"replay_identity_sha256": input_receipt["replay_identity_sha256"],
|
|
},
|
|
}
|
|
)
|
|
for index, evidence in enumerate(claim["evidence"]):
|
|
segment = segment_by_id[evidence["segment_id"]]
|
|
evidence_rows.append(
|
|
{
|
|
"id": row_ids["evidence_extraction_ids"][f"{claim['claim_key']}:{index}"],
|
|
"claim_extraction_id": claim_id,
|
|
"source_capture_id": row_ids["source_capture_id"],
|
|
"source_segment_id": segment["id"],
|
|
"source_locator": segment["locator"],
|
|
"source_json_pointer": segment["json_pointer"],
|
|
"role": evidence["role"],
|
|
"source_content_sha256": segment["content_sha256"],
|
|
"body": evidence["excerpt"],
|
|
"body_sha256": canary.sha256_bytes(evidence["excerpt"].encode()),
|
|
"metadata": {
|
|
**evidence.get("metadata", {}),
|
|
"source_text_sha256": segment["text_sha256"],
|
|
},
|
|
}
|
|
)
|
|
duplicate_rows = []
|
|
for index, duplicate in enumerate(fixture["extraction"]["duplicates"]):
|
|
duplicate_rows.append(
|
|
{
|
|
"id": row_ids["duplicate_assessment_ids"][str(index)],
|
|
"source_capture_id": row_ids["source_capture_id"],
|
|
"source_segment_id": duplicate["segment_id"],
|
|
"source_locator": duplicate["source_locator"],
|
|
"duplicate_of_claim_key": duplicate["duplicate_of_claim_key"],
|
|
"excerpt": duplicate["excerpt"],
|
|
"excerpt_sha256": canary.sha256_bytes(duplicate["excerpt"].encode()),
|
|
"reason": duplicate["reason"],
|
|
"metadata": {
|
|
"json_pointer": duplicate["source_json_pointer"],
|
|
"source_content_sha256": duplicate["source_content_sha256"],
|
|
"source_text_sha256": duplicate["source_text_sha256"],
|
|
},
|
|
}
|
|
)
|
|
conflict_rows = []
|
|
for index, conflict in enumerate(fixture["extraction"]["conflicts"]):
|
|
conflict_rows.append(
|
|
{
|
|
"id": row_ids["conflict_assessment_ids"][str(index)],
|
|
"source_capture_id": row_ids["source_capture_id"],
|
|
"from_claim_key": conflict["from_claim_key"],
|
|
"to_claim_key": conflict["to_claim_key"],
|
|
"relationship": conflict["relationship"],
|
|
"metadata": {
|
|
key: value
|
|
for key, value in conflict.items()
|
|
if key not in {"from_claim_key", "to_claim_key", "relationship"}
|
|
},
|
|
}
|
|
)
|
|
proposal = {
|
|
**child,
|
|
"reviewed_by_handle": None,
|
|
"reviewed_at": None,
|
|
"applied_by_handle": None,
|
|
"applied_at": None,
|
|
}
|
|
counts = input_receipt["counts"]
|
|
return {
|
|
"source_captures": [
|
|
{
|
|
"id": row_ids["source_capture_id"],
|
|
"source_identity": fixture["source"]["identity"],
|
|
"source_type": fixture["source"]["source_type"],
|
|
"title": fixture["source"]["title"],
|
|
"locator": fixture["source"]["locator"],
|
|
"artifact_path": input_receipt["artifact_path"],
|
|
"artifact_sha256": input_receipt["artifact_sha256"],
|
|
"content_sha256": input_receipt["source_content_sha256"],
|
|
"replay_identity_sha256": input_receipt["replay_identity_sha256"],
|
|
"metadata": {**fixture["source"]["metadata"], "extractor": input_receipt["extractor"]},
|
|
}
|
|
],
|
|
"claim_extractions": claim_rows,
|
|
"evidence_extractions": evidence_rows,
|
|
"duplicate_assessments": duplicate_rows,
|
|
"conflict_assessments": conflict_rows,
|
|
"staged_proposal": proposal,
|
|
"counts": {
|
|
"source_captures": 1,
|
|
"claim_extractions": counts["claim_candidates"],
|
|
"evidence_extractions": counts["evidence_candidates"],
|
|
"duplicate_assessments": counts["duplicate_judgments"],
|
|
"conflict_assessments": counts["conflict_relationships"],
|
|
"staged_proposals": 1,
|
|
},
|
|
}
|
|
|
|
|
|
@pytest.mark.parametrize("scenario_path", canary.DEFAULT_FIXTURES)
|
|
def test_source_artifact_is_separate_hash_bound_and_replayable(scenario_path: Path) -> None:
|
|
fixture, input_receipt, row_ids, parent, child = _loaded(scenario_path)
|
|
artifact_path = Path(input_receipt["artifact_path"])
|
|
|
|
assert input_receipt["artifact_sha256"] == hashlib.sha256(artifact_path.read_bytes()).hexdigest()
|
|
assert input_receipt["artifact_sha256"] != input_receipt["scenario_sha256"]
|
|
assert len(input_receipt["replay_identity_sha256"]) == 64
|
|
assert row_ids == canary.build_row_ids(input_receipt, fixture)
|
|
ingestion = child["payload"]["apply_payload"]["normalization_manifest"]["ingestion_manifest"]
|
|
assert ingestion["artifact_sha256"] == input_receipt["artifact_sha256"]
|
|
assert ingestion["extractor"] == input_receipt["extractor"]
|
|
assert ingestion["replay_identity_sha256"] == input_receipt["replay_identity_sha256"]
|
|
assert ingestion["row_ids"] == row_ids
|
|
assert parent["status"] == child["status"] == "pending_review"
|
|
|
|
|
|
def test_document_and_telegram_candidate_accounting_is_exact() -> None:
|
|
_doc, doc_input, _doc_ids, _doc_parent, doc_child = _loaded(canary.DEFAULT_DOCUMENT_FIXTURE)
|
|
telegram, telegram_input, _telegram_ids, _telegram_parent, telegram_child = _loaded(canary.DEFAULT_TELEGRAM_FIXTURE)
|
|
|
|
assert doc_input["counts"] == {
|
|
"source_segments": 3,
|
|
"claim_candidates": 1,
|
|
"evidence_candidates": 1,
|
|
"duplicate_judgments": 0,
|
|
"conflict_relationships": 0,
|
|
}
|
|
assert telegram_input["counts"] == {
|
|
"source_segments": 3,
|
|
"claim_candidates": 2,
|
|
"evidence_candidates": 3,
|
|
"duplicate_judgments": 1,
|
|
"conflict_relationships": 1,
|
|
}
|
|
doc_payload = doc_child["payload"]["apply_payload"]
|
|
telegram_payload = telegram_child["payload"]["apply_payload"]
|
|
assert {key: len(doc_payload[key]) for key in ("claims", "sources", "evidence", "edges")} == {
|
|
"claims": 1,
|
|
"sources": 2,
|
|
"evidence": 2,
|
|
"edges": 0,
|
|
}
|
|
assert {key: len(telegram_payload[key]) for key in ("claims", "sources", "evidence", "edges")} == {
|
|
"claims": 2,
|
|
"sources": 5,
|
|
"evidence": 5,
|
|
"edges": 1,
|
|
}
|
|
assessment = telegram_payload["normalization_manifest"]["dedupe_conflict_assessment"]
|
|
assert assessment["duplicates"] == telegram["extraction"]["duplicates"]
|
|
assert assessment["conflicts"] == telegram["extraction"]["conflicts"]
|
|
assert telegram_payload["edges"][0]["edge_type"] == "contradicts"
|
|
|
|
|
|
def test_telegram_duplicate_text_keeps_distinct_exact_message_provenance() -> None:
|
|
fixture, _input, _row_ids, _parent, _child = _loaded(canary.DEFAULT_TELEGRAM_FIXTURE)
|
|
first, _second, repeated = fixture["segments"]
|
|
|
|
assert first["body"] == repeated["body"]
|
|
assert first["text_sha256"] == repeated["text_sha256"]
|
|
assert first["content_sha256"] != repeated["content_sha256"]
|
|
assert first["locator"] == "telegram://chat/900001/message/7301"
|
|
assert repeated["locator"] == "telegram://chat/900001/message/7303"
|
|
duplicate = fixture["extraction"]["duplicates"][0]
|
|
assert duplicate["source_locator"] == repeated["locator"]
|
|
assert duplicate["source_json_pointer"] == "jsonl://line/3/message"
|
|
|
|
|
|
def test_fixture_validation_fails_closed_on_invented_evidence(tmp_path: Path) -> None:
|
|
scenario_path = _copy_scenario(tmp_path, canary.DEFAULT_TELEGRAM_FIXTURE)
|
|
scenario = json.loads(scenario_path.read_text(encoding="utf-8"))
|
|
scenario["extraction"]["claims"][0]["evidence"][0]["excerpt"] = "invented evidence"
|
|
scenario_path.write_text(json.dumps(scenario), encoding="utf-8")
|
|
|
|
with pytest.raises(canary.CanaryError, match="not an exact segment substring"):
|
|
canary.load_fixture(scenario_path)
|
|
|
|
|
|
def test_fixture_artifact_path_cannot_escape_scenario_directory(tmp_path: Path) -> None:
|
|
scenario = json.loads(canary.DEFAULT_DOCUMENT_FIXTURE.read_text(encoding="utf-8"))
|
|
scenario["artifact"]["path"] = "../outside.txt"
|
|
path = tmp_path / "scenario.json"
|
|
path.write_text(json.dumps(scenario), encoding="utf-8")
|
|
|
|
with pytest.raises(canary.CanaryError, match="remain inside"):
|
|
canary.load_fixture(path)
|
|
|
|
|
|
def test_capture_sql_escapes_quotes_and_backslashes(tmp_path: Path) -> None:
|
|
scenario_path = _copy_scenario(tmp_path, canary.DEFAULT_DOCUMENT_FIXTURE)
|
|
scenario = json.loads(scenario_path.read_text(encoding="utf-8"))
|
|
scenario["source"]["title"] = "O'Brien\\review"
|
|
scenario["source"]["metadata"]["note"] = "quoted ' value \\ path"
|
|
scenario_path.write_text(json.dumps(scenario), encoding="utf-8")
|
|
fixture, input_receipt, row_ids, _parent, _child = _loaded(scenario_path)
|
|
|
|
sql = canary.build_capture_sql(fixture, input_receipt, row_ids)
|
|
|
|
assert canary.ap.sql_literal(scenario["source"]["title"]) in sql
|
|
expected_metadata = {
|
|
**scenario["source"]["metadata"],
|
|
"extractor": input_receipt["extractor"],
|
|
}
|
|
assert canary.ap.sql_literal(json.dumps(expected_metadata, sort_keys=True)) in sql
|
|
|
|
|
|
def test_replay_identity_and_ids_change_with_extractor_version(tmp_path: Path) -> None:
|
|
scenario_path = _copy_scenario(tmp_path, canary.DEFAULT_DOCUMENT_FIXTURE)
|
|
fixture, original, original_ids, _parent, original_child = _loaded(scenario_path)
|
|
scenario = json.loads(scenario_path.read_text(encoding="utf-8"))
|
|
scenario["extractor"]["version"] = "3"
|
|
scenario_path.write_text(json.dumps(scenario), encoding="utf-8")
|
|
changed_fixture, changed, changed_ids, _changed_parent, changed_child = _loaded(scenario_path)
|
|
|
|
assert original["artifact_sha256"] == changed["artifact_sha256"]
|
|
assert original["replay_identity_sha256"] != changed["replay_identity_sha256"]
|
|
assert original_ids["source_capture_id"] != changed_ids["source_capture_id"]
|
|
assert original_ids["claim_extraction_ids"] != changed_ids["claim_extraction_ids"]
|
|
assert original_ids["parent_proposal_id"] != changed_ids["parent_proposal_id"]
|
|
assert original_child["payload_sha256"] != changed_child["payload_sha256"]
|
|
assert fixture["segments"] == changed_fixture["segments"]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("source_field", "replacement"),
|
|
(
|
|
("identity", "document:mutated-exact-source-identity"),
|
|
("locator", "fixture://working-leo/mutated-exact-source-locator"),
|
|
),
|
|
)
|
|
def test_replay_identity_and_every_row_id_bind_exact_source_identity(
|
|
tmp_path: Path, source_field: str, replacement: str
|
|
) -> None:
|
|
scenario_path = _copy_scenario(tmp_path, canary.DEFAULT_DOCUMENT_FIXTURE)
|
|
fixture, before, before_ids, _parent, before_child = _loaded(scenario_path)
|
|
scenario = json.loads(scenario_path.read_text(encoding="utf-8"))
|
|
scenario["source"][source_field] = replacement
|
|
scenario_path.write_text(json.dumps(scenario), encoding="utf-8")
|
|
|
|
changed_fixture, after, after_ids, _changed_parent, after_child = _loaded(scenario_path)
|
|
|
|
assert before["artifact_sha256"] == after["artifact_sha256"]
|
|
assert before["extraction_spec_sha256"] == after["extraction_spec_sha256"]
|
|
assert before["replay_identity_sha256"] != after["replay_identity_sha256"]
|
|
assert _row_id_values(before_ids).isdisjoint(_row_id_values(after_ids))
|
|
assert _planned_uuid_values(before_child).isdisjoint(_planned_uuid_values(after_child))
|
|
assert fixture["source"][source_field] != changed_fixture["source"][source_field]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("segment_field", "replacement"),
|
|
(
|
|
("id", "message-900001-mutated"),
|
|
("locator", "telegram://chat/900001/message/999999"),
|
|
("json_pointer", "jsonl://line/999/message"),
|
|
("content_sha256", "a" * 64),
|
|
("text_sha256", "b" * 64),
|
|
),
|
|
)
|
|
def test_replay_identity_and_every_row_id_bind_complete_normalized_segment(
|
|
segment_field: str, replacement: str
|
|
) -> None:
|
|
fixture, before, before_ids, _parent, before_child = _loaded(canary.DEFAULT_TELEGRAM_FIXTURE)
|
|
changed_fixture = copy.deepcopy(fixture)
|
|
after = copy.deepcopy(before)
|
|
segment = changed_fixture["segments"][-1]
|
|
original_id = segment["id"]
|
|
segment[segment_field] = replacement
|
|
for claim in changed_fixture["extraction"]["claims"]:
|
|
for evidence in claim["evidence"]:
|
|
if evidence["segment_id"] == original_id:
|
|
evidence["segment_id"] = segment["id"]
|
|
evidence["source_locator"] = segment["locator"]
|
|
evidence["source_json_pointer"] = segment["json_pointer"]
|
|
evidence["source_content_sha256"] = segment["content_sha256"]
|
|
evidence["source_text_sha256"] = segment["text_sha256"]
|
|
for duplicate in changed_fixture["extraction"]["duplicates"]:
|
|
if duplicate["segment_id"] == original_id:
|
|
duplicate["segment_id"] = segment["id"]
|
|
duplicate["source_locator"] = segment["locator"]
|
|
duplicate["source_json_pointer"] = segment["json_pointer"]
|
|
duplicate["source_content_sha256"] = segment["content_sha256"]
|
|
duplicate["source_text_sha256"] = segment["text_sha256"]
|
|
after["source_content_sha256"] = canary.canonical_sha256(
|
|
canary.normalized_segment_manifest(changed_fixture["segments"])
|
|
)
|
|
after["replay_identity_components"] = canary.replay_identity_payload(
|
|
artifact_sha256=after["artifact_sha256"],
|
|
artifact_format=after["artifact_format"],
|
|
source_identity=after["source_identity"],
|
|
source_locator=after["source_locator"],
|
|
normalized_segments_sha256=after["source_content_sha256"],
|
|
extractor=after["extractor"],
|
|
extraction_spec_sha256=after["extraction_spec_sha256"],
|
|
)
|
|
after["replay_identity_sha256"] = canary.canonical_sha256(after["replay_identity_components"])
|
|
after_ids = canary.build_row_ids(after, changed_fixture)
|
|
after_parent = canary.build_parent_packet(changed_fixture, after, after_ids)
|
|
after_child = canary.normalized_stage.prepare_normalized_child(after_parent)
|
|
|
|
assert before["artifact_sha256"] == after["artifact_sha256"]
|
|
assert before["scenario_sha256"] == after["scenario_sha256"]
|
|
assert before["extraction_spec_sha256"] == after["extraction_spec_sha256"]
|
|
assert before["replay_identity_sha256"] != after["replay_identity_sha256"]
|
|
assert _row_id_values(before_ids).isdisjoint(_row_id_values(after_ids))
|
|
assert _planned_uuid_values(before_child).isdisjoint(_planned_uuid_values(after_child))
|
|
|
|
|
|
def test_source_byte_tamper_changes_hashes_and_replay_ids(tmp_path: Path) -> None:
|
|
scenario_path = _copy_scenario(tmp_path, canary.DEFAULT_DOCUMENT_FIXTURE)
|
|
fixture, before, before_ids, _parent, _child = _loaded(scenario_path)
|
|
artifact_path = Path(before["artifact_path"])
|
|
artifact_path.write_text(
|
|
artifact_path.read_text(encoding="utf-8") + "\n\nA new source paragraph.", encoding="utf-8"
|
|
)
|
|
changed_fixture, after, after_ids, _changed_parent, _changed_child = _loaded(scenario_path)
|
|
|
|
assert before["artifact_sha256"] != after["artifact_sha256"]
|
|
assert before["source_content_sha256"] != after["source_content_sha256"]
|
|
assert before["replay_identity_sha256"] != after["replay_identity_sha256"]
|
|
assert before_ids["source_capture_id"] != after_ids["source_capture_id"]
|
|
assert before_ids["claim_extraction_ids"] != after_ids["claim_extraction_ids"]
|
|
assert len(changed_fixture["segments"]) == len(fixture["segments"]) + 1
|
|
|
|
|
|
def test_replay_properties_hold_across_many_version_labels(tmp_path: Path) -> None:
|
|
scenario_path = _copy_scenario(tmp_path, canary.DEFAULT_DOCUMENT_FIXTURE)
|
|
base = json.loads(scenario_path.read_text(encoding="utf-8"))
|
|
observed: set[str] = set()
|
|
for index in range(40):
|
|
scenario = copy.deepcopy(base)
|
|
scenario["extractor"]["version"] = f"property-{index}"
|
|
scenario_path.write_text(json.dumps(scenario), encoding="utf-8")
|
|
fixture_a, receipt_a = canary.load_fixture(scenario_path)
|
|
fixture_b, receipt_b = canary.load_fixture(scenario_path)
|
|
ids_a = canary.build_row_ids(receipt_a, fixture_a)
|
|
ids_b = canary.build_row_ids(receipt_b, fixture_b)
|
|
assert receipt_a == receipt_b
|
|
assert ids_a == ids_b
|
|
observed.add(receipt_a["replay_identity_sha256"])
|
|
assert len(observed) == 40
|
|
|
|
|
|
@pytest.mark.parametrize("scenario_path", canary.DEFAULT_FIXTURES)
|
|
def test_capture_and_stage_sql_do_not_mutate_canonical_tables(scenario_path: Path) -> None:
|
|
fixture, input_receipt, row_ids, parent, child = _loaded(scenario_path)
|
|
sql = "\n".join(
|
|
(
|
|
canary.build_capture_sql(fixture, input_receipt, row_ids),
|
|
canary.normalized_stage.build_stage_sql(child),
|
|
)
|
|
).lower()
|
|
|
|
assert "insert into public." not in sql
|
|
assert "update public." not in sql
|
|
assert "delete from public." not in sql
|
|
assert "insert into kb_canary.source_captures" in sql
|
|
assert "insert into kb_stage.kb_proposals" in sql
|
|
assert parent["payload"]["ingestion_manifest"]["row_ids"] == row_ids
|
|
|
|
|
|
def test_canonical_snapshot_covers_all_nonempty_canonical_tables() -> None:
|
|
schema_sql = canary.build_schema_sql().lower()
|
|
snapshot_sql = canary.build_canonical_snapshot_sql().lower()
|
|
|
|
for table in ("claims", "sources", "claim_evidence", "claim_edges"):
|
|
assert f"insert into public.{table}" in schema_sql
|
|
assert f"from public.{table}" in snapshot_sql
|
|
assert "order by" in snapshot_sql
|
|
assert "begin transaction read only" in snapshot_sql
|
|
assert canary.canonical_snapshot_complete({}) is False
|
|
assert (
|
|
canary.canonical_snapshot_complete(
|
|
{
|
|
"claims": [{"id": "1"}],
|
|
"sources": [{"id": "2"}],
|
|
"claim_evidence": [{"claim_id": "1"}],
|
|
"claim_edges": [{"from_claim": "1"}],
|
|
}
|
|
)
|
|
is True
|
|
)
|
|
|
|
|
|
def test_evaluation_fails_when_exact_evidence_locator_is_changed() -> None:
|
|
fixture, input_receipt, row_ids, _parent, child = _loaded(canary.DEFAULT_TELEGRAM_FIXTURE)
|
|
readback = _synthetic_readback(fixture, input_receipt, row_ids, child)
|
|
canonical = {
|
|
"claims": [{"id": "1"}],
|
|
"sources": [{"id": "2"}],
|
|
"claim_evidence": [{"claim_id": "1"}],
|
|
"claim_edges": [{"from_claim": "1"}],
|
|
}
|
|
|
|
checks = canary.evaluate_checks(fixture, input_receipt, row_ids, readback, canonical, copy.deepcopy(canonical))
|
|
assert all(checks.values())
|
|
broken = copy.deepcopy(readback)
|
|
broken["evidence_extractions"][0]["source_locator"] = "telegram://chat/900001/message/9999"
|
|
assert (
|
|
canary.evaluate_checks(fixture, input_receipt, row_ids, broken, canonical, copy.deepcopy(canonical))[
|
|
"all_evidence_has_exact_source_location"
|
|
]
|
|
is False
|
|
)
|
|
|
|
|
|
def test_evaluation_recomputes_instead_of_trusting_supplied_row_ids() -> None:
|
|
fixture, input_receipt, row_ids, _parent, _child = _loaded(canary.DEFAULT_TELEGRAM_FIXTURE)
|
|
tampered_ids = copy.deepcopy(row_ids)
|
|
tampered_ids["source_capture_id"] = "00000000-0000-4000-8000-000000009990"
|
|
tampered_parent = canary.build_parent_packet(fixture, input_receipt, tampered_ids)
|
|
tampered_child = canary.normalized_stage.prepare_normalized_child(tampered_parent)
|
|
readback = _synthetic_readback(fixture, input_receipt, tampered_ids, tampered_child)
|
|
|
|
checks = canary.evaluate_checks(
|
|
fixture, input_receipt, tampered_ids, readback, _canonical_snapshot(), _canonical_snapshot()
|
|
)
|
|
|
|
assert checks["deterministic_row_ids_recomputed_exact"] is False
|
|
assert checks["source_capture_identity_exact"] is False
|
|
assert checks["planned_proposal_identities_and_linkages_exact"] is False
|
|
|
|
|
|
def test_independent_graph_oracle_rejects_consistent_generator_edge_corruption(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
fixture, input_receipt = canary.load_fixture(canary.DEFAULT_TELEGRAM_FIXTURE)
|
|
row_ids = canary.build_row_ids(input_receipt, fixture)
|
|
parent = canary.build_parent_packet(fixture, input_receipt, row_ids)
|
|
original_prepare = canary.normalized_stage.prepare_normalized_child
|
|
|
|
def corrupted_prepare(parent_packet: dict) -> dict:
|
|
child = copy.deepcopy(original_prepare(parent_packet))
|
|
child["payload"]["apply_payload"]["edges"][0]["to_claim"] = "00000000-0000-4000-8000-000000009996"
|
|
return child
|
|
|
|
monkeypatch.setattr(canary.normalized_stage, "prepare_normalized_child", corrupted_prepare)
|
|
corrupted_child = corrupted_prepare(parent)
|
|
readback = _synthetic_readback(fixture, input_receipt, row_ids, corrupted_child)
|
|
|
|
checks = canary.evaluate_checks(
|
|
fixture, input_receipt, row_ids, readback, _canonical_snapshot(), _canonical_snapshot()
|
|
)
|
|
|
|
assert checks["planned_graph_matches_independent_source_oracle"] is False
|
|
assert checks["planned_proposal_identities_and_linkages_exact"] is False
|
|
assert checks["conflicts_and_relationships_persisted"] is False
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("mutation", "failed_check"),
|
|
(
|
|
("source_id", "source_capture_identity_exact"),
|
|
("source_identity", "source_capture_identity_exact"),
|
|
("source_locator", "source_capture_identity_exact"),
|
|
("claim_id", "claim_extraction_identities_and_fks_exact"),
|
|
("claim_source_fk", "claim_extraction_identities_and_fks_exact"),
|
|
("evidence_id", "evidence_extraction_identities_and_fks_exact"),
|
|
("evidence_claim_fk", "evidence_extraction_identities_and_fks_exact"),
|
|
("evidence_source_fk", "evidence_extraction_identities_and_fks_exact"),
|
|
("evidence_segment_id", "evidence_extraction_identities_and_fks_exact"),
|
|
("evidence_json_pointer", "evidence_extraction_identities_and_fks_exact"),
|
|
("evidence_role", "evidence_extraction_identities_and_fks_exact"),
|
|
("evidence_body_sha256", "evidence_extraction_identities_and_fks_exact"),
|
|
("proposal_id", "planned_proposal_identities_and_linkages_exact"),
|
|
("planned_claim_id", "planned_proposal_identities_and_linkages_exact"),
|
|
("planned_source_id", "planned_proposal_identities_and_linkages_exact"),
|
|
("planned_evidence_claim_fk", "planned_proposal_identities_and_linkages_exact"),
|
|
("planned_evidence_source_fk", "planned_proposal_identities_and_linkages_exact"),
|
|
("planned_evidence_role", "planned_proposal_identities_and_linkages_exact"),
|
|
("manifest_row_id", "planned_proposal_identities_and_linkages_exact"),
|
|
),
|
|
)
|
|
def test_evaluation_rejects_wrong_deterministic_ids_and_every_fk_linkage(mutation: str, failed_check: str) -> None:
|
|
fixture, input_receipt, row_ids, _parent, child = _loaded(canary.DEFAULT_TELEGRAM_FIXTURE)
|
|
readback = _synthetic_readback(fixture, input_receipt, row_ids, child)
|
|
broken = copy.deepcopy(readback)
|
|
wrong_uuid = "00000000-0000-4000-8000-000000009999"
|
|
apply_payload = broken["staged_proposal"]["payload"]["apply_payload"]
|
|
if mutation == "source_id":
|
|
broken["source_captures"][0]["id"] = wrong_uuid
|
|
elif mutation == "source_identity":
|
|
broken["source_captures"][0]["source_identity"] = "fixture:wrong-source-identity"
|
|
elif mutation == "source_locator":
|
|
broken["source_captures"][0]["locator"] = "fixture://working-leo/wrong-source-locator"
|
|
elif mutation == "claim_id":
|
|
broken["claim_extractions"][0]["id"] = broken["claim_extractions"][1]["id"]
|
|
elif mutation == "claim_source_fk":
|
|
broken["claim_extractions"][0]["source_capture_id"] = wrong_uuid
|
|
elif mutation == "evidence_id":
|
|
broken["evidence_extractions"][0]["id"] = broken["evidence_extractions"][1]["id"]
|
|
elif mutation == "evidence_claim_fk":
|
|
broken["evidence_extractions"][0]["claim_extraction_id"] = broken["claim_extractions"][1]["id"]
|
|
elif mutation == "evidence_source_fk":
|
|
broken["evidence_extractions"][0]["source_capture_id"] = wrong_uuid
|
|
elif mutation == "evidence_segment_id":
|
|
broken["evidence_extractions"][0]["source_segment_id"] = "message-900001-9999"
|
|
elif mutation == "evidence_json_pointer":
|
|
broken["evidence_extractions"][0]["source_json_pointer"] = "jsonl://line/999/message"
|
|
elif mutation == "evidence_role":
|
|
broken["evidence_extractions"][0]["role"] = "supports"
|
|
elif mutation == "evidence_body_sha256":
|
|
broken["evidence_extractions"][0]["body_sha256"] = "d" * 64
|
|
elif mutation == "proposal_id":
|
|
broken["staged_proposal"]["id"] = wrong_uuid
|
|
elif mutation == "planned_claim_id":
|
|
apply_payload["claims"][0]["id"] = apply_payload["claims"][1]["id"]
|
|
elif mutation == "planned_source_id":
|
|
apply_payload["sources"][0]["id"] = apply_payload["sources"][1]["id"]
|
|
elif mutation == "planned_evidence_claim_fk":
|
|
apply_payload["evidence"][0]["claim_id"] = apply_payload["claims"][1]["id"]
|
|
elif mutation == "planned_evidence_source_fk":
|
|
apply_payload["evidence"][0]["source_id"] = apply_payload["sources"][1]["id"]
|
|
elif mutation == "planned_evidence_role":
|
|
apply_payload["evidence"][0]["role"] = "supports"
|
|
elif mutation == "manifest_row_id":
|
|
apply_payload["normalization_manifest"]["ingestion_manifest"]["row_ids"]["source_capture_id"] = wrong_uuid
|
|
|
|
checks = canary.evaluate_checks(
|
|
fixture, input_receipt, row_ids, broken, _canonical_snapshot(), _canonical_snapshot()
|
|
)
|
|
|
|
assert checks[failed_check] is False
|
|
assert not all(checks.values())
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("surface", "field", "replacement", "failed_check"),
|
|
(
|
|
("duplicate", "id", "00000000-0000-4000-8000-000000009991", "duplicate_judgments_persisted"),
|
|
(
|
|
"duplicate",
|
|
"source_capture_id",
|
|
"00000000-0000-4000-8000-000000009992",
|
|
"duplicate_judgments_persisted",
|
|
),
|
|
("duplicate", "source_segment_id", "message-900001-9999", "duplicate_judgments_persisted"),
|
|
(
|
|
"duplicate",
|
|
"source_locator",
|
|
"telegram://chat/900001/message/9999",
|
|
"duplicate_judgments_persisted",
|
|
),
|
|
("duplicate", "duplicate_of_claim_key", "approval_alone_makes_canonical", "duplicate_judgments_persisted"),
|
|
("duplicate", "excerpt", "fabricated duplicate excerpt", "duplicate_judgments_persisted"),
|
|
("duplicate", "excerpt_sha256", "c" * 64, "duplicate_judgments_persisted"),
|
|
("duplicate", "reason", "fabricated duplicate reason", "duplicate_judgments_persisted"),
|
|
("duplicate", "metadata", {"json_pointer": "fabricated"}, "duplicate_judgments_persisted"),
|
|
("conflict", "id", "00000000-0000-4000-8000-000000009993", "conflicts_and_relationships_persisted"),
|
|
(
|
|
"conflict",
|
|
"source_capture_id",
|
|
"00000000-0000-4000-8000-000000009994",
|
|
"conflicts_and_relationships_persisted",
|
|
),
|
|
("conflict", "from_claim_key", "approval_alone_makes_canonical", "conflicts_and_relationships_persisted"),
|
|
(
|
|
"conflict",
|
|
"to_claim_key",
|
|
"pending_review_does_not_change_canonical",
|
|
"conflicts_and_relationships_persisted",
|
|
),
|
|
("conflict", "relationship", "supports", "conflicts_and_relationships_persisted"),
|
|
("conflict", "metadata", {"reason": "fabricated"}, "conflicts_and_relationships_persisted"),
|
|
("edge", "from_claim", "00000000-0000-4000-8000-000000009995", "conflicts_and_relationships_persisted"),
|
|
("edge", "to_claim", "00000000-0000-4000-8000-000000009996", "conflicts_and_relationships_persisted"),
|
|
("edge", "edge_type", "supports", "conflicts_and_relationships_persisted"),
|
|
("edge", "weight", 0.5, "conflicts_and_relationships_persisted"),
|
|
("edge", "created_by", "00000000-0000-4000-8000-000000009997", "conflicts_and_relationships_persisted"),
|
|
),
|
|
)
|
|
def test_evaluation_rejects_mutated_duplicate_conflict_and_edge_identities(
|
|
surface: str, field: str, replacement: object, failed_check: str
|
|
) -> None:
|
|
fixture, input_receipt, row_ids, _parent, child = _loaded(canary.DEFAULT_TELEGRAM_FIXTURE)
|
|
broken = _synthetic_readback(fixture, input_receipt, row_ids, child)
|
|
if surface == "duplicate":
|
|
broken["duplicate_assessments"][0][field] = replacement
|
|
elif surface == "conflict":
|
|
broken["conflict_assessments"][0][field] = replacement
|
|
else:
|
|
broken["staged_proposal"]["payload"]["apply_payload"]["edges"][0][field] = replacement
|
|
|
|
checks = canary.evaluate_checks(
|
|
fixture, input_receipt, row_ids, broken, _canonical_snapshot(), _canonical_snapshot()
|
|
)
|
|
|
|
assert checks[failed_check] is False
|
|
assert not all(checks.values())
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("field", "replacement"),
|
|
(
|
|
("reviewed_by_handle", "reviewer"),
|
|
("reviewed_by_agent_id", "00000000-0000-4000-8000-000000009981"),
|
|
("reviewed_at", "2026-07-15T00:00:00+00:00"),
|
|
("review_note", "fabricated review note"),
|
|
("applied_by_handle", "applier"),
|
|
("applied_by_agent_id", "00000000-0000-4000-8000-000000009982"),
|
|
("applied_at", "2026-07-15T00:00:01+00:00"),
|
|
),
|
|
)
|
|
def test_evaluation_rejects_every_review_and_apply_metadata_mutation(field: str, replacement: str) -> None:
|
|
fixture, input_receipt, row_ids, _parent, child = _loaded(canary.DEFAULT_TELEGRAM_FIXTURE)
|
|
broken = _synthetic_readback(fixture, input_receipt, row_ids, child)
|
|
broken["staged_proposal"][field] = replacement
|
|
|
|
checks = canary.evaluate_checks(
|
|
fixture, input_receipt, row_ids, broken, _canonical_snapshot(), _canonical_snapshot()
|
|
)
|
|
|
|
assert checks["no_review_or_apply_metadata"] is False
|
|
assert checks["planned_proposal_identities_and_linkages_exact"] is False
|
|
|
|
|
|
def _suite_child(artifact_format: str, *, passed: bool = True, canonical_unchanged: bool = True) -> dict:
|
|
before = "a" * 64
|
|
after = before if canonical_unchanged else "b" * 64
|
|
return {
|
|
"status": "pass" if passed else "fail",
|
|
"input": {"artifact_format": artifact_format},
|
|
"canonical": {
|
|
"fingerprint_before": before,
|
|
"fingerprint_after": after,
|
|
"unchanged": canonical_unchanged,
|
|
},
|
|
"checks": {"canonical_fingerprint_unchanged": canonical_unchanged},
|
|
}
|
|
|
|
|
|
def test_single_fixture_suite_is_truthfully_partial(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
|
monkeypatch.setattr(canary, "run_canary", lambda _path: _suite_child("plain_text"))
|
|
|
|
suite = canary.run_suite([tmp_path / "document.scenario.json"])
|
|
|
|
assert suite["status"] == "partial"
|
|
assert suite["full_suite_passed"] is False
|
|
assert suite["coverage_status"] == "partial"
|
|
assert suite["missing_required_coverage"] == ["social_conversation"]
|
|
|
|
|
|
def test_malformed_fixture_fails_closed_with_receipt_and_without_runtime(tmp_path: Path) -> None:
|
|
malformed = tmp_path / "malformed.scenario.json"
|
|
malformed.write_text("{not-json", encoding="utf-8")
|
|
|
|
receipt = canary.run_canary(malformed)
|
|
suite = canary.run_suite([malformed])
|
|
|
|
assert receipt["status"] == "fail"
|
|
assert receipt["error"]["type"] == "CanaryError"
|
|
assert receipt["cleanup"]["runtime_not_started"] is True
|
|
assert receipt["cleanup"]["container_absent"] is True
|
|
assert suite["status"] == "fail"
|
|
assert suite["receipts"][0]["error"]["type"] == "CanaryError"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("children", "expected_status"),
|
|
(
|
|
(
|
|
[_suite_child("plain_text"), _suite_child("telegram_jsonl")],
|
|
"pass",
|
|
),
|
|
(
|
|
[_suite_child("plain_text", canonical_unchanged=False), _suite_child("telegram_jsonl")],
|
|
"fail",
|
|
),
|
|
(
|
|
[_suite_child("plain_text"), _suite_child("telegram_jsonl", passed=False)],
|
|
"fail",
|
|
),
|
|
),
|
|
)
|
|
def test_full_suite_status_requires_both_shapes_all_children_and_canonical_invariance(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path: Path, children: list[dict], expected_status: str
|
|
) -> None:
|
|
queue = iter(children)
|
|
monkeypatch.setattr(canary, "run_canary", lambda _path: next(queue))
|
|
|
|
suite = canary.run_suite([tmp_path / "document.scenario.json", tmp_path / "telegram.scenario.json"])
|
|
|
|
assert suite["status"] == expected_status
|
|
assert suite["full_suite_passed"] is (expected_status == "pass")
|
|
|
|
|
|
def test_suite_recomputes_canonical_invariance_instead_of_trusting_stale_boolean(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
|
) -> None:
|
|
children = [_suite_child("plain_text"), _suite_child("telegram_jsonl")]
|
|
children[0]["canonical"]["fingerprint_after"] = "c" * 64
|
|
queue = iter(children)
|
|
monkeypatch.setattr(canary, "run_canary", lambda _path: next(queue))
|
|
|
|
suite = canary.run_suite([tmp_path / "document.scenario.json", tmp_path / "telegram.scenario.json"])
|
|
|
|
assert suite["status"] == "fail"
|
|
assert suite["canonical_fingerprint_invariant_all"] is False
|
|
assert suite["full_suite_passed"] is False
|
|
assert "canonical_fingerprint_invariance" in suite["missing_required_coverage"]
|