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..3cb9243 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,497 @@ 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.v1" +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 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") + 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()), + source_content_sha256 = canonical_sha256( + [ + { + "id": segment["id"], + "locator": segment["locator"], + "content_sha256": segment["content_sha256"], + } + for segment in segments + ] + ) + extraction_spec_sha256 = canonical_sha256(extraction) + replay_identity_sha256 = canonical_sha256( + { + "artifact_sha256": artifact_sha256, + "extractor": {"name": extractor_name, "version": extractor_version}, + "extraction_spec_sha256": extraction_spec_sha256, + } + ) + fixture = { + **scenario, + "artifact": {**artifact, "resolved_path": str(artifact_path)}, + "extractor": {"name": extractor_name, "version": extractor_version}, + "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": {"name": extractor_name, "version": extractor_version}, + "extraction_spec_sha256": extraction_spec_sha256, + "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(input_receipt["artifact_sha256"], "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_sha256": input_receipt["replay_identity_sha256"], + "segments": [ + { + "id": segment["id"], + "locator": segment["locator"], + "json_pointer": segment["json_pointer"], + "content_sha256": segment["content_sha256"], + "text_sha256": segment["text_sha256"], + } + for segment in fixture["segments"] + ], + "row_ids": row_ids, + "candidate_counts": input_receipt["counts"], + } return { "id": row_ids["parent_proposal_id"], "proposal_type": "attach_evidence", @@ -111,47 +528,55 @@ 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 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 +584,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 +600,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 +649,111 @@ 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 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 +821,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" ] @@ -386,60 +883,130 @@ def proposal_links(readback: dict[str, Any], input_receipt: dict[str, Any]) -> d "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 + "planned_edge_keys": [ + { + "from_claim": row["from_claim"], + "to_claim": row["to_claim"], + "edge_type": row["edge_type"], + } + for row in apply_payload["edges"] + ], + "planned_counts": { + "claims": len(apply_payload["claims"]), + "sources": len(source_rows), + "evidence": len(evidence_rows), + "edges": len(apply_payload["edges"]), + }, + "input_hash_in_manifest": ( + apply_payload["normalization_manifest"].get("ingestion_manifest", {}).get("artifact_sha256") + == input_receipt["artifact_sha256"] ), - "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 + "replay_identity_in_manifest": ( + apply_payload["normalization_manifest"].get("ingestion_manifest", {}).get("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 {} + source_rows = readback.get("source_captures") or [] + claim_rows = readback.get("claim_extractions") or [] + evidence_rows = readback.get("evidence_extractions") or [] + duplicate_rows = readback.get("duplicate_assessments") or [] + conflict_rows = readback.get("conflict_assessments") or [] proposal = readback.get("staged_proposal") or {} + counts = input_receipt["counts"] links = proposal_links(readback, input_receipt) + manifest = ((proposal.get("payload") or {}).get("apply_payload") or {}).get("normalization_manifest") or {} + ingestion_manifest = manifest.get("ingestion_manifest") or {} + assessment = manifest.get("dedupe_conflict_assessment") or {} + 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, + } + segment_by_id = {segment["id"]: segment for segment in fixture["segments"]} + evidence_locations_exact = all( + row.get("source_locator") == segment_by_id.get(row.get("source_segment_id"), {}).get("locator") + and row.get("source_content_sha256") + == segment_by_id.get(row.get("source_segment_id"), {}).get("content_sha256") + and row.get("body") in segment_by_id.get(row.get("source_segment_id"), {}).get("body", "") + for row in evidence_rows + ) 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_rows[0].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"] + "replay_identity_persisted": ( + len(source_rows) == 1 + and source_rows[0].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 all(row.get("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}, + "all_claim_candidates_persisted_once": ( + {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"] + ), + "all_evidence_has_exact_source_location": ( + len(evidence_rows) == counts["evidence_candidates"] and evidence_locations_exact + ), + "duplicate_judgments_persisted": ( + len(duplicate_rows) == counts["duplicate_judgments"] + and assessment.get("duplicates") == fixture["extraction"]["duplicates"] + ), + "conflicts_and_relationships_persisted": ( + len(conflict_rows) == counts["conflict_relationships"] + and assessment.get("conflicts") == fixture["extraction"]["conflicts"] + and links["planned_counts"]["edges"] == counts["conflict_relationships"] + ), + "database_candidate_counts_exact": readback.get("counts") == expected_db_counts, "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"], + "proposal_has_replayable_ingestion_manifest": ( + links["input_hash_in_manifest"] + and links["replay_identity_in_manifest"] + and ingestion_manifest.get("segments") + == [ + { + "id": segment["id"], + "locator": segment["locator"], + "json_pointer": segment["json_pointer"], + "content_sha256": segment["content_sha256"], + "text_sha256": segment["text_sha256"], + } + for segment in fixture["segments"] + ] + ), "no_review_or_apply_metadata": all( proposal.get(key) is None for key in ("reviewed_by_handle", "reviewed_at", "applied_by_handle", "applied_at") ), + "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"]) + row_ids = build_row_ids(input_receipt, fixture) parent = build_parent_packet(fixture, input_receipt, row_ids) child = normalized_stage.prepare_normalized_child(parent) container_name = f"working-leo-ingest-{uuid.uuid4().hex[:12]}" @@ -447,7 +1014,7 @@ def run_canary(fixture_path: Path, output: Path | None = None) -> dict[str, Any] receipt: dict[str, Any] = { "schema": SCHEMA, "status": "fail", - "required_tier": "isolated_realistic_ingestion_to_proposal", + "required_tier": "T2_runtime_to_review_queue_staging", "input": input_receipt, "row_ids": row_ids, "production_mutation_attempted": False, @@ -456,44 +1023,49 @@ def run_canary(fixture_path: Path, output: Path | None = None) -> dict[str, Any] try: 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, + "candidate_counts": input_receipt["counts"], "row_link_fields_proven": proposal_links(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, 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"], - ) - ): + 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." + "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." ) 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 +1073,45 @@ def run_canary(fixture_path: Path, output: Path | None = None) -> dict[str, Any] return receipt +def run_suite(fixture_paths: list[Path], output: Path | None = None) -> dict[str, Any]: + receipts = [run_canary(path.resolve()) for path in fixture_paths] + suite = { + "schema": SUITE_SCHEMA, + "status": "pass" if receipts and all(item["status"] == "pass" for item in receipts) else "fail", + "required_tier": "T2_runtime_to_review_queue_staging", + "fixture_count": len(receipts), + "document_fixture_passed": any( + item["status"] == "pass" and item["input"]["artifact_format"] == "plain_text" for item in receipts + ), + "social_conversation_fixture_passed": any( + item["status"] == "pass" and item["input"]["artifact_format"] == "telegram_jsonl" for item in receipts + ), + "canonical_fingerprint_invariant_all": bool(receipts) + and all(item.get("canonical", {}).get("unchanged") is True for item in receipts), + "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..cfae09e 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,247 @@ 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 _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": {"extractor": input_receipt["extractor"]}, + } + ) + 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_content_sha256": segment["content_sha256"], + "body": evidence["excerpt"], + } + ) + proposal = { + **child, + "reviewed_by_handle": None, + "reviewed_at": None, + "applied_by_handle": None, + "applied_at": None, + } + counts = input_receipt["counts"] + return { + "source_captures": [ + { + "artifact_sha256": input_receipt["artifact_sha256"], + "replay_identity_sha256": input_receipt["replay_identity_sha256"], + } + ], + "claim_extractions": claim_rows, + "evidence_extractions": evidence_rows, + "duplicate_assessments": [{} for _item in fixture["extraction"]["duplicates"]], + "conflict_assessments": [{} for _item in fixture["extraction"]["conflicts"]], + "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"] + + +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 +264,50 @@ 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 ) 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)