#!/usr/bin/env python3 """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 import time import uuid from pathlib import Path from typing import Any HERE = Path(__file__).resolve().parent REPO_ROOT = HERE.parent 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.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): """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(canonical_json_bytes(value)) def normalized_segment_manifest(segments: list[dict[str, Any]]) -> list[dict[str, Any]]: """Return the complete normalized provenance identity for replay hashing.""" return [ { "id": segment["id"], "locator": segment["locator"], "json_pointer": segment["json_pointer"], "content_sha256": segment["content_sha256"], "text_sha256": segment["text_sha256"], } for segment in segments ] def replay_identity_payload( *, artifact_sha256: str, artifact_format: str, source_identity: str, source_locator: str, normalized_segments_sha256: str, extractor: dict[str, str], extraction_spec_sha256: str, ) -> dict[str, Any]: """Build the explicit source-to-extraction identity used by every row ID.""" return { "artifact_sha256": artifact_sha256, "artifact_format": artifact_format, "source_identity": source_identity, "source_locator": source_locator, "normalized_segments_sha256": normalized_segments_sha256, "extractor": extractor, "extraction_spec_sha256": extraction_spec_sha256, } def canonical_snapshot_complete(snapshot: dict[str, Any]) -> bool: return set(snapshot) == set(CANONICAL_TABLES) and all( isinstance(snapshot.get(table), list) and bool(snapshot[table]) for table in CANONICAL_TABLES ) 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]]: scenario_path = path.resolve() try: 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) 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_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(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(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, Any] ) -> dict[str, Any]: source = fixture["source"] 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", "status": "pending_review", "proposed_by_handle": "leo", "proposed_by_agent_id": None, "channel": "local_ingestion_canary", "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": { "ingestion_manifest": ingestion_manifest, "claim_candidates": claim_candidates, "source_candidates": source_candidates, "dedupe_conflict_assessment": { "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 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, source_type text not null, 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}}$'), 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 unique, body text not null, metadata jsonb not null ); 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 ( id uuid primary key, proposal_type text not null, status text not null, proposed_by_handle text, proposed_by_agent_id uuid, channel text not null, source_ref text not null, rationale text not null, payload jsonb not null, reviewed_by_handle text, reviewed_by_agent_id uuid, reviewed_at timestamptz, review_note text, applied_by_handle text, applied_by_agent_id uuid, applied_at timestamptz ); """ 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 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_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; rollback; """ class DockerPostgres: def __init__(self, name: str) -> None: self.name = name self.docker = shutil.which("docker") if not self.docker: raise CanaryError("docker executable is unavailable") self.started = False self.volume_mounts: list[dict[str, Any]] = [] def _run( self, args: list[str], *, input_text: str | None = None, check: bool = True ) -> subprocess.CompletedProcess[str]: result = subprocess.run([self.docker, *args], input=input_text, text=True, capture_output=True, check=False) if check and result.returncode != 0: detail = (result.stderr or result.stdout).strip() raise CanaryError(f"docker {' '.join(args[:3])} failed ({result.returncode}): {detail}") return result def start(self) -> dict[str, Any]: result = self._run( [ "run", "--rm", "-d", "--name", self.name, "--network", "none", "--label", CONTAINER_LABEL, "--tmpfs", "/var/lib/postgresql/data:rw,nosuid,size=256m", "-e", "POSTGRES_HOST_AUTH_METHOD=trust", "-e", "POSTGRES_DB=teleo", "postgres:16-alpine", ] ) self.started = True container_id = result.stdout.strip() deadline = time.monotonic() + 30 ready_streak = 0 while time.monotonic() < deadline: ready = self._run(["exec", self.name, "pg_isready", "-U", "postgres", "-d", "teleo"], check=False) if ready.returncode == 0: ready_streak += 1 if ready_streak >= 2: break else: ready_streak = 0 time.sleep(0.25) else: raise CanaryError("disposable Postgres did not become ready within 30 seconds") inspect = json.loads(self._run(["inspect", self.name]).stdout)[0] if inspect["HostConfig"]["NetworkMode"] != "none": raise CanaryError("disposable Postgres must use network mode none") 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")} for mount in inspect.get("Mounts", []) if mount.get("Type") == "volume" ] if self.volume_mounts: raise CanaryError("disposable Postgres unexpectedly created a Docker volume") return { "container_id": container_id, "name": self.name, "image": inspect["Config"]["Image"], "network_mode": inspect["HostConfig"]["NetworkMode"], "tmpfs_data_dir": "/var/lib/postgresql/data" in inspect["HostConfig"].get("Tmpfs", {}), "docker_volume_mounts": self.volume_mounts, "canary_label": inspect["Config"]["Labels"].get("livingip.canary"), } def psql(self, sql: str) -> str: result = self._run( ["exec", "-i", self.name, "psql", "-U", "postgres", "-d", "teleo", "-Atq"], input_text=sql, ) return result.stdout.strip() def psql_json(self, sql: str) -> dict[str, Any]: output = self.psql(sql) lines = [line for line in output.splitlines() if line.strip()] if not lines: raise CanaryError("Postgres readback returned no JSON") try: value = json.loads(lines[-1]) except json.JSONDecodeError as exc: raise CanaryError(f"Postgres readback was not JSON: {lines[-1]}") from exc if not isinstance(value, dict): raise CanaryError("Postgres readback must be a JSON object") return value def cleanup(self) -> dict[str, Any]: remove = self._run(["rm", "-f", self.name], check=False) if self.started else None inspect = self._run(["inspect", self.name], check=False) return { "remove_attempted": self.started, "remove_returncode": remove.returncode if remove else None, "container_absent": inspect.returncode != 0, "docker_volume_mounts_observed": self.volume_mounts, "anonymous_or_named_volume_created": bool(self.volume_mounts), } 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.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": [ {key: row[key] for key in ("claim_id", "source_id", "role")} for row in planned_evidence_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, Any], readback: dict[str, Any], canonical_before: dict[str, Any], canonical_after: dict[str, Any], ) -> dict[str, bool]: 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 { "source_only_artifact_hash_persisted": ( len(source_rows) == 1 and source_row.get("artifact_sha256") == input_receipt["artifact_sha256"] ), "source_capture_identity_exact": source_rows_exact, "deterministic_row_ids_recomputed_exact": row_ids == deterministic_row_ids, "replay_identity_persisted": ( len(source_rows) == 1 and source_row.get("replay_identity_sha256") == input_receipt["replay_identity_sha256"] ), "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 ) ), "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]: receipt: dict[str, Any] = { "schema": SCHEMA, "status": "fail", "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(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, "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, TypeError, AttributeError, IndexError, normalized_stage.bound.CheckpointError, ) as exc: receipt["error"] = {"type": type(exc).__name__, "message": str(exc)} finally: 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" if receipt["status"] == "pass": receipt["strongest_claim"] = ( "One source-only artifact was parsed into exact-location candidate claims, duplicate/conflict " "assessments, and a replay-bound pending_review proposal in disposable networkless Postgres; " "representative canonical row contents remained fingerprint-identical." ) else: receipt["strongest_claim"] = ( "No source-to-review-staging capability claim is made because acceptance failed closed." ) receipt["residual_gap"] = ( "This proves the deterministic local fixture extractor and review queue only; it does not prove " "arbitrary generative extraction, approval/apply, hosted runtime, or production mutation." ) if output: output.parent.mkdir(parents=True, exist_ok=True) output.write_text(json.dumps(receipt, indent=2, sort_keys=True) + "\n", encoding="utf-8") 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, 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) 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 if __name__ == "__main__": raise SystemExit(main())