diff --git a/scripts/run_leo_local_ingestion_proposal_canary.py b/scripts/run_leo_local_ingestion_proposal_canary.py index 3cb9243..12854b2 100644 --- a/scripts/run_leo_local_ingestion_proposal_canary.py +++ b/scripts/run_leo_local_ingestion_proposal_canary.py @@ -31,7 +31,7 @@ import apply_proposal as ap # noqa: E402 import stage_normalized_proposal as normalized_stage # noqa: E402 SCHEMA = "livingip.workingLeoLocalIngestionProposalCanary.v2" -SUITE_SCHEMA = "livingip.workingLeoLocalIngestionProposalSuite.v1" +SUITE_SCHEMA = "livingip.workingLeoLocalIngestionProposalSuite.v2" FIXTURE_SCHEMA = "livingip.workingLeoSourceIngestionScenario.v2" DEFAULT_DOCUMENT_FIXTURE = REPO_ROOT / "fixtures" / "working-leo" / "document-ingestion-v2.scenario.json" DEFAULT_TELEGRAM_FIXTURE = REPO_ROOT / "fixtures" / "working-leo" / "telegram-transcript-ingestion-v1.scenario.json" @@ -60,6 +60,42 @@ def canonical_sha256(value: Any) -> str: return sha256_bytes(canonical_json_bytes(value)) +def normalized_segment_manifest(segments: list[dict[str, Any]]) -> list[dict[str, Any]]: + """Return the complete normalized provenance identity for replay hashing.""" + return [ + { + "id": segment["id"], + "locator": segment["locator"], + "json_pointer": segment["json_pointer"], + "content_sha256": segment["content_sha256"], + "text_sha256": segment["text_sha256"], + } + for segment in segments + ] + + +def replay_identity_payload( + *, + artifact_sha256: str, + artifact_format: str, + source_identity: str, + source_locator: str, + normalized_segments_sha256: str, + extractor: dict[str, str], + extraction_spec_sha256: str, +) -> dict[str, Any]: + """Build the explicit source-to-extraction identity used by every row ID.""" + return { + "artifact_sha256": artifact_sha256, + "artifact_format": artifact_format, + "source_identity": source_identity, + "source_locator": source_locator, + "normalized_segments_sha256": normalized_segments_sha256, + "extractor": extractor, + "extraction_spec_sha256": extraction_spec_sha256, + } + + def canonical_snapshot_complete(snapshot: dict[str, Any]) -> bool: return set(snapshot) == set(CANONICAL_TABLES) and all( isinstance(snapshot.get(table), list) and bool(snapshot[table]) for table in CANONICAL_TABLES @@ -231,6 +267,9 @@ def load_fixture(path: Path) -> tuple[dict[str, Any], dict[str, Any]]: raise CanaryError(f"source.source_type must be one of {sorted(ap.SOURCE_TYPES)}") if not isinstance(source.get("metadata"), dict): raise CanaryError("source.metadata must be an object") + source_identity = _require_text(source.get("identity"), "source.identity") + source_locator = _require_text(source.get("locator"), "source.locator") + source = {**source, "identity": source_identity, "locator": source_locator} segments = _load_segments(raw, artifact_format, source) segment_by_id = {segment["id"]: segment for segment in segments} @@ -349,28 +388,25 @@ def load_fixture(path: Path) -> tuple[dict[str, Any], dict[str, Any]]: normalized_conflicts.append({**conflict, "from_claim_key": from_key, "to_claim_key": to_key}) artifact_sha256 = sha256_bytes(raw) - source_content_sha256 = canonical_sha256( - [ - { - "id": segment["id"], - "locator": segment["locator"], - "content_sha256": segment["content_sha256"], - } - for segment in segments - ] - ) + segment_manifest = normalized_segment_manifest(segments) + source_content_sha256 = canonical_sha256(segment_manifest) extraction_spec_sha256 = canonical_sha256(extraction) - replay_identity_sha256 = canonical_sha256( - { - "artifact_sha256": artifact_sha256, - "extractor": {"name": extractor_name, "version": extractor_version}, - "extraction_spec_sha256": extraction_spec_sha256, - } + extractor_identity = {"name": extractor_name, "version": extractor_version} + replay_components = replay_identity_payload( + artifact_sha256=artifact_sha256, + artifact_format=artifact_format, + source_identity=source_identity, + source_locator=source_locator, + normalized_segments_sha256=source_content_sha256, + extractor=extractor_identity, + extraction_spec_sha256=extraction_spec_sha256, ) + replay_identity_sha256 = canonical_sha256(replay_components) fixture = { **scenario, "artifact": {**artifact, "resolved_path": str(artifact_path)}, - "extractor": {"name": extractor_name, "version": extractor_version}, + "source": source, + "extractor": extractor_identity, "segments": segments, "extraction": { "claims": normalized_claims, @@ -385,10 +421,11 @@ def load_fixture(path: Path) -> tuple[dict[str, Any], dict[str, Any]]: "artifact_format": artifact_format, "artifact_sha256": artifact_sha256, "source_content_sha256": source_content_sha256, - "source_identity": source["identity"], - "source_locator": source["locator"], - "extractor": {"name": extractor_name, "version": extractor_version}, + "source_identity": source_identity, + "source_locator": source_locator, + "extractor": extractor_identity, "extraction_spec_sha256": extraction_spec_sha256, + "replay_identity_components": replay_components, "replay_identity_sha256": replay_identity_sha256, "counts": { "source_segments": len(segments), @@ -421,7 +458,7 @@ def build_row_ids(input_receipt: dict[str, Any], fixture: dict[str, Any]) -> dic for index, _conflict in enumerate(fixture["extraction"]["conflicts"]) } return { - "source_capture_id": stable_uuid(input_receipt["artifact_sha256"], "source-capture"), + "source_capture_id": stable_uuid(replay_seed, "source-capture"), "claim_extraction_ids": claim_ids, "evidence_extraction_ids": evidence_ids, "duplicate_assessment_ids": duplicate_ids, @@ -507,17 +544,9 @@ def build_parent_packet( "source_locator": input_receipt["source_locator"], "extractor": input_receipt["extractor"], "extraction_spec_sha256": input_receipt["extraction_spec_sha256"], + "replay_identity_components": input_receipt["replay_identity_components"], "replay_identity_sha256": input_receipt["replay_identity_sha256"], - "segments": [ - { - "id": segment["id"], - "locator": segment["locator"], - "json_pointer": segment["json_pointer"], - "content_sha256": segment["content_sha256"], - "text_sha256": segment["text_sha256"], - } - for segment in fixture["segments"] - ], + "segments": normalized_segment_manifest(fixture["segments"]), "row_ids": row_ids, "candidate_counts": input_receipt["counts"], } @@ -547,6 +576,100 @@ def build_parent_packet( } +def _independent_planned_uuid(parent_proposal_id: str, kind: str, key: str) -> str: + """Reproduce the public deterministic-ID contract without calling the planner.""" + return str(uuid.uuid5(uuid.NAMESPACE_URL, f"teleo:kb-rich-contract:{parent_proposal_id}:{kind}:{key}")) + + +def expected_planned_graph_identity(fixture: dict[str, Any], row_ids: dict[str, Any]) -> dict[str, Any]: + """Derive planned row IDs and FK identities directly from normalized source candidates.""" + parent_id = row_ids["parent_proposal_id"] + claims = fixture["extraction"]["claims"] + claim_ids = { + claim["claim_key"]: _independent_planned_uuid(parent_id, "claim", claim["claim_key"]) for claim in claims + } + referenced_segment_ids: list[str] = [] + for claim in claims: + for evidence in claim["evidence"]: + if evidence["segment_id"] not in referenced_segment_ids: + referenced_segment_ids.append(evidence["segment_id"]) + for duplicate in fixture["extraction"]["duplicates"]: + if duplicate["segment_id"] not in referenced_segment_ids: + referenced_segment_ids.append(duplicate["segment_id"]) + + segment_source_ids = { + segment_id: _independent_planned_uuid( + parent_id, + "source", + _segment_source_key(fixture["source"]["source_key"], segment_id), + ) + for segment_id in referenced_segment_ids + } + body_source_ids = { + claim["claim_key"]: _independent_planned_uuid(parent_id, "claim-body-source", claim["claim_key"]) + for claim in claims + } + + evidence_rows: list[dict[str, Any]] = [] + seen_evidence: set[tuple[str, str, str]] = set() + + def append_evidence(claim_id: str, source_id: str, role: str) -> None: + key = (claim_id, source_id, role) + if key in seen_evidence: + return + seen_evidence.add(key) + evidence_rows.append( + { + "claim_id": claim_id, + "source_id": source_id, + "role": role, + "weight": None, + "created_by": None, + } + ) + + for claim in claims: + claim_key = claim["claim_key"] + claim_id = claim_ids[claim_key] + append_evidence(claim_id, body_source_ids[claim_key], "illustrates") + for evidence in claim["evidence"]: + append_evidence(claim_id, segment_source_ids[evidence["segment_id"]], evidence["role"]) + + edge_rows: list[dict[str, Any]] = [] + seen_edges: set[tuple[str, str, str]] = set() + for claim in claims: + from_claim = claim_ids[claim["claim_key"]] + for edge in claim["edges"]: + target_key = _require_text(edge.get("target"), f"claim {claim['claim_key']} edge target") + to_claim = claim_ids.get(target_key) + if to_claim is None: + raise CanaryError(f"claim {claim['claim_key']} edge references unknown claim {target_key}") + edge_type = _require_text(edge.get("edge_type"), f"claim {claim['claim_key']} edge type") + identity = (from_claim, to_claim, edge_type) + if identity in seen_edges: + continue + seen_edges.add(identity) + edge_rows.append( + { + "from_claim": from_claim, + "to_claim": to_claim, + "edge_type": edge_type, + "weight": edge.get("weight"), + "created_by": None, + } + ) + + return { + "planned_claim_ids": [claim_ids[claim["claim_key"]] for claim in claims], + "planned_source_ids": [ + *(segment_source_ids[segment_id] for segment_id in referenced_segment_ids), + *(body_source_ids[claim["claim_key"]] for claim in claims), + ], + "planned_evidence_rows": evidence_rows, + "planned_edge_rows": edge_rows, + } + + def build_schema_sql() -> str: return rf"""\set ON_ERROR_STOP on create schema kb_canary; @@ -664,6 +787,116 @@ rollback; """ +def expected_persisted_rows( + fixture: dict[str, Any], input_receipt: dict[str, Any], row_ids: dict[str, Any] +) -> dict[str, list[dict[str, Any]]]: + """Project the complete relational identity expected from one fixture.""" + source = fixture["source"] + expected: dict[str, list[dict[str, Any]]] = { + "source_captures": [ + { + "id": row_ids["source_capture_id"], + "source_identity": source["identity"], + "source_type": source["source_type"], + "title": source["title"], + "locator": 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": {**source["metadata"], "extractor": input_receipt["extractor"]}, + } + ], + "claim_extractions": [], + "evidence_extractions": [], + "duplicate_assessments": [], + "conflict_assessments": [], + } + 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"]] + expected["claim_extractions"].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"]): + evidence_key = f"{claim['claim_key']}:{index}" + segment = segment_by_id[evidence["segment_id"]] + expected["evidence_extractions"].append( + { + "id": row_ids["evidence_extraction_ids"][evidence_key], + "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"], + "body": evidence["excerpt"], + "body_sha256": sha256_bytes(evidence["excerpt"].encode()), + "source_content_sha256": segment["content_sha256"], + "metadata": { + **_require_object(evidence.get("metadata", {}), f"evidence {evidence_key}.metadata"), + "source_text_sha256": segment["text_sha256"], + }, + } + ) + for index, duplicate in enumerate(fixture["extraction"]["duplicates"]): + expected["duplicate_assessments"].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": 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"], + }, + } + ) + for index, conflict in enumerate(fixture["extraction"]["conflicts"]): + expected["conflict_assessments"].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"} + }, + } + ) + return expected + + +def rows_exact(actual: Any, expected: list[dict[str, Any]]) -> bool: + """Compare complete row identities independent of database result order.""" + return ( + isinstance(actual, list) + and all(isinstance(row, dict) for row in actual) + and sorted(actual, key=canonical_json_bytes) == sorted(expected, key=canonical_json_bytes) + ) + + def build_capture_sql(fixture: dict[str, Any], input_receipt: dict[str, Any], row_ids: dict[str, Any]) -> str: source = fixture["source"] q = ap.sql_literal @@ -869,41 +1102,88 @@ class DockerPostgres: } -def proposal_links(readback: dict[str, Any], input_receipt: dict[str, Any]) -> dict[str, Any]: - proposal = readback["staged_proposal"] - apply_payload = proposal["payload"]["apply_payload"] - source_rows = apply_payload["sources"] - evidence_rows = apply_payload["evidence"] +def proposal_readback_parts( + readback: dict[str, Any], +) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any], dict[str, Any], dict[str, Any]]: + """Extract typed proposal layers so every verifier path fails closed consistently.""" + proposal_value = readback.get("staged_proposal") + proposal = proposal_value if isinstance(proposal_value, dict) else {} + payload_value = proposal.get("payload") + payload = payload_value if isinstance(payload_value, dict) else {} + apply_value = payload.get("apply_payload") + apply_payload = apply_value if isinstance(apply_value, dict) else {} + manifest_value = apply_payload.get("normalization_manifest") + manifest = manifest_value if isinstance(manifest_value, dict) else {} + ingestion_value = manifest.get("ingestion_manifest") + ingestion_manifest = ingestion_value if isinstance(ingestion_value, dict) else {} + assessment_value = manifest.get("dedupe_conflict_assessment") + assessment = assessment_value if isinstance(assessment_value, dict) else {} + return proposal, apply_payload, manifest, ingestion_manifest, assessment + + +def proposal_links(readback: dict[str, Any]) -> dict[str, Any]: + """Project observed proposal row identities and links without judging them.""" + proposal, apply_payload, _manifest, ingestion_manifest, _assessment = proposal_readback_parts(readback) + claim_rows = apply_payload.get("claims") if isinstance(apply_payload.get("claims"), list) else [] + source_rows = apply_payload.get("sources") if isinstance(apply_payload.get("sources"), list) else [] + evidence_rows = apply_payload.get("evidence") if isinstance(apply_payload.get("evidence"), list) else [] + edge_rows = apply_payload.get("edges") if isinstance(apply_payload.get("edges"), list) else [] + planned_evidence_rows = [ + { + "claim_id": row.get("claim_id"), + "source_id": row.get("source_id"), + "role": row.get("role"), + "weight": row.get("weight"), + "created_by": row.get("created_by"), + } + for row in evidence_rows + if isinstance(row, dict) + ] + planned_edge_rows = [ + { + "from_claim": row.get("from_claim"), + "to_claim": row.get("to_claim"), + "edge_type": row.get("edge_type"), + "weight": row.get("weight"), + "created_by": row.get("created_by"), + } + for row in edge_rows + if isinstance(row, dict) + ] return { - "proposal_id": proposal["id"], - "proposal_status": proposal["status"], - "proposal_source_ref": proposal["source_ref"], - "planned_claim_ids": [row["id"] for row in apply_payload["claims"]], - "planned_source_ids": [row["id"] for row in source_rows], + "proposal_id": proposal.get("id"), + "proposal_status": proposal.get("status"), + "proposal_source_ref": proposal.get("source_ref"), + "planned_claim_ids": [row.get("id") for row in claim_rows if isinstance(row, dict)], + "planned_source_ids": [row.get("id") for row in source_rows if isinstance(row, dict)], "planned_evidence_keys": [ - {"claim_id": row["claim_id"], "source_id": row["source_id"], "role": row["role"]} for row in evidence_rows + {key: row[key] for key in ("claim_id", "source_id", "role")} for row in planned_evidence_rows ], "planned_edge_keys": [ - { - "from_claim": row["from_claim"], - "to_claim": row["to_claim"], - "edge_type": row["edge_type"], - } - for row in apply_payload["edges"] + {key: row[key] for key in ("from_claim", "to_claim", "edge_type")} for row in planned_edge_rows ], + "planned_evidence_rows": planned_evidence_rows, + "planned_edge_rows": planned_edge_rows, "planned_counts": { - "claims": len(apply_payload["claims"]), + "claims": len(claim_rows), "sources": len(source_rows), "evidence": len(evidence_rows), - "edges": len(apply_payload["edges"]), + "edges": len(edge_rows), }, - "input_hash_in_manifest": ( - apply_payload["normalization_manifest"].get("ingestion_manifest", {}).get("artifact_sha256") - == input_receipt["artifact_sha256"] - ), + "manifest_artifact_sha256": ingestion_manifest.get("artifact_sha256"), + "manifest_replay_identity_sha256": ingestion_manifest.get("replay_identity_sha256"), + "manifest_row_ids": ingestion_manifest.get("row_ids"), + } + + +def proposal_link_proof(readback: dict[str, Any], input_receipt: dict[str, Any]) -> dict[str, Any]: + """Attach explicit comparisons to the pure observed link projection for receipts.""" + links = proposal_links(readback) + return { + **links, + "input_hash_in_manifest": links["manifest_artifact_sha256"] == input_receipt["artifact_sha256"], "replay_identity_in_manifest": ( - apply_payload["normalization_manifest"].get("ingestion_manifest", {}).get("replay_identity_sha256") - == input_receipt["replay_identity_sha256"] + links["manifest_replay_identity_sha256"] == input_receipt["replay_identity_sha256"] ), } @@ -916,17 +1196,43 @@ def evaluate_checks( canonical_before: dict[str, Any], canonical_after: dict[str, Any], ) -> dict[str, bool]: - source_rows = readback.get("source_captures") or [] - claim_rows = readback.get("claim_extractions") or [] - evidence_rows = readback.get("evidence_extractions") or [] - duplicate_rows = readback.get("duplicate_assessments") or [] - conflict_rows = readback.get("conflict_assessments") or [] - proposal = readback.get("staged_proposal") or {} + source_value = readback.get("source_captures") + claim_value = readback.get("claim_extractions") + evidence_value = readback.get("evidence_extractions") + duplicate_value = readback.get("duplicate_assessments") + conflict_value = readback.get("conflict_assessments") + source_rows = source_value if isinstance(source_value, list) else [] + claim_rows = claim_value if isinstance(claim_value, list) else [] + evidence_rows = evidence_value if isinstance(evidence_value, list) else [] + duplicate_rows = duplicate_value if isinstance(duplicate_value, list) else [] + conflict_rows = conflict_value if isinstance(conflict_value, list) else [] + proposal, apply_payload, _manifest, ingestion_manifest, assessment = proposal_readback_parts(readback) counts = input_receipt["counts"] - links = proposal_links(readback, input_receipt) - manifest = ((proposal.get("payload") or {}).get("apply_payload") or {}).get("normalization_manifest") or {} - ingestion_manifest = manifest.get("ingestion_manifest") or {} - assessment = manifest.get("dedupe_conflict_assessment") or {} + links = proposal_links(readback) + deterministic_row_ids = build_row_ids(input_receipt, fixture) + expected_rows = expected_persisted_rows(fixture, input_receipt, deterministic_row_ids) + expected_parent = build_parent_packet(fixture, input_receipt, deterministic_row_ids) + expected_child = normalized_stage.prepare_normalized_child(expected_parent) + expected_apply_payload = expected_child["payload"]["apply_payload"] + persisted_proposal_envelope_fields = ( + "id", + "proposal_type", + "status", + "proposed_by_handle", + "proposed_by_agent_id", + "channel", + "source_ref", + "rationale", + "reviewed_by_handle", + "reviewed_by_agent_id", + "reviewed_at", + "review_note", + "applied_by_handle", + "applied_by_agent_id", + "applied_at", + ) + expected_proposal_envelope = {key: expected_child.get(key) for key in persisted_proposal_envelope_fields} + actual_proposal_envelope = {key: proposal.get(key) for key in persisted_proposal_envelope_fields} expected_db_counts = { "source_captures": 1, "claim_extractions": counts["claim_candidates"], @@ -935,63 +1241,91 @@ def evaluate_checks( "conflict_assessments": counts["conflict_relationships"], "staged_proposals": 1, } - segment_by_id = {segment["id"]: segment for segment in fixture["segments"]} - evidence_locations_exact = all( - row.get("source_locator") == segment_by_id.get(row.get("source_segment_id"), {}).get("locator") - and row.get("source_content_sha256") - == segment_by_id.get(row.get("source_segment_id"), {}).get("content_sha256") - and row.get("body") in segment_by_id.get(row.get("source_segment_id"), {}).get("body", "") - for row in evidence_rows + source_rows_exact = rows_exact(source_rows, expected_rows["source_captures"]) + claim_rows_exact = rows_exact(claim_rows, expected_rows["claim_extractions"]) + evidence_rows_exact = rows_exact(evidence_rows, expected_rows["evidence_extractions"]) + duplicate_rows_exact = rows_exact(duplicate_rows, expected_rows["duplicate_assessments"]) + conflict_rows_exact = rows_exact(conflict_rows, expected_rows["conflict_assessments"]) + proposal_envelope_exact = actual_proposal_envelope == expected_proposal_envelope + planned_rows_exact = all( + apply_payload.get(collection) == expected_apply_payload.get(collection) + for collection in ("claims", "sources", "evidence", "edges") ) + graph_collections = {"claims", "sources", "evidence", "edges"} + apply_auxiliary_exact = {key: value for key, value in apply_payload.items() if key not in graph_collections} == { + key: value for key, value in expected_apply_payload.items() if key not in graph_collections + } + expected_graph_identity = expected_planned_graph_identity(fixture, deterministic_row_ids) + actual_graph_identity = { + key: links[key] + for key in ("planned_claim_ids", "planned_source_ids", "planned_evidence_rows", "planned_edge_rows") + } + independent_graph_identity_exact = actual_graph_identity == expected_graph_identity + pending_state = normalized_stage.bound._state_semantics(proposal, "pending_review") + source_row = source_rows[0] if len(source_rows) == 1 and isinstance(source_rows[0], dict) else {} + claim_rows_are_objects = all(isinstance(row, dict) for row in claim_rows) return { "source_only_artifact_hash_persisted": ( - len(source_rows) == 1 and source_rows[0].get("artifact_sha256") == input_receipt["artifact_sha256"] + len(source_rows) == 1 and source_row.get("artifact_sha256") == input_receipt["artifact_sha256"] ), + "source_capture_identity_exact": source_rows_exact, + "deterministic_row_ids_recomputed_exact": row_ids == deterministic_row_ids, "replay_identity_persisted": ( len(source_rows) == 1 - and source_rows[0].get("replay_identity_sha256") == input_receipt["replay_identity_sha256"] + and source_row.get("replay_identity_sha256") == input_receipt["replay_identity_sha256"] ), "extractor_name_and_version_persisted": ( ingestion_manifest.get("extractor") == input_receipt["extractor"] - and all(row.get("metadata", {}).get("extractor") == input_receipt["extractor"] for row in claim_rows) + and claim_rows_are_objects + and all( + isinstance(row.get("metadata"), dict) and row["metadata"].get("extractor") == input_receipt["extractor"] + for row in claim_rows + ) ), "all_claim_candidates_persisted_once": ( - {row.get("claim_key") for row in claim_rows} + claim_rows_are_objects + and {row.get("claim_key") for row in claim_rows} == {claim["claim_key"] for claim in fixture["extraction"]["claims"]} and len(claim_rows) == counts["claim_candidates"] ), + "claim_extraction_identities_and_fks_exact": claim_rows_exact, "all_evidence_has_exact_source_location": ( - len(evidence_rows) == counts["evidence_candidates"] and evidence_locations_exact + len(evidence_rows) == counts["evidence_candidates"] and evidence_rows_exact ), + "evidence_extraction_identities_and_fks_exact": evidence_rows_exact, "duplicate_judgments_persisted": ( len(duplicate_rows) == counts["duplicate_judgments"] + and duplicate_rows_exact and assessment.get("duplicates") == fixture["extraction"]["duplicates"] ), "conflicts_and_relationships_persisted": ( len(conflict_rows) == counts["conflict_relationships"] + and conflict_rows_exact and assessment.get("conflicts") == fixture["extraction"]["conflicts"] - and links["planned_counts"]["edges"] == counts["conflict_relationships"] + and links["planned_edge_rows"] == expected_graph_identity["planned_edge_rows"] + ), + "planned_graph_matches_independent_source_oracle": independent_graph_identity_exact, + "planned_proposal_identities_and_linkages_exact": ( + proposal_envelope_exact + and planned_rows_exact + and apply_auxiliary_exact + and independent_graph_identity_exact + and links["manifest_row_ids"] == deterministic_row_ids + and links["proposal_id"] == expected_child["id"] + and links["proposal_status"] == expected_child["status"] + and links["proposal_source_ref"] == expected_child["source_ref"] ), "database_candidate_counts_exact": readback.get("counts") == expected_db_counts, - "proposal_is_pending_review": proposal.get("status") == "pending_review", + "proposal_is_pending_review": pending_state["status_matches"], "proposal_has_replayable_ingestion_manifest": ( - links["input_hash_in_manifest"] - and links["replay_identity_in_manifest"] - and ingestion_manifest.get("segments") - == [ - { - "id": segment["id"], - "locator": segment["locator"], - "json_pointer": segment["json_pointer"], - "content_sha256": segment["content_sha256"], - "text_sha256": segment["text_sha256"], - } - for segment in fixture["segments"] - ] + links["manifest_artifact_sha256"] == input_receipt["artifact_sha256"] + and links["manifest_replay_identity_sha256"] == input_receipt["replay_identity_sha256"] + and ingestion_manifest == expected_apply_payload["normalization_manifest"]["ingestion_manifest"] ), - "no_review_or_apply_metadata": all( - proposal.get(key) is None - for key in ("reviewed_by_handle", "reviewed_at", "applied_by_handle", "applied_at") + "no_review_or_apply_metadata": ( + pending_state["review_fields_match"] + and pending_state["apply_fields_match"] + and proposal.get("review_note") is None ), "canonical_snapshot_nonempty": ( canonical_snapshot_complete(canonical_before) and canonical_snapshot_complete(canonical_after) @@ -1005,22 +1339,24 @@ def evaluate_checks( def run_canary(fixture_path: Path, output: Path | None = None) -> dict[str, Any]: - fixture, input_receipt = load_fixture(fixture_path) - row_ids = build_row_ids(input_receipt, fixture) - parent = build_parent_packet(fixture, input_receipt, row_ids) - child = normalized_stage.prepare_normalized_child(parent) - container_name = f"working-leo-ingest-{uuid.uuid4().hex[:12]}" - postgres = DockerPostgres(container_name) receipt: dict[str, Any] = { "schema": SCHEMA, "status": "fail", "required_tier": "T2_runtime_to_review_queue_staging", - "input": input_receipt, - "row_ids": row_ids, + "input": {"scenario_path": str(fixture_path.resolve())}, + "row_ids": {}, "production_mutation_attempted": False, "canonical_apply_attempted": False, } + postgres: DockerPostgres | None = None try: + fixture, input_receipt = load_fixture(fixture_path) + row_ids = build_row_ids(input_receipt, fixture) + parent = build_parent_packet(fixture, input_receipt, row_ids) + child = normalized_stage.prepare_normalized_child(parent) + receipt["input"] = input_receipt + receipt["row_ids"] = row_ids + postgres = DockerPostgres(f"working-leo-ingest-{uuid.uuid4().hex[:12]}") receipt["environment"] = postgres.start() postgres.psql(build_schema_sql()) canonical_before = postgres.psql_json(build_canonical_snapshot_sql()) @@ -1039,7 +1375,7 @@ def run_canary(fixture_path: Path, output: Path | None = None) -> dict[str, Any] "stage_command_result": stage_result, "rows": readback, "candidate_counts": input_receipt["counts"], - "row_link_fields_proven": proposal_links(readback, input_receipt), + "row_link_fields_proven": proposal_link_proof(readback, input_receipt), "canonical": { "fingerprint_before": canonical_sha256(canonical_before), "fingerprint_after": canonical_sha256(canonical_after), @@ -1050,19 +1386,45 @@ def run_canary(fixture_path: Path, output: Path | None = None) -> dict[str, Any] "status": "pass" if all(checks.values()) else "fail", } ) - except (CanaryError, OSError, ValueError, KeyError, normalized_stage.bound.CheckpointError) as exc: + except ( + CanaryError, + OSError, + ValueError, + KeyError, + TypeError, + AttributeError, + IndexError, + normalized_stage.bound.CheckpointError, + ) as exc: receipt["error"] = {"type": type(exc).__name__, "message": str(exc)} finally: - receipt["cleanup"] = postgres.cleanup() - receipt["cleanup"]["network_mode_none"] = receipt.get("environment", {}).get("network_mode") == "none" + if postgres is None: + receipt["cleanup"] = { + "remove_attempted": False, + "remove_returncode": None, + "container_absent": True, + "docker_volume_mounts_observed": [], + "anonymous_or_named_volume_created": False, + "runtime_not_started": True, + "network_mode_none": True, + } + else: + receipt["cleanup"] = postgres.cleanup() + receipt["cleanup"]["runtime_not_started"] = not postgres.started + receipt["cleanup"]["network_mode_none"] = receipt.get("environment", {}).get("network_mode") == "none" receipt["cleanup"]["no_production_target_referenced"] = receipt["cleanup"]["network_mode_none"] if not all((receipt["cleanup"]["container_absent"], receipt["cleanup"]["network_mode_none"])): receipt["status"] = "fail" - receipt["strongest_claim"] = ( - "One source-only artifact was parsed into exact-location candidate claims, duplicate/conflict " - "assessments, and a replay-bound pending_review proposal in disposable networkless Postgres; " - "representative canonical row contents remained fingerprint-identical." - ) + if receipt["status"] == "pass": + receipt["strongest_claim"] = ( + "One source-only artifact was parsed into exact-location candidate claims, duplicate/conflict " + "assessments, and a replay-bound pending_review proposal in disposable networkless Postgres; " + "representative canonical row contents remained fingerprint-identical." + ) + else: + receipt["strongest_claim"] = ( + "No source-to-review-staging capability claim is made because acceptance failed closed." + ) receipt["residual_gap"] = ( "This proves the deterministic local fixture extractor and review queue only; it does not prove " "arbitrary generative extraction, approval/apply, hosted runtime, or production mutation." @@ -1073,21 +1435,74 @@ def run_canary(fixture_path: Path, output: Path | None = None) -> dict[str, Any] return receipt +def receipt_proves_canonical_invariance(receipt: dict[str, Any]) -> bool: + """Recompute the suite-level canonical verdict from receipt evidence.""" + canonical = receipt.get("canonical") + checks = receipt.get("checks") + if not isinstance(canonical, dict) or not isinstance(checks, dict): + return False + before = canonical.get("fingerprint_before") + after = canonical.get("fingerprint_after") + return ( + isinstance(before, str) + and isinstance(after, str) + and re.fullmatch(r"[0-9a-f]{64}", before) is not None + and re.fullmatch(r"[0-9a-f]{64}", after) is not None + and before == after + and canonical.get("unchanged") is True + and checks.get("canonical_fingerprint_unchanged") is True + ) + + def run_suite(fixture_paths: list[Path], output: Path | None = None) -> dict[str, Any]: receipts = [run_canary(path.resolve()) for path in fixture_paths] + all_supplied_pass = bool(receipts) and all(item.get("status") == "pass" for item in receipts) + document_fixture_passed = any( + item.get("status") == "pass" and item.get("input", {}).get("artifact_format") == "plain_text" + for item in receipts + ) + social_conversation_fixture_passed = any( + item.get("status") == "pass" and item.get("input", {}).get("artifact_format") == "telegram_jsonl" + for item in receipts + ) + canonical_fingerprint_invariant_all = bool(receipts) and all( + receipt_proves_canonical_invariance(item) for item in receipts + ) + required_shape_coverage = document_fixture_passed and social_conversation_fixture_passed + full_suite_passed = all_supplied_pass and required_shape_coverage and canonical_fingerprint_invariant_all + if not all_supplied_pass or not canonical_fingerprint_invariant_all: + status = "fail" + elif full_suite_passed: + status = "pass" + else: + status = "partial" + missing_required_coverage = [] + if not document_fixture_passed: + missing_required_coverage.append("document") + if not social_conversation_fixture_passed: + missing_required_coverage.append("social_conversation") + if not canonical_fingerprint_invariant_all: + missing_required_coverage.append("canonical_fingerprint_invariance") suite = { "schema": SUITE_SCHEMA, - "status": "pass" if receipts and all(item["status"] == "pass" for item in receipts) else "fail", + "status": status, "required_tier": "T2_runtime_to_review_queue_staging", + "current_tier": ( + "T2_runtime_to_review_queue_staging" + if full_suite_passed + else "T2_runtime_single_fixture" + if status == "partial" + else "runtime_verification_failed" + ), "fixture_count": len(receipts), - "document_fixture_passed": any( - item["status"] == "pass" and item["input"]["artifact_format"] == "plain_text" for item in receipts - ), - "social_conversation_fixture_passed": any( - item["status"] == "pass" and item["input"]["artifact_format"] == "telegram_jsonl" for item in receipts - ), - "canonical_fingerprint_invariant_all": bool(receipts) - and all(item.get("canonical", {}).get("unchanged") is True for item in receipts), + "all_supplied_fixtures_passed": all_supplied_pass, + "document_fixture_passed": document_fixture_passed, + "social_conversation_fixture_passed": social_conversation_fixture_passed, + "canonical_fingerprint_invariant_all": canonical_fingerprint_invariant_all, + "required_shape_coverage_met": required_shape_coverage, + "coverage_status": "full" if required_shape_coverage else "partial", + "missing_required_coverage": missing_required_coverage, + "full_suite_passed": full_suite_passed, "receipts": receipts, } if output: diff --git a/tests/test_run_leo_local_ingestion_proposal_canary.py b/tests/test_run_leo_local_ingestion_proposal_canary.py index cfae09e..1290dc4 100644 --- a/tests/test_run_leo_local_ingestion_proposal_canary.py +++ b/tests/test_run_leo_local_ingestion_proposal_canary.py @@ -32,6 +32,37 @@ def _copy_scenario(tmp_path: Path, scenario_path: Path) -> Path: 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 = [] @@ -44,7 +75,14 @@ def _synthetic_readback(fixture: dict, input_receipt: dict, row_ids: dict, child "source_capture_id": row_ids["source_capture_id"], "claim_key": claim["claim_key"], "body": claim["body"], - "metadata": {"extractor": input_receipt["extractor"]}, + "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"]): @@ -56,10 +94,52 @@ def _synthetic_readback(fixture: dict, input_receipt: dict, row_ids: dict, child "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, @@ -71,14 +151,22 @@ def _synthetic_readback(fixture: dict, input_receipt: dict, row_ids: dict, child 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": [{} for _item in fixture["extraction"]["duplicates"]], - "conflict_assessments": [{} for _item in fixture["extraction"]["conflicts"]], + "duplicate_assessments": duplicate_rows, + "conflict_assessments": conflict_rows, "staged_proposal": proposal, "counts": { "source_captures": 1, @@ -208,13 +296,98 @@ def test_replay_identity_and_ids_change_with_extractor_version(tmp_path: 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["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) @@ -311,3 +484,299 @@ def test_evaluation_fails_when_exact_evidence_locator_is_changed() -> None: ] 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"]