feat: stage source-linked ingestion proposals without canonical writes

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": []
+  }
+}

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.

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}

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."
+      }
+    ]
+  }
+}

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]:
+def build_source_fragment_index(text: str) -> tuple[dict[str, str], dict[str, dict[str, Any]]]:
-    """Label non-empty source lines so the model selects text instead of copying it."""
+    """Return selectable lines plus exact line/character provenance."""
     fragments: dict[str, str] = {}
-    for line in text.splitlines():
+    locations: dict[str, dict[str, Any]] = {}
-        if line.strip():
+    offset = 0
-            fragments[f"S{len(fragments) + 1:05d}"] = line
+    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(
+def build_source_fragments(text: str) -> dict[str, str]:
-    fragments: dict[str, str], context: dict[str, Any], repair_error: str | None = None
+    """Label non-empty source lines so the model selects text instead of copying it."""
-) -> str:
+    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(
+            raise PreparationError(f"model extraction claims[{index}] keys must equal {sorted(MODEL_CLAIM_KEYS)}")
-                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(
+                    "quote": evidence_quote,
-                        evidence_ref,
-                        fragments,
-                        f"claims[{claim_index}].evidence[{evidence_index}].quote_ref",
-                    ),
                     "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),

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)

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"] == [
+    selected = receipt["extraction"]["selected_source_references"]
-        {"kind": "claim", "reference": "S00003"},
+    assert [(row["kind"], row["reference"]) for row in selected] == [
-        {"kind": "evidence", "reference": "S00003"},
+        ("claim", "S00003"),
-        {"kind": "duplicate", "reference": "S00002"},
+        ("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(

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]:
+def _loaded(path: Path) -> tuple[dict, dict, dict, dict, dict]:
-    fixture, input_receipt = canary.load_fixture(canary.DEFAULT_FIXTURE)
+    fixture, input_receipt = canary.load_fixture(path)
-    row_ids = canary.build_row_ids(input_receipt["artifact_sha256"])
+    row_ids = canary.build_row_ids(input_receipt, fixture)
-    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()
     parent = canary.build_parent_packet(fixture, input_receipt, row_ids)
     child = canary.normalized_stage.prepare_normalized_child(parent)
-    payload = child["payload"]["apply_payload"]
+    return fixture, input_receipt, row_ids, parent, child
-
-    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
 
 
-def test_capture_and_stage_sql_never_mutate_canonical_public_tables() -> None:
+def _copy_scenario(tmp_path: Path, scenario_path: Path) -> Path:
-    fixture, input_receipt, row_ids = _loaded()
+    scenario = json.loads(scenario_path.read_text(encoding="utf-8"))
-    child = canary.normalized_stage.prepare_normalized_child(
+    artifact_source = scenario_path.parent / scenario["artifact"]["path"]
-        canary.build_parent_packet(fixture, input_receipt, row_ids)
+    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:
+def test_canonical_snapshot_covers_all_nonempty_canonical_tables() -> None:
-    fixture, input_receipt, row_ids = _loaded()
+    schema_sql = canary.build_schema_sql().lower()
-    child = canary.normalized_stage.prepare_normalized_child(
+    snapshot_sql = canary.build_canonical_snapshot_sql().lower()
-        canary.build_parent_packet(fixture, input_receipt, row_ids)
+
+    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": {
+def test_evaluation_fails_when_exact_evidence_locator_is_changed() -> None:
-            "id": row_ids["source_capture_id"],
+    fixture, input_receipt, row_ids, _parent, child = _loaded(canary.DEFAULT_TELEGRAM_FIXTURE)
-            "artifact_sha256": input_receipt["artifact_sha256"],
+    readback = _synthetic_readback(fixture, input_receipt, row_ids, child)
-            "content_sha256": input_receipt["source_content_sha256"],
+    canonical = {
-            "source_identity": input_receipt["source_identity"],
+        "claims": [{"id": "1"}],
-        },
+        "sources": [{"id": "2"}],
-        "claim_extraction": {
+        "claim_evidence": [{"claim_id": "1"}],
-            "id": row_ids["claim_extraction_id"],
+        "claim_edges": [{"from_claim": "1"}],
-            "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,
-        },
     }
 
-    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
     )

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)