diff --git a/fixtures/working-leo/document-ingestion-v2.scenario.json b/fixtures/working-leo/document-ingestion-v2.scenario.json new file mode 100644 index 0000000..bf9a5f6 --- /dev/null +++ b/fixtures/working-leo/document-ingestion-v2.scenario.json @@ -0,0 +1,52 @@ +{ + "schema": "livingip.workingLeoSourceIngestionScenario.v2", + "artifact": { + "path": "document-ingestion-v2.txt", + "format": "plain_text" + }, + "source": { + "identity": "document:working-leo-review-gated-composition-v2", + "source_key": "working_leo_document_ingestion_v2", + "source_type": "article", + "title": "Working Leo review-gated composition note", + "locator": "fixture://working-leo/document-ingestion-v2", + "metadata": { + "author": "Working Leo synthetic canary fixture", + "captured_at": "2026-07-15T00:00:00Z", + "document_kind": "operating_note", + "language": "en", + "synthetic": true + } + }, + "extractor": { + "name": "deterministic_fixture_replay", + "version": "2" + }, + "extraction": { + "claims": [ + { + "claim_key": "staged_proposal_remains_noncanonical", + "type": "structural", + "confidence": 0.75, + "text": "A staged knowledge proposal remains non-canonical until review and guarded apply succeed.", + "body": "A staged proposal remains non-canonical until an operator reviews it and a separate guarded apply succeeds.", + "metadata": { + "needs_research": false + }, + "evidence": [ + { + "segment_id": "paragraph-2", + "role": "grounds", + "excerpt": "A staged proposal remains non-canonical until an operator reviews it and a separate guarded apply succeeds.", + "metadata": { + "evidence_kind": "exact_quote" + } + } + ], + "edges": [] + } + ], + "duplicates": [], + "conflicts": [] + } +} diff --git a/fixtures/working-leo/document-ingestion-v2.txt b/fixtures/working-leo/document-ingestion-v2.txt new file mode 100644 index 0000000..470fc38 --- /dev/null +++ b/fixtures/working-leo/document-ingestion-v2.txt @@ -0,0 +1,5 @@ +Working Leo composition note. A newly captured document may produce claim and evidence candidates, but those candidates are not canonical knowledge. + +A staged proposal remains non-canonical until an operator reviews it and a separate guarded apply succeeds. The proposal must retain the source content hash, exact evidence excerpt, and source identity so reviewers can trace every planned row back to the captured artifact. + +For a staging-only rehearsal, pending_review is the terminal state. No public knowledge-base row should be written. diff --git a/fixtures/working-leo/telegram-transcript-ingestion-v1.jsonl b/fixtures/working-leo/telegram-transcript-ingestion-v1.jsonl new file mode 100644 index 0000000..dd3c7cf --- /dev/null +++ b/fixtures/working-leo/telegram-transcript-ingestion-v1.jsonl @@ -0,0 +1,3 @@ +{"ts":"2026-07-15T09:00:01Z","chat_id":900001,"chat_title":"Synthetic review gate thread","message_id":7301,"type":"message","username":"fixture_analyst","display_name":"Fixture Analyst","user_id":910001,"message":"A pending review does not change canonical knowledge; only a separate guarded apply can do that.","reply_to":null,"synthetic":true} +{"ts":"2026-07-15T09:00:12Z","chat_id":900001,"chat_title":"Synthetic review gate thread","message_id":7302,"type":"message","username":"fixture_operator","display_name":"Fixture Operator","user_id":910002,"message":"Reviewer approval alone makes the proposal canonical immediately; no apply step is needed.","reply_to":7301,"synthetic":true} +{"ts":"2026-07-15T09:00:24Z","chat_id":900001,"chat_title":"Synthetic review gate thread","message_id":7303,"type":"message","username":"fixture_reviewer","display_name":"Fixture Reviewer","user_id":910003,"message":"A pending review does not change canonical knowledge; only a separate guarded apply can do that.","reply_to":7302,"synthetic":true} diff --git a/fixtures/working-leo/telegram-transcript-ingestion-v1.scenario.json b/fixtures/working-leo/telegram-transcript-ingestion-v1.scenario.json new file mode 100644 index 0000000..696ecbb --- /dev/null +++ b/fixtures/working-leo/telegram-transcript-ingestion-v1.scenario.json @@ -0,0 +1,99 @@ +{ + "schema": "livingip.workingLeoSourceIngestionScenario.v2", + "artifact": { + "path": "telegram-transcript-ingestion-v1.jsonl", + "format": "telegram_jsonl" + }, + "source": { + "identity": "fixture:telegram-chat-900001-export-20260715", + "source_key": "telegram_review_gate_exchange_20260715", + "source_type": "transcript", + "title": "Synthetic Telegram review-gate exchange", + "locator": "fixture://working-leo/telegram-transcript-ingestion-v1", + "metadata": { + "captured_at": "2026-07-15T09:01:00Z", + "export_format": "teleo_append_only_telegram_jsonl", + "language": "en", + "synthetic": true + } + }, + "extractor": { + "name": "deterministic_telegram_jsonl_replay", + "version": "1" + }, + "extraction": { + "claims": [ + { + "claim_key": "pending_review_does_not_change_canonical", + "type": "structural", + "confidence": 0.7, + "text": "Pending review alone does not change canonical knowledge.", + "body": "A pending review does not change canonical knowledge; only a separate guarded apply can do that.", + "metadata": { + "needs_research": false + }, + "evidence": [ + { + "segment_id": "message-900001-7301", + "role": "grounds", + "excerpt": "A pending review does not change canonical knowledge; only a separate guarded apply can do that.", + "metadata": { + "evidence_kind": "exact_message" + } + }, + { + "segment_id": "message-900001-7303", + "role": "illustrates", + "excerpt": "A pending review does not change canonical knowledge; only a separate guarded apply can do that.", + "metadata": { + "evidence_kind": "duplicate_exact_message" + } + } + ], + "edges": [ + { + "edge_type": "contradicts", + "target": "approval_alone_makes_canonical" + } + ] + }, + { + "claim_key": "approval_alone_makes_canonical", + "type": "empirical", + "confidence": 0.4, + "text": "Reviewer approval alone makes a proposal canonical without apply.", + "body": "Reviewer approval alone makes the proposal canonical immediately; no apply step is needed.", + "metadata": { + "needs_research": true + }, + "evidence": [ + { + "segment_id": "message-900001-7302", + "role": "grounds", + "excerpt": "Reviewer approval alone makes the proposal canonical immediately; no apply step is needed.", + "metadata": { + "evidence_kind": "exact_message" + } + } + ], + "edges": [] + } + ], + "duplicates": [ + { + "segment_id": "message-900001-7303", + "duplicate_of_claim_key": "pending_review_does_not_change_canonical", + "excerpt": "A pending review does not change canonical knowledge; only a separate guarded apply can do that.", + "reason": "Message 7303 repeats message 7301 exactly and is retained as explicit duplicate evidence." + } + ], + "conflicts": [ + { + "from_claim_key": "pending_review_does_not_change_canonical", + "to_claim_key": "approval_alone_makes_canonical", + "relationship": "contradicts", + "reason": "The two candidate propositions assert incompatible canonical-state transitions." + } + ] + } +} diff --git a/scripts/prepare_kb_source_manifest.py b/scripts/prepare_kb_source_manifest.py index 8a16403..6823e1d 100644 --- a/scripts/prepare_kb_source_manifest.py +++ b/scripts/prepare_kb_source_manifest.py @@ -34,7 +34,7 @@ DEFAULT_MODEL = "anthropic/claude-sonnet-4.5" EXTRACTOR_VERSION = "2" DEFAULT_OPENROUTER_URL = "https://openrouter.ai/api/v1/chat/completions" DEFAULT_KEY_FILE = Path("/opt/teleo-eval/secrets/openrouter-key") -TEXT_SUFFIXES = {".csv", ".htm", ".html", ".json", ".md", ".rst", ".txt", ".xml"} +TEXT_SUFFIXES = {".csv", ".htm", ".html", ".json", ".jsonl", ".md", ".rst", ".txt", ".xml"} MAX_SOURCE_CHARS = 120_000 MAX_CANDIDATES = 20 MAX_CLAIMS = 3 @@ -124,20 +124,37 @@ def _load_context(path: Path) -> dict[str, Any]: return {"database_search_query": query.strip(), "candidate_claims": normalized} -def build_source_fragments(text: str) -> dict[str, str]: - """Label non-empty source lines so the model selects text instead of copying it.""" +def build_source_fragment_index(text: str) -> tuple[dict[str, str], dict[str, dict[str, Any]]]: + """Return selectable lines plus exact line/character provenance.""" fragments: dict[str, str] = {} - for line in text.splitlines(): - if line.strip(): - fragments[f"S{len(fragments) + 1:05d}"] = line + locations: dict[str, dict[str, Any]] = {} + offset = 0 + for line_number, raw_line in enumerate(text.splitlines(keepends=True), start=1): + line = raw_line.rstrip("\r\n") + quote = line.strip() + if quote: + reference = f"S{len(fragments) + 1:05d}" + start = offset + line.find(quote) + fragments[reference] = quote + locations[reference] = { + "reference": reference, + "line": line_number, + "char_start": start, + "char_end": start + len(quote), + "quote_sha256": sha256_bytes(quote.encode("utf-8")), + } + offset += len(raw_line) if not fragments: raise PreparationError("extracted text has no selectable source lines") - return fragments + return fragments, locations -def build_prompt( - fragments: dict[str, str], context: dict[str, Any], repair_error: str | None = None -) -> str: +def build_source_fragments(text: str) -> dict[str, str]: + """Label non-empty source lines so the model selects text instead of copying it.""" + return build_source_fragment_index(text)[0] + + +def build_prompt(fragments: dict[str, str], context: dict[str, Any], repair_error: str | None = None) -> str: candidates = json.dumps(context["candidate_claims"], indent=2, ensure_ascii=True) source = "\n".join(f"[{reference}] {line}" for reference, line in fragments.items()) repair = "" @@ -273,9 +290,7 @@ def parse_model_output(content: str) -> dict[str, Any]: if not isinstance(claim, dict): raise PreparationError(f"model extraction claims[{index}] must be an object") if set(claim) != MODEL_CLAIM_KEYS: - raise PreparationError( - f"model extraction claims[{index}] keys must equal {sorted(MODEL_CLAIM_KEYS)}" - ) + raise PreparationError(f"model extraction claims[{index}] keys must equal {sorted(MODEL_CLAIM_KEYS)}") confidence = claim.get("confidence") if confidence is not None and ( isinstance(confidence, bool) or not isinstance(confidence, (int, float)) or not 0 <= confidence <= 0.75 @@ -304,27 +319,47 @@ def _resolve_reference(reference: Any, fragments: dict[str, str], label: str) -> return fragments[reference] -def resolve_model_output(selection: dict[str, Any], fragments: dict[str, str]) -> dict[str, Any]: +def resolve_model_output( + selection: dict[str, Any], + fragments: dict[str, str], + fragment_locations: dict[str, dict[str, Any]] | None = None, +) -> dict[str, Any]: + def selected(kind: str, reference: str, **identity: Any) -> dict[str, Any]: + result = {"kind": kind, "reference": reference, **identity} + if fragment_locations is not None: + result.update(fragment_locations[reference]) + return result + claims: list[dict[str, Any]] = [] - selected_references: list[dict[str, str]] = [] + selected_references: list[dict[str, Any]] = [] for claim_index, claim in enumerate(selection["claims"]): quote_ref = claim["quote_ref"] quote = _resolve_reference(quote_ref, fragments, f"claims[{claim_index}].quote_ref") - selected_references.append({"kind": "claim", "reference": quote_ref}) + selected_references.append(selected("claim", quote_ref, claim_key=claim["claim_key"], quote=quote)) evidence: list[dict[str, str]] = [] for evidence_index, row in enumerate(claim["evidence"]): evidence_ref = row["quote_ref"] + evidence_quote = _resolve_reference( + evidence_ref, + fragments, + f"claims[{claim_index}].evidence[{evidence_index}].quote_ref", + ) evidence.append( { - "quote": _resolve_reference( - evidence_ref, - fragments, - f"claims[{claim_index}].evidence[{evidence_index}].quote_ref", - ), + "quote": evidence_quote, "role": row["role"], } ) - selected_references.append({"kind": "evidence", "reference": evidence_ref}) + selected_references.append( + selected( + "evidence", + evidence_ref, + claim_key=claim["claim_key"], + evidence_index=evidence_index, + evidence_role=row["role"], + quote=evidence_quote, + ) + ) claims.append( { "claim_key": claim["claim_key"], @@ -351,7 +386,14 @@ def resolve_model_output(selection: dict[str, Any], fragments: dict[str, str]) - "reason": duplicate["reason"], } ) - selected_references.append({"kind": "duplicate", "reference": source_quote_ref}) + selected_references.append( + selected( + "duplicate", + source_quote_ref, + claim_id=duplicate["claim_id"], + quote=duplicates[-1]["source_quote"], + ) + ) return { "schema": RESOLVED_OUTPUT_SCHEMA, @@ -362,6 +404,37 @@ def resolve_model_output(selection: dict[str, Any], fragments: dict[str, str]) - } +def validate_bundle_source_locations(bundle: dict[str, Any], selected_references: list[dict[str, Any]]) -> None: + """Ensure compiler quote offsets match the exact selected fragment occurrence.""" + available = list(bundle.get("quote_bindings") or []) + for selected in selected_references: + if selected.get("kind") not in {"claim", "evidence"}: + continue + match_index = next( + ( + index + for index, binding in enumerate(available) + if binding.get("kind") == selected.get("kind") + and binding.get("claim_key") == selected.get("claim_key") + and binding.get("quote") == selected.get("quote") + and ( + selected.get("kind") != "evidence" or binding.get("evidence_role") == selected.get("evidence_role") + ) + ), + None, + ) + if match_index is None: + raise PreparationError("compiler dropped a selected claim/evidence source binding") + binding = available.pop(match_index) + if binding.get("first_start") != selected.get("char_start") or binding.get("first_end") != selected.get( + "char_end" + ): + raise PreparationError( + f"selected {selected['reference']} is an ambiguous repeated quote; " + "the current compiler cannot preserve that exact occurrence" + ) + + def build_manifest( *, args: argparse.Namespace, @@ -427,7 +500,7 @@ def prepare(args: argparse.Namespace) -> dict[str, Any]: artifact_bytes = _read_nonempty(args.artifact, "artifact") text_bytes = extract_utf8_text(args.artifact, args.text) text = text_bytes.decode("utf-8", errors="strict") - fragments = build_source_fragments(text) + fragments, fragment_locations = build_source_fragment_index(text) context = _load_context(args.kb_context) requested_output_dir = args.output_dir.expanduser() @@ -464,7 +537,7 @@ def prepare(args: argparse.Namespace) -> dict[str, Any]: attempts.append({"attempt": attempt, "response_sha256": sha256_bytes(raw.encode("utf-8")), "usage": usage}) try: selection = parse_model_output(raw) - extraction = resolve_model_output(selection, fragments) + extraction = resolve_model_output(selection, fragments, fragment_locations) compiler._validate_dedupe( { "database_search_query": context["database_search_query"], @@ -494,6 +567,7 @@ def prepare(args: argparse.Namespace) -> dict[str, Any]: with os.fdopen(fd, "wb") as handle: handle.write(_json_bytes(manifest)) bundle = compiler.compile_source_packet(args.artifact, temp_text, temp_manifest) + validate_bundle_source_locations(bundle, extraction["selected_source_references"]) finally: if temp_text.exists(): temp_text.unlink() @@ -518,6 +592,7 @@ def prepare(args: argparse.Namespace) -> dict[str, Any]: raise PreparationError("prepared extraction has no manifest") _write_private(manifest_path, _json_bytes(manifest)) bundle = compiler.compile_source_packet(args.artifact, text_path, manifest_path) + validate_bundle_source_locations(bundle, extraction["selected_source_references"]) receipt = { "schema": RECEIPT_SCHEMA, @@ -545,10 +620,20 @@ def prepare(args: argparse.Namespace) -> dict[str, Any]: "model": args.model, "attempts": attempts, "claim_count": len(extraction["claims"]), + "evidence_count": sum(len(claim["evidence"]) for claim in extraction["claims"]), + "duplicate_judgment_count": len(extraction["duplicates"]), "selectable_source_line_count": len(fragments), "selected_source_references": extraction["selected_source_references"], "notes": extraction["extraction_notes"], }, + "replay": { + "artifact_sha256": artifact_sha256, + "extracted_text_sha256": text_sha256, + "kb_context_sha256": sha256_bytes(_json_bytes(context)), + "fragment_index_sha256": sha256_bytes(_json_bytes(fragment_locations)), + "extractor": {"name": f"openrouter:{args.model}", "version": EXTRACTOR_VERSION}, + "exact_selected_locations_validated": bundle is not None, + }, "outputs": { "output_dir": str(output_dir), "extracted_text": str(text_path), diff --git a/scripts/run_leo_local_ingestion_proposal_canary.py b/scripts/run_leo_local_ingestion_proposal_canary.py index 89270dc..12854b2 100644 --- a/scripts/run_leo_local_ingestion_proposal_canary.py +++ b/scripts/run_leo_local_ingestion_proposal_canary.py @@ -1,11 +1,20 @@ #!/usr/bin/env python3 -"""Run a document-ingestion-to-staged-proposal canary in disposable local Postgres.""" +"""Ingest source-only fixtures into review staging in disposable Postgres. + +The canary parses a raw document or repo-native Telegram JSONL transcript, +replays a versioned deterministic extraction sidecar, persists exact source +locations, normalizes the candidates, and stages one ``pending_review`` +proposal. Representative canonical ``public.*`` rows are fingerprinted before +and after. No review, apply, network access, production target, or durable +Docker volume is used. +""" from __future__ import annotations import argparse import hashlib import json +import re import shutil import subprocess import sys @@ -21,89 +30,526 @@ sys.path.insert(0, str(HERE)) import apply_proposal as ap # noqa: E402 import stage_normalized_proposal as normalized_stage # noqa: E402 -SCHEMA = "livingip.workingLeoLocalIngestionProposalCanary.v1" -FIXTURE_SCHEMA = "livingip.workingLeoDocumentIngestionFixture.v1" -DEFAULT_FIXTURE = REPO_ROOT / "fixtures" / "working-leo" / "document-ingestion-v1.json" +SCHEMA = "livingip.workingLeoLocalIngestionProposalCanary.v2" +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" +DEFAULT_FIXTURE = DEFAULT_DOCUMENT_FIXTURE +DEFAULT_FIXTURES = (DEFAULT_DOCUMENT_FIXTURE, DEFAULT_TELEGRAM_FIXTURE) CONTAINER_LABEL = "livingip.canary=working-leo-local-ingestion" +CANONICAL_TABLES = ("claims", "sources", "claim_evidence", "claim_edges") +SEED_CLAIM_A = "00000000-0000-4000-8000-000000000101" +SEED_CLAIM_B = "00000000-0000-4000-8000-000000000102" +SEED_SOURCE = "00000000-0000-4000-8000-000000000201" class CanaryError(RuntimeError): - pass + """Raised when a fixture or disposable runtime fails closed.""" def sha256_bytes(value: bytes) -> str: return hashlib.sha256(value).hexdigest() +def canonical_json_bytes(value: Any) -> bytes: + return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode() + + def canonical_sha256(value: Any) -> str: - return sha256_bytes(json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode()) + return sha256_bytes(canonical_json_bytes(value)) -def stable_uuid(artifact_sha256: str, label: str) -> str: - return str(uuid.uuid5(uuid.NAMESPACE_URL, f"livingip:working-leo-ingestion:{artifact_sha256}:{label}")) +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 + ) + + +def stable_uuid(seed: str, label: str) -> str: + return str(uuid.uuid5(uuid.NAMESPACE_URL, f"livingip:working-leo-ingestion:{seed}:{label}")) + + +def _require_object(value: Any, label: str) -> dict[str, Any]: + if not isinstance(value, dict): + raise CanaryError(f"{label} must be an object") + return value + + +def _require_list(value: Any, label: str) -> list[Any]: + if not isinstance(value, list): + raise CanaryError(f"{label} must be a list") + return value + + +def _require_text(value: Any, label: str) -> str: + if not isinstance(value, str) or not value.strip(): + raise CanaryError(f"{label} must be a non-empty string") + return value.strip() + + +def _safe_source_key(value: str) -> str: + normalized = re.sub(r"[^a-z0-9._:-]+", "_", value.lower()).strip("_") + if not normalized: + raise CanaryError(f"could not derive a stable source key from {value!r}") + return normalized[:128] + + +def _load_plain_text_segments(raw: bytes, source: dict[str, Any]) -> list[dict[str, Any]]: + try: + text = raw.decode("utf-8", errors="strict") + except UnicodeDecodeError as exc: + raise CanaryError(f"plain_text artifact must be strict UTF-8: {exc}") from exc + locator = _require_text(source.get("locator"), "source.locator") + segments: list[dict[str, Any]] = [] + cursor = 0 + for part in re.split(r"\n[ \t]*\n", text): + part_start = text.find(part, cursor) + cursor = part_start + len(part) + body = part.strip() + if not body: + continue + start = part_start + part.find(body) + index = len(segments) + 1 + canonical_record = {"paragraph": index, "text": body} + segments.append( + { + "id": f"paragraph-{index}", + "body": body, + "locator": f"{locator}#paragraph-{index}", + "json_pointer": None, + "content_sha256": canonical_sha256(canonical_record), + "text_sha256": sha256_bytes(body.encode()), + "metadata": {"paragraph": index, "char_start": start, "char_end": start + len(body)}, + } + ) + if not segments: + raise CanaryError("plain_text artifact contains no non-empty paragraphs") + return segments + + +def _load_telegram_jsonl_segments(raw: bytes) -> list[dict[str, Any]]: + try: + text = raw.decode("utf-8", errors="strict") + except UnicodeDecodeError as exc: + raise CanaryError(f"telegram_jsonl artifact must be strict UTF-8: {exc}") from exc + segments: list[dict[str, Any]] = [] + seen: set[tuple[int, int]] = set() + for line_number, line in enumerate(text.splitlines(), start=1): + if not line.strip(): + continue + try: + record = json.loads(line) + except json.JSONDecodeError as exc: + raise CanaryError(f"telegram JSONL line {line_number} is invalid JSON: {exc}") from exc + record = _require_object(record, f"telegram JSONL line {line_number}") + chat_id = record.get("chat_id") + message_id = record.get("message_id") + message = record.get("message") + if isinstance(chat_id, bool) or not isinstance(chat_id, int): + raise CanaryError(f"telegram JSONL line {line_number}.chat_id must be an integer") + if isinstance(message_id, bool) or not isinstance(message_id, int): + raise CanaryError(f"telegram JSONL line {line_number}.message_id must be an integer") + if not isinstance(message, str) or not message.strip(): + raise CanaryError(f"telegram JSONL line {line_number}.message must be non-empty") + key = (chat_id, message_id) + if key in seen: + raise CanaryError(f"telegram JSONL repeats chat/message identity {key}") + seen.add(key) + body = message.strip() + segments.append( + { + "id": f"message-{chat_id}-{message_id}", + "body": body, + "locator": f"telegram://chat/{chat_id}/message/{message_id}", + "json_pointer": f"jsonl://line/{line_number}/message", + "content_sha256": canonical_sha256(record), + "text_sha256": sha256_bytes(body.encode()), + "metadata": { + "line_number": line_number, + "ts": record.get("ts"), + "chat_id": chat_id, + "chat_title": record.get("chat_title"), + "message_id": message_id, + "type": record.get("type"), + "username": record.get("username"), + "display_name": record.get("display_name"), + "user_id": record.get("user_id"), + "reply_to": record.get("reply_to"), + "synthetic": record.get("synthetic") is True, + }, + } + ) + if not segments: + raise CanaryError("telegram_jsonl artifact contains no messages") + return segments + + +def _load_segments(raw: bytes, artifact_format: str, source: dict[str, Any]) -> list[dict[str, Any]]: + if artifact_format == "plain_text": + return _load_plain_text_segments(raw, source) + if artifact_format == "telegram_jsonl": + return _load_telegram_jsonl_segments(raw) + raise CanaryError("artifact.format must be plain_text or telegram_jsonl") def load_fixture(path: Path) -> tuple[dict[str, Any], dict[str, Any]]: - raw = path.read_bytes() + scenario_path = path.resolve() try: - fixture = json.loads(raw) - except json.JSONDecodeError as exc: - raise CanaryError(f"fixture is not valid JSON: {exc}") from exc - if not isinstance(fixture, dict) or fixture.get("schema") != FIXTURE_SCHEMA: - raise CanaryError(f"fixture schema must be {FIXTURE_SCHEMA}") - source = fixture.get("source") - extraction = fixture.get("extraction") - claim = extraction.get("claim") if isinstance(extraction, dict) else None - evidence = extraction.get("evidence") if isinstance(extraction, dict) else None - if not all(isinstance(item, dict) for item in (source, claim, evidence)): - raise CanaryError("fixture requires source, extraction.claim, and extraction.evidence objects") - required_source = ("identity", "source_key", "source_type", "title", "locator", "body", "metadata") - required_claim = ("claim_key", "type", "confidence", "text", "body", "metadata") - required_evidence = ("role", "body", "metadata") - for label, value, required in ( - ("source", source, required_source), - ("claim", claim, required_claim), - ("evidence", evidence, required_evidence), - ): - missing = [key for key in required if value.get(key) in (None, "")] - if missing: - raise CanaryError(f"fixture {label} is missing: {', '.join(missing)}") - if not isinstance(value.get("metadata"), dict): - raise CanaryError(f"fixture {label}.metadata must be an object") - if claim["body"] not in source["body"]: - raise CanaryError("extracted claim body must be an exact substring of source.body") - if evidence["body"] not in source["body"]: - raise CanaryError("extracted evidence body must be an exact substring of source.body") + scenario_raw = scenario_path.read_bytes() + scenario = json.loads(scenario_raw) + except (OSError, json.JSONDecodeError) as exc: + raise CanaryError(f"scenario is not readable JSON: {exc}") from exc + if not isinstance(scenario, dict) or scenario.get("schema") != FIXTURE_SCHEMA: + raise CanaryError(f"scenario schema must be {FIXTURE_SCHEMA}") + artifact = _require_object(scenario.get("artifact"), "artifact") + source = _require_object(scenario.get("source"), "source") + extractor = _require_object(scenario.get("extractor"), "extractor") + extraction = _require_object(scenario.get("extraction"), "extraction") + artifact_rel = Path(_require_text(artifact.get("path"), "artifact.path")) + if artifact_rel.is_absolute(): + raise CanaryError("artifact.path must be relative to the scenario") + artifact_path = (scenario_path.parent / artifact_rel).resolve() + try: + artifact_path.relative_to(scenario_path.parent.resolve()) + except ValueError as exc: + raise CanaryError("artifact.path must remain inside the scenario directory") from exc + try: + raw = artifact_path.read_bytes() + except OSError as exc: + raise CanaryError(f"source artifact is unreadable: {exc}") from exc + if not raw: + raise CanaryError("source artifact must not be empty") + artifact_format = _require_text(artifact.get("format"), "artifact.format") + extractor_name = _require_text(extractor.get("name"), "extractor.name") + extractor_version = _require_text(extractor.get("version"), "extractor.version") + required_source = ("identity", "source_key", "source_type", "title", "locator", "metadata") + missing_source = [key for key in required_source if source.get(key) in (None, "")] + if missing_source: + raise CanaryError(f"source is missing: {', '.join(missing_source)}") + if source["source_type"] not in ap.SOURCE_TYPES: + 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} + + claims = _require_list(extraction.get("claims"), "extraction.claims") + duplicates = _require_list(extraction.get("duplicates"), "extraction.duplicates") + conflicts = _require_list(extraction.get("conflicts"), "extraction.conflicts") + if not claims: + raise CanaryError("extraction.claims must contain at least one review candidate") + claim_keys: set[str] = set() + normalized_claims: list[dict[str, Any]] = [] + evidence_count = 0 + for claim_index, raw_claim in enumerate(claims): + claim = _require_object(raw_claim, f"extraction.claims[{claim_index}]") + claim_key = _require_text(claim.get("claim_key"), f"extraction.claims[{claim_index}].claim_key") + if claim_key in claim_keys: + raise CanaryError(f"duplicate claim_key {claim_key}") + claim_keys.add(claim_key) + claim_type = _require_text(claim.get("type"), f"extraction.claims[{claim_index}].type") + if claim_type not in ap.CLAIM_TYPES: + raise CanaryError(f"claim {claim_key} has unsupported type {claim_type}") + confidence = claim.get("confidence") + if isinstance(confidence, bool) or not isinstance(confidence, (int, float)) or not 0 <= confidence <= 1: + raise CanaryError(f"claim {claim_key} confidence must be between 0 and 1") + evidence_rows = _require_list(claim.get("evidence"), f"claim {claim_key}.evidence") + if not evidence_rows: + raise CanaryError(f"claim {claim_key} requires evidence") + normalized_evidence: list[dict[str, Any]] = [] + for evidence_index, raw_evidence in enumerate(evidence_rows): + evidence = _require_object(raw_evidence, f"claim {claim_key}.evidence[{evidence_index}]") + segment_id = _require_text(evidence.get("segment_id"), f"claim {claim_key} evidence segment_id") + segment = segment_by_id.get(segment_id) + if segment is None: + raise CanaryError(f"claim {claim_key} evidence references unknown segment {segment_id}") + excerpt = _require_text(evidence.get("excerpt"), f"claim {claim_key} evidence excerpt") + if excerpt not in segment["body"]: + raise CanaryError(f"claim {claim_key} evidence excerpt is not an exact segment substring") + role = _require_text(evidence.get("role"), f"claim {claim_key} evidence role") + if role not in ap.EVIDENCE_ROLES: + raise CanaryError(f"claim {claim_key} evidence has unsupported role {role}") + normalized_evidence.append( + { + **evidence, + "segment_id": segment_id, + "excerpt": excerpt, + "role": role, + "source_locator": segment["locator"], + "source_json_pointer": segment["json_pointer"], + "source_content_sha256": segment["content_sha256"], + "source_text_sha256": segment["text_sha256"], + } + ) + evidence_count += 1 + body = _require_text(claim.get("body"), f"claim {claim_key}.body") + if not any(body in segment_by_id[row["segment_id"]]["body"] for row in normalized_evidence): + raise CanaryError(f"claim {claim_key} body must be an exact substring of its evidence segments") + edges = _require_list(claim.get("edges", []), f"claim {claim_key}.edges") + normalized_claims.append( + { + **claim, + "claim_key": claim_key, + "type": claim_type, + "confidence": confidence, + "text": _require_text(claim.get("text"), f"claim {claim_key}.text"), + "body": body, + "metadata": _require_object(claim.get("metadata", {}), f"claim {claim_key}.metadata"), + "evidence": normalized_evidence, + "edges": edges, + } + ) + + normalized_duplicates: list[dict[str, Any]] = [] + for duplicate_index, raw_duplicate in enumerate(duplicates): + duplicate = _require_object(raw_duplicate, f"extraction.duplicates[{duplicate_index}]") + segment_id = _require_text(duplicate.get("segment_id"), f"duplicate {duplicate_index}.segment_id") + target = _require_text( + duplicate.get("duplicate_of_claim_key"), f"duplicate {duplicate_index}.duplicate_of_claim_key" + ) + if target not in claim_keys: + raise CanaryError(f"duplicate {duplicate_index} references unknown claim {target}") + segment = segment_by_id.get(segment_id) + if segment is None: + raise CanaryError(f"duplicate {duplicate_index} references unknown segment {segment_id}") + excerpt = _require_text(duplicate.get("excerpt"), f"duplicate {duplicate_index}.excerpt") + if excerpt not in segment["body"]: + raise CanaryError(f"duplicate {duplicate_index} excerpt is not an exact segment substring") + normalized_duplicates.append( + { + **duplicate, + "segment_id": segment_id, + "duplicate_of_claim_key": target, + "excerpt": excerpt, + "source_locator": segment["locator"], + "source_json_pointer": segment["json_pointer"], + "source_content_sha256": segment["content_sha256"], + "source_text_sha256": segment["text_sha256"], + } + ) + + normalized_conflicts: list[dict[str, Any]] = [] + for conflict_index, raw_conflict in enumerate(conflicts): + conflict = _require_object(raw_conflict, f"extraction.conflicts[{conflict_index}]") + from_key = _require_text(conflict.get("from_claim_key"), f"conflict {conflict_index}.from_claim_key") + to_key = _require_text(conflict.get("to_claim_key"), f"conflict {conflict_index}.to_claim_key") + relationship = _require_text(conflict.get("relationship"), f"conflict {conflict_index}.relationship") + if from_key not in claim_keys or to_key not in claim_keys: + raise CanaryError(f"conflict {conflict_index} must reference produced claim keys") + if relationship != "contradicts": + raise CanaryError("fixture conflicts must use relationship=contradicts") + matching_edge = any( + claim["claim_key"] == from_key + and any(edge.get("edge_type") == relationship and edge.get("target") == to_key for edge in claim["edges"]) + for claim in normalized_claims + ) + if not matching_edge: + raise CanaryError(f"conflict {from_key}->{to_key} has no matching candidate edge") + normalized_conflicts.append({**conflict, "from_claim_key": from_key, "to_claim_key": to_key}) + artifact_sha256 = sha256_bytes(raw) - source_content_sha256 = sha256_bytes(source["body"].encode()) - return fixture, { - "path": str(path.resolve()), + segment_manifest = normalized_segment_manifest(segments) + source_content_sha256 = canonical_sha256(segment_manifest) + extraction_spec_sha256 = canonical_sha256(extraction) + 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)}, + "source": source, + "extractor": extractor_identity, + "segments": segments, + "extraction": { + "claims": normalized_claims, + "duplicates": normalized_duplicates, + "conflicts": normalized_conflicts, + }, + } + receipt = { + "scenario_path": str(scenario_path), + "scenario_sha256": sha256_bytes(scenario_raw), + "artifact_path": str(artifact_path), + "artifact_format": artifact_format, "artifact_sha256": artifact_sha256, "source_content_sha256": source_content_sha256, - "source_identity": source["identity"], + "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), + "claim_candidates": len(normalized_claims), + "evidence_candidates": evidence_count, + "duplicate_judgments": len(normalized_duplicates), + "conflict_relationships": len(normalized_conflicts), + }, } + return fixture, receipt -def build_row_ids(artifact_sha256: str) -> dict[str, str]: +def build_row_ids(input_receipt: dict[str, Any], fixture: dict[str, Any]) -> dict[str, Any]: + replay_seed = input_receipt["replay_identity_sha256"] + claim_ids = { + claim["claim_key"]: stable_uuid(replay_seed, f"claim-extraction:{claim['claim_key']}") + for claim in fixture["extraction"]["claims"] + } + evidence_ids: dict[str, str] = {} + for claim in fixture["extraction"]["claims"]: + for index, _evidence in enumerate(claim["evidence"]): + key = f"{claim['claim_key']}:{index}" + evidence_ids[key] = stable_uuid(replay_seed, f"evidence-extraction:{key}") + duplicate_ids = { + str(index): stable_uuid(replay_seed, f"duplicate-assessment:{index}") + for index, _duplicate in enumerate(fixture["extraction"]["duplicates"]) + } + conflict_ids = { + str(index): stable_uuid(replay_seed, f"conflict-assessment:{index}") + for index, _conflict in enumerate(fixture["extraction"]["conflicts"]) + } return { - "source_capture_id": stable_uuid(artifact_sha256, "source-capture"), - "claim_extraction_id": stable_uuid(artifact_sha256, "claim-extraction"), - "evidence_extraction_id": stable_uuid(artifact_sha256, "evidence-extraction"), - "parent_proposal_id": stable_uuid(artifact_sha256, "rich-parent-proposal"), + "source_capture_id": stable_uuid(replay_seed, "source-capture"), + "claim_extraction_ids": claim_ids, + "evidence_extraction_ids": evidence_ids, + "duplicate_assessment_ids": duplicate_ids, + "conflict_assessment_ids": conflict_ids, + "parent_proposal_id": stable_uuid(replay_seed, "rich-parent-proposal"), } +def _segment_source_key(source_key: str, segment_id: str) -> str: + return _safe_source_key(f"{source_key}__{segment_id}") + + def build_parent_packet( - fixture: dict[str, Any], input_receipt: dict[str, Any], row_ids: dict[str, str] + fixture: dict[str, Any], input_receipt: dict[str, Any], row_ids: dict[str, Any] ) -> dict[str, Any]: source = fixture["source"] - claim = fixture["extraction"]["claim"] - evidence = fixture["extraction"]["evidence"] - parent_source_ref = ( - f"local-ingestion:{input_receipt['artifact_sha256']}:" - f"source={row_ids['source_capture_id']}:claim={row_ids['claim_extraction_id']}:" - f"evidence={row_ids['evidence_extraction_id']}" - ) + claims = fixture["extraction"]["claims"] + segment_by_id = {segment["id"]: segment for segment in fixture["segments"]} + 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"]) + + source_candidates = [] + for segment_id in referenced_segment_ids: + segment = segment_by_id[segment_id] + segment_source_key = _segment_source_key(source["source_key"], segment_id) + provenance = { + "artifact_sha256": input_receipt["artifact_sha256"], + "segment_id": segment_id, + "locator": segment["locator"], + "json_pointer": segment["json_pointer"], + "content_sha256": segment["content_sha256"], + "text_sha256": segment["text_sha256"], + } + source_candidates.append( + { + "source_key": segment_source_key, + "source_type": "dm" if input_receipt["artifact_format"] == "telegram_jsonl" else source["source_type"], + "title": f"{source['title']} - {segment_id}", + "storage_path": segment["locator"], + "url": None, + "source_quality": json.dumps(provenance, sort_keys=True, separators=(",", ":")), + "usage_limits": "Review-only exact excerpt; no canonical apply is authorized.", + "key_excerpt": segment["body"], + "content_sha256": segment["content_sha256"], + } + ) + + claim_candidates = [] + for claim in claims: + claim_candidates.append( + { + "claim_key": claim["claim_key"], + "type": claim["type"], + "confidence": claim["confidence"], + "text": claim["text"], + "body": claim["body"], + "evidence_tier": "source_artifact_exact_location", + "needs_research": claim["metadata"].get("needs_research", False), + "evidence": [ + { + "source_key": _segment_source_key(source["source_key"], evidence["segment_id"]), + "role": evidence["role"], + } + for evidence in claim["evidence"] + ], + "edges": claim["edges"], + } + ) + + ingestion_manifest = { + "scenario_schema": fixture["schema"], + "scenario_sha256": input_receipt["scenario_sha256"], + "artifact_format": input_receipt["artifact_format"], + "artifact_sha256": input_receipt["artifact_sha256"], + "source_content_sha256": input_receipt["source_content_sha256"], + "source_identity": input_receipt["source_identity"], + "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": normalized_segment_manifest(fixture["segments"]), + "row_ids": row_ids, + "candidate_counts": input_receipt["counts"], + } return { "id": row_ids["parent_proposal_id"], "proposal_type": "attach_evidence", @@ -111,47 +557,149 @@ def build_parent_packet( "proposed_by_handle": "leo", "proposed_by_agent_id": None, "channel": "local_ingestion_canary", - "source_ref": parent_source_ref, - "rationale": "Stage one hash-bound claim/evidence extraction from a captured local document fixture.", + "source_ref": ( + f"local-ingestion:{input_receipt['artifact_sha256']}:" + f"replay={input_receipt['replay_identity_sha256']}:capture={row_ids['source_capture_id']}" + ), + "rationale": "Stage source-linked candidate claims for review without canonical apply.", "payload": { - "claim_candidates": [ - { - "claim_key": claim["claim_key"], - "type": claim["type"], - "confidence": claim["confidence"], - "text": claim["text"], - "body": claim["body"], - "evidence_tier": "isolated_fixture", - "needs_research": claim["metadata"].get("needs_research", False), - "evidence": [{"source_key": source["source_key"], "role": evidence["role"]}], - "edges": [], - } - ], - "source_candidates": [ - { - "source_key": source["source_key"], - "source_type": source["source_type"], - "title": source["title"], - "storage_path": input_receipt["path"], - "url": None, - "source_quality": "local realistic ingestion fixture", - "key_excerpt": evidence["body"], - "content_sha256": input_receipt["source_content_sha256"], - } - ], + "ingestion_manifest": ingestion_manifest, + "claim_candidates": claim_candidates, + "source_candidates": source_candidates, "dedupe_conflict_assessment": { - "database_search_query": claim["claim_key"].replace("_", " "), - "potential_existing_claim_ids": [], - "conflicts": [], + "database_search_query": " ".join(claim["claim_key"].replace("_", " ") for claim in claims), + "duplicates": fixture["extraction"]["duplicates"], + "conflicts": fixture["extraction"]["conflicts"], + "review_required": True, }, }, } +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 r"""\set ON_ERROR_STOP on + return rf"""\set ON_ERROR_STOP on create schema kb_canary; create schema kb_stage; +create table public.claims ( + id uuid primary key, type text not null, text text not null, status text not null, + confidence double precision, tags text[] not null default '{{}}' +); +create table public.sources ( + id uuid primary key, source_type text not null, url text, storage_path text, + excerpt text, hash text not null +); +create table public.claim_evidence ( + claim_id uuid not null, source_id uuid not null, role text not null, weight double precision, + primary key (claim_id, source_id, role) +); +create table public.claim_edges ( + from_claim uuid not null, to_claim uuid not null, edge_type text not null, weight double precision, + primary key (from_claim, to_claim, edge_type) +); +insert into public.claims(id, type, text, status, confidence, tags) values + ('{SEED_CLAIM_A}', 'structural', 'A staged proposal is non-canonical until guarded apply succeeds.', 'open', 0.99, array['seed']), + ('{SEED_CLAIM_B}', 'meta', 'Review and canonical application are separate state transitions.', 'open', 0.98, array['seed']); +insert into public.sources(id, source_type, storage_path, excerpt, hash) values + ('{SEED_SOURCE}', 'observation', 'fixture://canonical/seed', 'Representative canonical seed.', + '{"a" * 64}'); +insert into public.claim_evidence(claim_id, source_id, role, weight) values + ('{SEED_CLAIM_A}', '{SEED_SOURCE}', 'grounds', 1.0); +insert into public.claim_edges(from_claim, to_claim, edge_type, weight) values + ('{SEED_CLAIM_A}', '{SEED_CLAIM_B}', 'supports', 1.0); create table kb_canary.source_captures ( id uuid primary key, source_identity text not null unique, @@ -159,15 +707,15 @@ create table kb_canary.source_captures ( title text not null, locator text not null, artifact_path text not null, - artifact_sha256 text not null check (artifact_sha256 ~ '^[0-9a-f]{64}$'), - content_sha256 text not null check (content_sha256 ~ '^[0-9a-f]{64}$'), - body text not null, + artifact_sha256 text not null check (artifact_sha256 ~ '^[0-9a-f]{{64}}$'), + content_sha256 text not null check (content_sha256 ~ '^[0-9a-f]{{64}}$'), + replay_identity_sha256 text not null check (replay_identity_sha256 ~ '^[0-9a-f]{{64}}$'), metadata jsonb not null ); create table kb_canary.claim_extractions ( id uuid primary key, source_capture_id uuid not null references kb_canary.source_captures(id), - claim_key text not null, + claim_key text not null unique, body text not null, metadata jsonb not null ); @@ -175,8 +723,32 @@ create table kb_canary.evidence_extractions ( id uuid primary key, claim_extraction_id uuid not null references kb_canary.claim_extractions(id), source_capture_id uuid not null references kb_canary.source_captures(id), + source_segment_id text not null, + source_locator text not null, + source_json_pointer text, role text not null, body text not null, + body_sha256 text not null, + source_content_sha256 text not null, + metadata jsonb not null +); +create table kb_canary.duplicate_assessments ( + id uuid primary key, + source_capture_id uuid not null references kb_canary.source_captures(id), + source_segment_id text not null, + source_locator text not null, + duplicate_of_claim_key text not null, + excerpt text not null, + excerpt_sha256 text not null, + reason text not null, + metadata jsonb not null +); +create table kb_canary.conflict_assessments ( + id uuid primary key, + source_capture_id uuid not null references kb_canary.source_captures(id), + from_claim_key text not null, + to_claim_key text not null, + relationship text not null, metadata jsonb not null ); create table kb_stage.kb_proposals ( @@ -200,59 +772,221 @@ create table kb_stage.kb_proposals ( """ -def build_capture_sql(fixture: dict[str, Any], input_receipt: dict[str, Any], row_ids: dict[str, str]) -> str: - source = fixture["source"] - claim = fixture["extraction"]["claim"] - evidence = fixture["extraction"]["evidence"] - q = ap.sql_literal - claim_metadata = { - **claim["metadata"], - "claim_type": claim["type"], - "confidence": claim["confidence"], - "text": claim["text"], - } - return rf"""\set ON_ERROR_STOP on -begin; -insert into kb_canary.source_captures - (id, source_identity, source_type, title, locator, artifact_path, - artifact_sha256, content_sha256, body, metadata) -values - ({q(row_ids["source_capture_id"])}::uuid, {q(source["identity"])}, {q(source["source_type"])}, - {q(source["title"])}, {q(source["locator"])}, {q(input_receipt["path"])}, - {q(input_receipt["artifact_sha256"])}, {q(input_receipt["source_content_sha256"])}, - {q(source["body"])}, {q(json.dumps(source["metadata"], sort_keys=True))}::jsonb); -insert into kb_canary.claim_extractions - (id, source_capture_id, claim_key, body, metadata) -values - ({q(row_ids["claim_extraction_id"])}::uuid, {q(row_ids["source_capture_id"])}::uuid, - {q(claim["claim_key"])}, {q(claim["body"])}, {q(json.dumps(claim_metadata, sort_keys=True))}::jsonb); -insert into kb_canary.evidence_extractions - (id, claim_extraction_id, source_capture_id, role, body, metadata) -values - ({q(row_ids["evidence_extraction_id"])}::uuid, {q(row_ids["claim_extraction_id"])}::uuid, - {q(row_ids["source_capture_id"])}::uuid, {q(evidence["role"])}, {q(evidence["body"])}, - {q(json.dumps(evidence["metadata"], sort_keys=True))}::jsonb); -commit; +def build_canonical_snapshot_sql() -> str: + return r"""\set ON_ERROR_STOP on +begin transaction read only; +select jsonb_build_object( + 'claims', coalesce((select jsonb_agg(to_jsonb(t) order by t.id) from public.claims t), '[]'::jsonb), + 'sources', coalesce((select jsonb_agg(to_jsonb(t) order by t.id) from public.sources t), '[]'::jsonb), + 'claim_evidence', coalesce((select jsonb_agg(to_jsonb(t) order by t.claim_id, t.source_id, t.role) + from public.claim_evidence t), '[]'::jsonb), + 'claim_edges', coalesce((select jsonb_agg(to_jsonb(t) order by t.from_claim, t.to_claim, t.edge_type) + from public.claim_edges t), '[]'::jsonb) +)::text; +rollback; """ -def build_readback_sql(row_ids: dict[str, str], proposal_id: str) -> str: +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 + statements = [ + r"\set ON_ERROR_STOP on", + "begin;", + "insert into kb_canary.source_captures " + "(id, source_identity, source_type, title, locator, artifact_path, artifact_sha256, content_sha256, " + "replay_identity_sha256, metadata) values " + f"({q(row_ids['source_capture_id'])}::uuid, {q(source['identity'])}, {q(source['source_type'])}, " + f"{q(source['title'])}, {q(source['locator'])}, {q(input_receipt['artifact_path'])}, " + f"{q(input_receipt['artifact_sha256'])}, {q(input_receipt['source_content_sha256'])}, " + f"{q(input_receipt['replay_identity_sha256'])}, " + f"{q(json.dumps({**source['metadata'], 'extractor': input_receipt['extractor']}, sort_keys=True))}::jsonb);", + ] + 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_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"], + } + statements.append( + "insert into kb_canary.claim_extractions " + "(id, source_capture_id, claim_key, body, metadata) values " + f"({q(claim_id)}::uuid, {q(row_ids['source_capture_id'])}::uuid, {q(claim['claim_key'])}, " + f"{q(claim['body'])}, {q(json.dumps(claim_metadata, sort_keys=True))}::jsonb);" + ) + for index, evidence in enumerate(claim["evidence"]): + evidence_key = f"{claim['claim_key']}:{index}" + segment = segment_by_id[evidence["segment_id"]] + evidence_metadata = { + **_require_object(evidence.get("metadata", {}), f"evidence {evidence_key}.metadata"), + "source_text_sha256": segment["text_sha256"], + } + statements.append( + "insert into kb_canary.evidence_extractions " + "(id, claim_extraction_id, source_capture_id, source_segment_id, source_locator, " + "source_json_pointer, role, body, body_sha256, source_content_sha256, metadata) values " + f"({q(row_ids['evidence_extraction_ids'][evidence_key])}::uuid, {q(claim_id)}::uuid, " + f"{q(row_ids['source_capture_id'])}::uuid, {q(segment['id'])}, {q(segment['locator'])}, " + f"{q(segment['json_pointer'])}, {q(evidence['role'])}, {q(evidence['excerpt'])}, " + f"{q(sha256_bytes(evidence['excerpt'].encode()))}, {q(segment['content_sha256'])}, " + f"{q(json.dumps(evidence_metadata, sort_keys=True))}::jsonb);" + ) + for index, duplicate in enumerate(fixture["extraction"]["duplicates"]): + statements.append( + "insert into kb_canary.duplicate_assessments " + "(id, source_capture_id, source_segment_id, source_locator, duplicate_of_claim_key, excerpt, " + "excerpt_sha256, reason, metadata) values " + f"({q(row_ids['duplicate_assessment_ids'][str(index)])}::uuid, " + f"{q(row_ids['source_capture_id'])}::uuid, {q(duplicate['segment_id'])}, " + f"{q(duplicate['source_locator'])}, {q(duplicate['duplicate_of_claim_key'])}, " + f"{q(duplicate['excerpt'])}, {q(sha256_bytes(duplicate['excerpt'].encode()))}, " + f"{q(duplicate['reason'])}, {q(json.dumps({'json_pointer': duplicate['source_json_pointer'], 'source_content_sha256': duplicate['source_content_sha256'], 'source_text_sha256': duplicate['source_text_sha256']}, sort_keys=True))}::jsonb);" + ) + for index, conflict in enumerate(fixture["extraction"]["conflicts"]): + statements.append( + "insert into kb_canary.conflict_assessments " + "(id, source_capture_id, from_claim_key, to_claim_key, relationship, metadata) values " + f"({q(row_ids['conflict_assessment_ids'][str(index)])}::uuid, " + f"{q(row_ids['source_capture_id'])}::uuid, {q(conflict['from_claim_key'])}, " + f"{q(conflict['to_claim_key'])}, {q(conflict['relationship'])}, " + f"{q(json.dumps({key: value for key, value in conflict.items() if key not in {'from_claim_key', 'to_claim_key', 'relationship'}}, sort_keys=True))}::jsonb);" + ) + statements.append("commit;") + return "\n".join(statements) + "\n" + + +def build_readback_sql(proposal_id: str) -> str: q = ap.sql_literal return rf"""\set ON_ERROR_STOP on begin transaction read only; select jsonb_build_object( - 'source_capture', (select to_jsonb(s) from kb_canary.source_captures s - where id = {q(row_ids["source_capture_id"])}::uuid), - 'claim_extraction', (select to_jsonb(c) from kb_canary.claim_extractions c - where id = {q(row_ids["claim_extraction_id"])}::uuid), - 'evidence_extraction', (select to_jsonb(e) from kb_canary.evidence_extractions e - where id = {q(row_ids["evidence_extraction_id"])}::uuid), - 'staged_proposal', (select to_jsonb(p) from kb_stage.kb_proposals p - where id = {q(proposal_id)}::uuid), + 'source_captures', coalesce((select jsonb_agg(to_jsonb(t) order by t.id) from kb_canary.source_captures t), '[]'::jsonb), + 'claim_extractions', coalesce((select jsonb_agg(to_jsonb(t) order by t.claim_key) from kb_canary.claim_extractions t), '[]'::jsonb), + 'evidence_extractions', coalesce((select jsonb_agg(to_jsonb(t) order by t.id) from kb_canary.evidence_extractions t), '[]'::jsonb), + 'duplicate_assessments', coalesce((select jsonb_agg(to_jsonb(t) order by t.id) from kb_canary.duplicate_assessments t), '[]'::jsonb), + 'conflict_assessments', coalesce((select jsonb_agg(to_jsonb(t) order by t.id) from kb_canary.conflict_assessments t), '[]'::jsonb), + 'staged_proposal', (select to_jsonb(p) from kb_stage.kb_proposals p where id = {q(proposal_id)}::uuid), 'counts', jsonb_build_object( 'source_captures', (select count(*) from kb_canary.source_captures), 'claim_extractions', (select count(*) from kb_canary.claim_extractions), 'evidence_extractions', (select count(*) from kb_canary.evidence_extractions), + 'duplicate_assessments', (select count(*) from kb_canary.duplicate_assessments), + 'conflict_assessments', (select count(*) from kb_canary.conflict_assessments), 'staged_proposals', (select count(*) from kb_stage.kb_proposals) ) )::text; @@ -320,11 +1054,7 @@ class DockerPostgres: if inspect["Config"]["Labels"].get("livingip.canary") != "working-leo-local-ingestion": raise CanaryError("disposable Postgres is missing its canary label") self.volume_mounts = [ - { - "type": mount.get("Type"), - "name": mount.get("Name"), - "destination": mount.get("Destination"), - } + {"type": mount.get("Type"), "name": mount.get("Name"), "destination": mount.get("Destination")} for mount in inspect.get("Mounts", []) if mount.get("Type") == "volume" ] @@ -372,128 +1102,332 @@ 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 ], - "source_content_hash_matches": any( - row.get("hash") == input_receipt["source_content_sha256"] for row in source_rows - ), - "parent_source_ref_embedded": any( - readback["source_capture"]["artifact_sha256"] in str(row.get("excerpt")) - and readback["source_capture"]["id"] in str(row.get("excerpt")) - and readback["claim_extraction"]["id"] in str(row.get("excerpt")) - and readback["evidence_extraction"]["id"] in str(row.get("excerpt")) - for row in source_rows + "planned_edge_keys": [ + {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(claim_rows), + "sources": len(source_rows), + "evidence": len(evidence_rows), + "edges": len(edge_rows), + }, + "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": ( + links["manifest_replay_identity_sha256"] == input_receipt["replay_identity_sha256"] ), } def evaluate_checks( - fixture: dict[str, Any], input_receipt: dict[str, Any], row_ids: dict[str, str], readback: dict[str, Any] + fixture: dict[str, Any], + input_receipt: dict[str, Any], + row_ids: dict[str, Any], + readback: dict[str, Any], + canonical_before: dict[str, Any], + canonical_after: dict[str, Any], ) -> dict[str, bool]: - source = readback.get("source_capture") or {} - claim = readback.get("claim_extraction") or {} - evidence = readback.get("evidence_extraction") or {} - proposal = readback.get("staged_proposal") or {} - links = proposal_links(readback, input_receipt) + 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) + 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"], + "evidence_extractions": counts["evidence_candidates"], + "duplicate_assessments": counts["duplicate_judgments"], + "conflict_assessments": counts["conflict_relationships"], + "staged_proposals": 1, + } + 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 { - "input_artifact_hash_persisted": source.get("artifact_sha256") == input_receipt["artifact_sha256"], - "source_content_hash_persisted": source.get("content_sha256") == input_receipt["source_content_sha256"], - "source_identity_persisted": source.get("source_identity") == input_receipt["source_identity"], - "claim_links_to_source_capture": claim.get("source_capture_id") == row_ids["source_capture_id"], - "claim_body_and_metadata_persisted": ( - claim.get("body") == fixture["extraction"]["claim"]["body"] - and claim.get("metadata", {}).get("text") == fixture["extraction"]["claim"]["text"] + "source_only_artifact_hash_persisted": ( + len(source_rows) == 1 and source_row.get("artifact_sha256") == input_receipt["artifact_sha256"] ), - "evidence_links_to_claim_and_source": ( - evidence.get("claim_extraction_id") == row_ids["claim_extraction_id"] - and evidence.get("source_capture_id") == row_ids["source_capture_id"] + "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_row.get("replay_identity_sha256") == input_receipt["replay_identity_sha256"] ), - "evidence_body_and_metadata_persisted": ( - evidence.get("body") == fixture["extraction"]["evidence"]["body"] - and evidence.get("metadata") == fixture["extraction"]["evidence"]["metadata"] + "extractor_name_and_version_persisted": ( + ingestion_manifest.get("extractor") == input_receipt["extractor"] + 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 + ) ), - "one_row_per_ingestion_stage": readback.get("counts") - == {"source_captures": 1, "claim_extractions": 1, "evidence_extractions": 1, "staged_proposals": 1}, - "proposal_is_pending_review": proposal.get("status") == "pending_review", - "proposal_has_source_ref": bool(proposal.get("source_ref")), - "proposal_payload_contains_input_hash": links["source_content_hash_matches"], - "proposal_payload_contains_capture_row_linkage": links["parent_source_ref_embedded"], - "no_review_or_apply_metadata": all( - proposal.get(key) is None - for key in ("reviewed_by_handle", "reviewed_at", "applied_by_handle", "applied_at") + "all_claim_candidates_persisted_once": ( + 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_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_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": pending_state["status_matches"], + "proposal_has_replayable_ingestion_manifest": ( + 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": ( + 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) + ), + "canonical_fingerprint_unchanged": ( + canonical_snapshot_complete(canonical_before) + and canonical_snapshot_complete(canonical_after) + and canonical_sha256(canonical_before) == canonical_sha256(canonical_after) ), } 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["artifact_sha256"]) - 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": "isolated_realistic_ingestion_to_proposal", - "input": input_receipt, - "row_ids": row_ids, + "required_tier": "T2_runtime_to_review_queue_staging", + "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()) postgres.psql(build_capture_sql(fixture, input_receipt, row_ids)) stage_result = postgres.psql_json(normalized_stage.build_stage_sql(child)) - readback = postgres.psql_json(build_readback_sql(row_ids, child["id"])) - checks = evaluate_checks(fixture, input_receipt, row_ids, readback) + readback = postgres.psql_json(build_readback_sql(child["id"])) + canonical_after = postgres.psql_json(build_canonical_snapshot_sql()) + canonical_unchanged = ( + canonical_snapshot_complete(canonical_before) + and canonical_snapshot_complete(canonical_after) + and canonical_sha256(canonical_before) == canonical_sha256(canonical_after) + ) + checks = evaluate_checks(fixture, input_receipt, row_ids, readback, canonical_before, canonical_after) receipt.update( { "stage_command_result": stage_result, "rows": readback, - "row_link_fields_proven": proposal_links(readback, input_receipt), + "candidate_counts": input_receipt["counts"], + "row_link_fields_proven": proposal_link_proof(readback, input_receipt), + "canonical": { + "fingerprint_before": canonical_sha256(canonical_before), + "fingerprint_after": canonical_sha256(canonical_after), + "table_counts": {key: len(canonical_before[key]) for key in CANONICAL_TABLES}, + "unchanged": canonical_unchanged, + }, "checks": checks, "status": "pass" if all(checks.values()) else "fail", } ) - except (CanaryError, OSError, ValueError, KeyError) 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" - ) - receipt["cleanup"]["no_production_target_referenced"] = receipt[ - "cleanup" - ]["network_mode_none"] - if not all( - ( - receipt["cleanup"]["container_absent"], - receipt["cleanup"]["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"] = ( - "A realistic local document fixture was hash-captured, replayed through deterministic claim/evidence " - "extraction, normalized by the existing rich-proposal normalizer, and staged/read back as one " - "pending_review proposal in disposable networkless Postgres." - ) + 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 does not prove live generative extraction, arbitrary document coverage, canonical apply, or any " - "hosted/production runtime." + "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." ) if output: output.parent.mkdir(parents=True, exist_ok=True) @@ -501,16 +1435,98 @@ 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": 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), + "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: + output.parent.mkdir(parents=True, exist_ok=True) + output.write_text(json.dumps(suite, indent=2, sort_keys=True) + "\n", encoding="utf-8") + return suite + + def parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument("--fixture", type=Path, default=DEFAULT_FIXTURE) + parser.add_argument( + "--fixture", + type=Path, + action="append", + help="repeatable scenario path; defaults to the document and Telegram scenarios", + ) parser.add_argument("--output", type=Path) return parser.parse_args(argv) def main(argv: list[str] | None = None) -> int: args = parse_args(argv) - receipt = run_canary(args.fixture.resolve(), args.output.resolve() if args.output else None) + fixtures = args.fixture or list(DEFAULT_FIXTURES) + receipt = run_suite([path.resolve() for path in fixtures], args.output.resolve() if args.output else None) print(json.dumps(receipt, sort_keys=True, separators=(",", ":"))) return 0 if receipt["status"] == "pass" else 1 diff --git a/scripts/stage_normalized_proposal.py b/scripts/stage_normalized_proposal.py index 9750929..9fe9a56 100644 --- a/scripts/stage_normalized_proposal.py +++ b/scripts/stage_normalized_proposal.py @@ -57,6 +57,12 @@ def prepare_normalized_child(parent: dict[str, Any]) -> dict[str, Any]: manifest = apply_payload.setdefault("normalization_manifest", {}) if not isinstance(manifest, dict): raise bound.CheckpointError("normalized child normalization_manifest must be an object") + parent_payload = parent.get("payload") + ingestion_manifest = parent_payload.get("ingestion_manifest") if isinstance(parent_payload, dict) else None + if ingestion_manifest is not None: + if not isinstance(ingestion_manifest, dict): + raise bound.CheckpointError("parent payload.ingestion_manifest must be an object") + manifest["ingestion_manifest"] = json.loads(json.dumps(ingestion_manifest, sort_keys=True)) parent_packet_sha256 = canonical_sha256(parent) manifest["parent_packet_sha256"] = parent_packet_sha256 child_payload_sha256 = canonical_sha256(payload) diff --git a/tests/test_prepare_kb_source_manifest.py b/tests/test_prepare_kb_source_manifest.py index ec41f52..d85a0c2 100644 --- a/tests/test_prepare_kb_source_manifest.py +++ b/tests/test_prepare_kb_source_manifest.py @@ -123,11 +123,17 @@ def test_prepare_builds_private_compiler_validated_manifest_without_db_write( assert manifest["dedupe"]["duplicates"][0]["claim_id"] == CANDIDATE_ID assert manifest["claims"][0]["quote"] == CLAIM_QUOTE assert manifest["dedupe"]["duplicates"][0]["source_quote"] == DUPLICATE_QUOTE - assert receipt["extraction"]["selected_source_references"] == [ - {"kind": "claim", "reference": "S00003"}, - {"kind": "evidence", "reference": "S00003"}, - {"kind": "duplicate", "reference": "S00002"}, + selected = receipt["extraction"]["selected_source_references"] + assert [(row["kind"], row["reference"]) for row in selected] == [ + ("claim", "S00003"), + ("evidence", "S00003"), + ("duplicate", "S00002"), ] + assert all(row["char_end"] > row["char_start"] for row in selected) + assert all(len(row["quote_sha256"]) == 64 for row in selected) + assert receipt["replay"]["exact_selected_locations_validated"] is True + assert receipt["extraction"]["evidence_count"] == 1 + assert receipt["extraction"]["duplicate_judgment_count"] == 1 assert stat.S_IMODE(output_dir.stat().st_mode) == 0o700 for path in output_dir.iterdir(): assert stat.S_IMODE(path.stat().st_mode) == 0o600 @@ -173,6 +179,63 @@ def test_source_fragment_ids_are_dense_across_blank_lines() -> None: assert fragments == {"S00001": "first", "S00002": "second"} +def test_source_fragment_index_trims_indentation_but_preserves_exact_offsets() -> None: + text = " indented Unicode: \u201creviewed\u201d \nnext\n" + + fragments, locations = preparer.build_source_fragment_index(text) + + assert fragments["S00001"] == "indented Unicode: \u201creviewed\u201d" + selected = locations["S00001"] + assert selected["line"] == 1 + assert text[selected["char_start"] : selected["char_end"]] == fragments["S00001"] + + +def test_bundle_location_validation_rejects_ambiguous_rebound() -> None: + text = "same line\nsame line\n" + fragments, locations = preparer.build_source_fragment_index(text) + extraction = preparer.resolve_model_output( + { + "claims": [ + { + "claim_key": "second_occurrence", + "type": "structural", + "text": "The second occurrence was selected.", + "quote_ref": "S00002", + "confidence": 0.5, + "tags": [], + "evidence": [{"quote_ref": "S00002", "role": "grounds"}], + } + ], + "duplicates": [], + "extraction_notes": "Select the second repeated line.", + }, + fragments, + locations, + ) + fake_bundle = { + "quote_bindings": [ + { + "kind": "claim", + "claim_key": "second_occurrence", + "quote": "same line", + "first_start": 0, + "first_end": 9, + }, + { + "kind": "evidence", + "claim_key": "second_occurrence", + "evidence_role": "grounds", + "quote": "same line", + "first_start": 0, + "first_end": 9, + }, + ] + } + + with pytest.raises(preparer.PreparationError, match="ambiguous repeated quote"): + preparer.validate_bundle_source_locations(fake_bundle, extraction["selected_source_references"]) + + def test_prepare_returns_no_novel_claims_without_manifest_or_stageable_packet( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: @@ -195,6 +258,13 @@ def test_binary_artifact_requires_explicit_utf8_extraction(tmp_path: Path) -> No preparer.extract_utf8_text(artifact, None) +def test_repo_native_jsonl_transcript_is_accepted_as_strict_utf8(tmp_path: Path) -> None: + artifact = tmp_path / "telegram.jsonl" + artifact.write_text('{"chat_id":1,"message_id":2,"message":"hello"}\n', encoding="utf-8") + + assert preparer.extract_utf8_text(artifact, None) == artifact.read_bytes() + + def test_openrouter_key_cannot_be_redirected_to_an_unapproved_endpoint(tmp_path: Path) -> None: with pytest.raises(preparer.PreparationError, match=r"must equal https://openrouter\.ai"): preparer.call_openrouter( diff --git a/tests/test_run_leo_local_ingestion_proposal_canary.py b/tests/test_run_leo_local_ingestion_proposal_canary.py index ec57f0b..1290dc4 100644 --- a/tests/test_run_leo_local_ingestion_proposal_canary.py +++ b/tests/test_run_leo_local_ingestion_proposal_canary.py @@ -1,6 +1,7 @@ from __future__ import annotations import copy +import hashlib import json import sys from pathlib import Path @@ -13,90 +14,420 @@ sys.path.insert(0, str(REPO_ROOT / "scripts")) import run_leo_local_ingestion_proposal_canary as canary # noqa: E402 -def _loaded() -> tuple[dict, dict, dict]: - fixture, input_receipt = canary.load_fixture(canary.DEFAULT_FIXTURE) - row_ids = canary.build_row_ids(input_receipt["artifact_sha256"]) - return fixture, input_receipt, row_ids - - -def test_fixture_is_realistic_hash_bound_and_exactly_evidenced() -> None: - fixture, input_receipt, row_ids = _loaded() - - assert fixture["extraction"]["evidence"]["body"] in fixture["source"]["body"] - assert len(input_receipt["artifact_sha256"]) == 64 - assert len(input_receipt["source_content_sha256"]) == 64 - assert input_receipt["artifact_sha256"] != input_receipt["source_content_sha256"] - assert len(set(row_ids.values())) == 4 - - -def test_fixture_validation_fails_closed_when_evidence_is_not_in_source(tmp_path: Path) -> None: - fixture = json.loads(canary.DEFAULT_FIXTURE.read_text(encoding="utf-8")) - fixture["extraction"]["evidence"]["body"] = "invented evidence" - path = tmp_path / "bad-fixture.json" - path.write_text(json.dumps(fixture), encoding="utf-8") - - with pytest.raises(canary.CanaryError, match="exact substring"): - canary.load_fixture(path) - - -def test_fixture_validation_fails_closed_when_claim_body_is_not_in_source( - tmp_path: Path, -) -> None: - fixture = json.loads(canary.DEFAULT_FIXTURE.read_text(encoding="utf-8")) - fixture["extraction"]["claim"]["body"] = "invented claim body" - path = tmp_path / "bad-claim-fixture.json" - path.write_text(json.dumps(fixture), encoding="utf-8") - - with pytest.raises(canary.CanaryError, match="claim body must be an exact substring"): - canary.load_fixture(path) - - -def test_capture_sql_escapes_quotes_and_backslashes_from_fixture( - tmp_path: Path, -) -> None: - fixture = json.loads(canary.DEFAULT_FIXTURE.read_text(encoding="utf-8")) - fixture["source"]["title"] = "O'Brien\\review" - fixture["source"]["metadata"]["note"] = "quoted ' value \\ path" - path = tmp_path / "quoted-fixture.json" - path.write_text(json.dumps(fixture), encoding="utf-8") - loaded, input_receipt = canary.load_fixture(path) - row_ids = canary.build_row_ids(input_receipt["artifact_sha256"]) - - sql = canary.build_capture_sql(loaded, input_receipt, row_ids) - - assert canary.ap.sql_literal(fixture["source"]["title"]) in sql - assert canary.ap.sql_literal( - json.dumps(fixture["source"]["metadata"], sort_keys=True) - ) in sql - - -def test_parent_normalizes_to_hash_bound_pending_proposal_with_embedded_row_links() -> None: - fixture, input_receipt, row_ids = _loaded() +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) - payload = child["payload"]["apply_payload"] - - assert child["status"] == "pending_review" - assert child["proposal_type"] == "approve_claim" - assert len(payload["claims"]) == 1 - assert len(payload["sources"]) == 2 - assert len(payload["evidence"]) == 2 - assert input_receipt["source_content_sha256"] in {row["hash"] for row in payload["sources"]} - excerpts = "\n".join(row["excerpt"] for row in payload["sources"]) - assert input_receipt["artifact_sha256"] in excerpts - assert row_ids["source_capture_id"] in excerpts - assert row_ids["claim_extraction_id"] in excerpts - assert row_ids["evidence_extraction_id"] in excerpts + return fixture, input_receipt, row_ids, parent, child -def test_capture_and_stage_sql_never_mutate_canonical_public_tables() -> None: - fixture, input_receipt, row_ids = _loaded() - child = canary.normalized_stage.prepare_normalized_child( - canary.build_parent_packet(fixture, input_receipt, row_ids) +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_schema_sql(), canary.build_capture_sql(fixture, input_receipt, row_ids), canary.normalized_stage.build_stage_sql(child), ) @@ -106,59 +437,346 @@ def test_capture_and_stage_sql_never_mutate_canonical_public_tables() -> None: 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_canary.claim_extractions" in sql - assert "insert into kb_canary.evidence_extractions" in sql assert "insert into kb_stage.kb_proposals" in sql + assert parent["payload"]["ingestion_manifest"]["row_ids"] == row_ids -def test_check_evaluation_requires_every_row_link_and_staging_boundary() -> None: - fixture, input_receipt, row_ids = _loaded() - child = canary.normalized_stage.prepare_normalized_child( - canary.build_parent_packet(fixture, input_receipt, 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 ) - apply_payload = child["payload"]["apply_payload"] - readback = { - "source_capture": { - "id": row_ids["source_capture_id"], - "artifact_sha256": input_receipt["artifact_sha256"], - "content_sha256": input_receipt["source_content_sha256"], - "source_identity": input_receipt["source_identity"], - }, - "claim_extraction": { - "id": row_ids["claim_extraction_id"], - "source_capture_id": row_ids["source_capture_id"], - "body": fixture["extraction"]["claim"]["body"], - "metadata": {"text": fixture["extraction"]["claim"]["text"]}, - }, - "evidence_extraction": { - "id": row_ids["evidence_extraction_id"], - "claim_extraction_id": row_ids["claim_extraction_id"], - "source_capture_id": row_ids["source_capture_id"], - "body": fixture["extraction"]["evidence"]["body"], - "metadata": fixture["extraction"]["evidence"]["metadata"], - }, - "staged_proposal": { - "id": child["id"], - "status": "pending_review", - "source_ref": child["source_ref"], - "payload": {"apply_payload": apply_payload}, - "reviewed_by_handle": None, - "reviewed_at": None, - "applied_by_handle": None, - "applied_at": None, - }, - "counts": { - "source_captures": 1, - "claim_extractions": 1, - "evidence_extractions": 1, - "staged_proposals": 1, - }, + + +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) + checks = canary.evaluate_checks(fixture, input_receipt, row_ids, readback, canonical, copy.deepcopy(canonical)) assert all(checks.values()) broken = copy.deepcopy(readback) - broken["evidence_extraction"]["claim_extraction_id"] = canary.stable_uuid("0" * 64, "wrong") + broken["evidence_extractions"][0]["source_locator"] = "telegram://chat/900001/message/9999" assert ( - canary.evaluate_checks(fixture, input_receipt, row_ids, broken)["evidence_links_to_claim_and_source"] is False + 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"] diff --git a/tests/test_stage_normalized_proposal.py b/tests/test_stage_normalized_proposal.py index 09f86d7..a4309a8 100644 --- a/tests/test_stage_normalized_proposal.py +++ b/tests/test_stage_normalized_proposal.py @@ -116,6 +116,37 @@ def test_invalid_source_hash_refuses_to_prepare_a_child() -> None: stage.prepare_normalized_child(parent) +def test_prepare_normalized_child_preserves_replayable_ingestion_manifest() -> None: + parent = _parent() + parent["payload"]["ingestion_manifest"] = { + "artifact_sha256": "a" * 64, + "extractor": {"name": "deterministic_fixture_replay", "version": "2"}, + "replay_identity_sha256": "b" * 64, + "segments": [ + { + "id": "message-1", + "locator": "telegram://chat/1/message/1", + "content_sha256": "c" * 64, + } + ], + } + + child = stage.prepare_normalized_child(parent) + + assert ( + child["payload"]["apply_payload"]["normalization_manifest"]["ingestion_manifest"] + == parent["payload"]["ingestion_manifest"] + ) + + +def test_prepare_normalized_child_rejects_non_object_ingestion_manifest() -> None: + parent = _parent() + parent["payload"]["ingestion_manifest"] = "not-an-object" + + with pytest.raises(stage.bound.CheckpointError, match="ingestion_manifest must be an object"): + stage.prepare_normalized_child(parent) + + def test_stage_sql_validates_exact_pending_row_before_commit_and_never_writes_public() -> None: child = stage.prepare_normalized_child(_parent()) sql = stage.build_stage_sql(child)