#!/usr/bin/env python3 """Run a document-ingestion-to-staged-proposal canary in disposable local Postgres.""" from __future__ import annotations import argparse import hashlib import json 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.v1" FIXTURE_SCHEMA = "livingip.workingLeoDocumentIngestionFixture.v1" DEFAULT_FIXTURE = REPO_ROOT / "fixtures" / "working-leo" / "document-ingestion-v1.json" CONTAINER_LABEL = "livingip.canary=working-leo-local-ingestion" class CanaryError(RuntimeError): pass def sha256_bytes(value: bytes) -> str: return hashlib.sha256(value).hexdigest() def canonical_sha256(value: Any) -> str: return sha256_bytes(json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode()) 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 load_fixture(path: Path) -> tuple[dict[str, Any], dict[str, Any]]: raw = path.read_bytes() 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") artifact_sha256 = sha256_bytes(raw) source_content_sha256 = sha256_bytes(source["body"].encode()) return fixture, { "path": str(path.resolve()), "artifact_sha256": artifact_sha256, "source_content_sha256": source_content_sha256, "source_identity": source["identity"], } def build_row_ids(artifact_sha256: str) -> dict[str, str]: 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"), } def build_parent_packet( fixture: dict[str, Any], input_receipt: dict[str, Any], row_ids: dict[str, str] ) -> 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']}" ) 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": parent_source_ref, "rationale": "Stage one hash-bound claim/evidence extraction from a captured local document fixture.", "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"], } ], "dedupe_conflict_assessment": { "database_search_query": claim["claim_key"].replace("_", " "), "potential_existing_claim_ids": [], "conflicts": [], }, }, } def build_schema_sql() -> str: return r"""\set ON_ERROR_STOP on create schema kb_canary; create schema kb_stage; 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}$'), body text not null, 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, 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), role text not null, body 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_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_readback_sql(row_ids: dict[str, str], 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), '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), '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_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"] 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], "planned_evidence_keys": [ {"claim_id": row["claim_id"], "source_id": row["source_id"], "role": row["role"]} for row in 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 ), } def evaluate_checks( fixture: dict[str, Any], input_receipt: dict[str, Any], row_ids: dict[str, str], readback: 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) 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"] ), "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"] ), "evidence_body_and_metadata_persisted": ( evidence.get("body") == fixture["extraction"]["evidence"]["body"] and evidence.get("metadata") == fixture["extraction"]["evidence"]["metadata"] ), "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") ), } 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, "production_mutation_attempted": False, "canonical_apply_attempted": False, } try: receipt["environment"] = postgres.start() postgres.psql(build_schema_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) receipt.update( { "stage_command_result": stage_result, "rows": readback, "row_link_fields_proven": proposal_links(readback, input_receipt), "checks": checks, "status": "pass" if all(checks.values()) else "fail", } ) except (CanaryError, OSError, ValueError, KeyError) 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"], ) ): 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." ) receipt["residual_gap"] = ( "This does not prove live generative extraction, arbitrary document coverage, canonical apply, or any " "hosted/production runtime." ) 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 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("--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) print(json.dumps(receipt, sort_keys=True, separators=(",", ":"))) return 0 if receipt["status"] == "pass" else 1 if __name__ == "__main__": raise SystemExit(main())