Merge pull request #86 from living-ip/codex/working-leo-compose-apply-20260712

Prove Leo KB composition and approved apply lifecycle

fixtures/working-leo/document-ingestion-v1.json

@@ -0,0 +1,39 @@
+{
+  "schema": "livingip.workingLeoDocumentIngestionFixture.v1",
+  "source": {
+    "identity": "document:working-leo-review-gated-composition-v1",
+    "source_key": "operating_document",
+    "source_type": "article",
+    "title": "Working Leo review-gated composition note",
+    "locator": "fixture://working-leo/document-ingestion-v1",
+    "body": "Working Leo composition note. A newly captured document may produce claim and evidence candidates, but those candidates are not canonical knowledge.\n\nA 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.\n\nFor a staging-only rehearsal, pending_review is the terminal state. No public knowledge-base row should be written.",
+    "metadata": {
+      "author": "Working Leo canary fixture",
+      "captured_at": "2026-07-12T00:00:00Z",
+      "document_kind": "operating_note",
+      "language": "en"
+    }
+  },
+  "extraction": {
+    "claim": {
+      "claim_key": "staged_proposal_remains_noncanonical",
+      "type": "structural",
+      "confidence": 0.99,
+      "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": {
+        "extractor": "deterministic_fixture_replay",
+        "extraction_version": 1,
+        "needs_research": false
+      }
+    },
+    "evidence": {
+      "role": "grounds",
+      "body": "A staged proposal remains non-canonical until an operator reviews it and a separate guarded apply succeeds.",
+      "metadata": {
+        "evidence_kind": "exact_quote",
+        "section": "paragraph_2"
+      }
+    }
+  }
+}

scripts/apply_proposal.py

@@ -85,6 +85,7 @@ from __future__ import annotations
 
 import argparse
 import json
+import shutil
 import subprocess
 import sys
 import uuid
@@ -857,8 +858,9 @@ def load_password(secrets_file: str) -> str:
 
 
 def run_psql(args: argparse.Namespace, sql: str, password: str) -> str:
+    docker_binary = shutil.which("docker") or "docker"
     command = [
-        "docker", "exec", "-e", "PGPASSWORD", "-i", args.container,
+        docker_binary, "exec", "-e", "PGPASSWORD", "-i", args.container,
         "psql", "-U", args.role, "-h", args.host, "-d", args.db,
         "-v", "ON_ERROR_STOP=1", "-At", "-q",
     ]

scripts/run_approve_claim_clone_canary.py

@@ -15,6 +15,7 @@ import argparse
 import hashlib
 import json
 import re
+import shutil
 import subprocess
 import sys
 import uuid
@@ -135,7 +136,7 @@ def _role_psql(
     password: str,
 ) -> subprocess.CompletedProcess[str]:
     command = [
-        "docker",
+        shutil.which("docker") or "docker",
         "exec",
         "-e",
         "PGPASSWORD",
@@ -213,8 +214,7 @@ def _schema_copy(container: str, source_db: str, clone_db: str) -> None:
 
 
 def _service_state(service: str) -> dict[str, Any]:
-    result = _run(
+    command = [
-        [
         "systemctl",
         "show",
         service,
@@ -229,10 +229,19 @@ def _service_state(service: str) -> dict[str, Any]:
         "NRestarts",
         "-p",
         "ExecMainStartTimestamp",
-        ],
+    ]
-        check=False,
+    try:
-    )
+        result = _run(command, check=False)
-    state: dict[str, Any] = {"returncode": result.returncode}
+    except FileNotFoundError:
+        return {
+            "available": False,
+            "returncode": 127,
+            "reason": "systemctl_unavailable",
+        }
+    state: dict[str, Any] = {
+        "available": result.returncode == 0,
+        "returncode": result.returncode,
+    }
     for line in result.stdout.splitlines():
         key, separator, value = line.partition("=")
         if separator:
@@ -826,6 +835,108 @@ def _table_deltas(before: dict[str, int], after: dict[str, int]) -> dict[str, in
     return {key: after[key] - before[key] for key in sorted(before)}
 
 
+def _canonical_row_keys(rows: dict[str, list[dict[str, Any]]]) -> dict[str, Any]:
+    """Return stable primary/composite keys for every canonical row in a bundle."""
+    return {
+        "public.claims": sorted(row["id"] for row in rows.get("claims", [])),
+        "public.sources": sorted(row["id"] for row in rows.get("sources", [])),
+        "public.claim_evidence": sorted(
+            [
+                {
+                    "claim_id": row["claim_id"],
+                    "source_id": row["source_id"],
+                    "role": row["role"],
+                }
+                for row in rows.get("claim_evidence", [])
+            ],
+            key=lambda row: (row["claim_id"], row["source_id"], row["role"]),
+        ),
+        "public.claim_edges": sorted(
+            [
+                {
+                    "from_claim": row["from_claim"],
+                    "to_claim": row["to_claim"],
+                    "edge_type": row["edge_type"],
+                }
+                for row in rows.get("claim_edges", [])
+            ],
+            key=lambda row: (row["from_claim"], row["to_claim"], row["edge_type"]),
+        ),
+        "public.reasoning_tools": sorted(
+            row["id"] for row in rows.get("reasoning_tools", [])
+        ),
+    }
+
+
+def _strict_receipt(
+    result: dict[str, Any],
+    expected_rows: dict[str, list[dict[str, Any]]],
+    expected_table_deltas: dict[str, int],
+) -> dict[str, Any]:
+    """Build the compact pass/fail surface from actual database readbacks."""
+    applied = result.get("apply_transition", {}).get("applied_readback") or {}
+    actual_rows = result.get("row_readback") or {}
+    expected_keys = _canonical_row_keys(expected_rows)
+    actual_keys = _canonical_row_keys(actual_rows)
+    checks = {
+        "proposal_id_exact": applied.get("id")
+        == result.get("fixture_ids", {}).get("proposal_id"),
+        "proposal_status_applied": applied.get("status") == "applied",
+        "proposal_applied_at_present": bool(applied.get("applied_at")),
+        "canonical_rows_exact": actual_rows == expected_rows,
+        "canonical_row_keys_exact": actual_keys == expected_keys,
+        "before_after_counts_present": (
+            isinstance(result.get("clone_counts_before_apply"), dict)
+            and isinstance(result.get("clone_counts_after_apply"), dict)
+        ),
+        "table_deltas_exact": result.get("clone_apply_table_deltas")
+        == expected_table_deltas,
+        "conflict_transaction_rolled_back": result.get("checks", {}).get(
+            "conflict_transaction_rolled_back_exactly"
+        )
+        is True,
+        "clone_database_removed": (
+            result.get("clone_dropped") is True
+            and result.get("cleanup_readback", {}).get(
+                "leftover_clone_database_count"
+            )
+            == 0
+        ),
+        "source_database_unchanged": result.get("source_database_mutated") is False,
+    }
+    return {
+        "schema": "working-leo.approved-change-strict-receipt.v1",
+        "pass": result.get("status") == "pass" and all(checks.values()),
+        "checks": checks,
+        "proposal": {
+            "id": applied.get("id"),
+            "status": applied.get("status"),
+            "applied_at": applied.get("applied_at"),
+            "applied_by_handle": applied.get("applied_by_handle"),
+        },
+        "canonical_row_keys": {
+            "expected": expected_keys,
+            "actual": actual_keys,
+        },
+        "row_counts": {
+            "before_apply": result.get("clone_counts_before_apply"),
+            "after_apply": result.get("clone_counts_after_apply"),
+            "expected_deltas": expected_table_deltas,
+            "actual_deltas": result.get("clone_apply_table_deltas"),
+        },
+        "rollback_cleanup": {
+            "conflict_transaction_rolled_back": checks[
+                "conflict_transaction_rolled_back"
+            ],
+            "clone_database_removed": checks["clone_database_removed"],
+            "leftover_clone_database_count": result.get("cleanup_readback", {}).get(
+                "leftover_clone_database_count"
+            ),
+            "source_database_unchanged": checks["source_database_unchanged"],
+        },
+    }
+
+
 def _exact_bundle_readback(
     container: str, database: str, payload: dict[str, Any]
 ) -> dict[str, list[dict[str, Any]]]:
@@ -1905,6 +2016,18 @@ drop database if exists {clone_db};
                 f"{type(exc).__name__}: {exc}"
             )
 
+    gateway_process_observable = (
+        result.get("service_before", {}).get("available") is True
+        and result.get("service_after", {}).get("available") is True
+    )
+    gateway_process_unchanged = (
+        result.get("service_before", {}).get("MainPID")
+        == result.get("service_after", {}).get("MainPID")
+        and result.get("service_before", {}).get("NRestarts")
+        == result.get("service_after", {}).get("NRestarts")
+        if gateway_process_observable
+        else None
+    )
     result["checks"] = {
         "security_definer_dependencies_ready": _security_dependencies_ready(
             result.get("security_definer_readback") or {}
@@ -2058,12 +2181,7 @@ drop database if exists {clone_db};
         "source_files_unchanged_during_run": (
             result.get("source_binding", {}).get("unchanged_during_run") is True
         ),
-        "gateway_process_unchanged": (
+        "gateway_process_unchanged": gateway_process_unchanged,
-            result.get("service_before", {}).get("MainPID")
-            == result.get("service_after", {}).get("MainPID")
-            and result.get("service_before", {}).get("NRestarts")
-            == result.get("service_after", {}).get("NRestarts")
-        ),
         "clone_cleanup_complete": (
             result.get("clone_dropped") is True
             and result.get("leftover_clone_count") == 0
@@ -2072,9 +2190,11 @@ drop database if exists {clone_db};
     result["source_counts_observed_unchanged"] = (
         result.get("source_counts_before") == result.get("source_counts_after")
     )
+    result["gateway_process_observable"] = gateway_process_observable
     result["status"] = (
         "pass"
-        if all(result["checks"].values()) and "error" not in result
+        if all(value is True for value in result["checks"].values() if value is not None)
+        and "error" not in result
         else "fail"
     )
     result["source_database_mutated"] = not all(
@@ -2085,16 +2205,24 @@ drop database if exists {clone_db};
             "source_table_deltas_exactly_zero",
         )
     )
-    result["service_restarted"] = not result["checks"][
+    result["service_restarted"] = (
-        "gateway_process_unchanged"
+        None
-    ]
+        if gateway_process_unchanged is None
+        else not gateway_process_unchanged
+    )
     result["post_run_cleanup"] = {
         "leftover_clone_database_count": result.get("leftover_clone_count"),
+        "gateway_process_observable": gateway_process_observable,
         "gateway_active_state": result.get("service_after", {}).get("ActiveState"),
         "gateway_sub_state": result.get("service_after", {}).get("SubState"),
         "gateway_main_pid": result.get("service_after", {}).get("MainPID"),
         "gateway_restart_count": result.get("service_after", {}).get("NRestarts"),
     }
+    result["strict_receipt"] = _strict_receipt(
+        result, expected_rows, expected_table_deltas
+    )
+    if not result["strict_receipt"]["pass"]:
+        result["status"] = "fail"
     return result
 
 

scripts/run_approve_claim_isolated_container_canary.sh

@@ -6,6 +6,7 @@ source_container="${SOURCE_CONTAINER:-teleo-pg}"
 source_db="${SOURCE_DB:-teleo}"
 service="${LEO_SERVICE:-leoclean-gateway.service}"
 image="${POSTGRES_CANARY_IMAGE:-postgres:16}"
+source_tier="${SOURCE_TIER:-live-readonly}"
 normalization_json=""
 output="${repo_root}/docs/reports/leo-working-state-20260709/approve-claim-clone-canary-current.json"
 
@@ -31,6 +32,10 @@ while [[ $# -gt 0 ]]; do
       service="$2"
       shift 2
       ;;
+    --source-tier)
+      source_tier="$2"
+      shift 2
+      ;;
     *)
       echo "unknown argument: $1" >&2
       exit 2
@@ -38,6 +43,22 @@ while [[ $# -gt 0 ]]; do
   esac
 done
 
+if [[ "$source_tier" != "isolated-local" && "$source_tier" != "live-readonly" ]]; then
+  echo "--source-tier must be isolated-local or live-readonly" >&2
+  exit 2
+fi
+
+source_network_mode="$(docker inspect -f '{{.HostConfig.NetworkMode}}' "$source_container")"
+source_canary_label="$(
+  docker inspect -f '{{index .Config.Labels "livingip.canary"}}' "$source_container"
+)"
+if [[ "$source_tier" == "isolated-local" ]]; then
+  if [[ "$source_network_mode" != "none" || "$source_canary_label" != "working-leo-approved-source" ]]; then
+    echo "isolated-local source must be networkless and labeled livingip.canary=working-leo-approved-source" >&2
+    exit 2
+  fi
+fi
+
 container="working-leo-approve-claim-$RANDOM-$$"
 workdir="/tmp/${container}"
 apply_password="isolated-apply-${RANDOM}-${RANDOM}"
@@ -64,8 +85,17 @@ source_counts() {
 }
 
 service_state() {
-  systemctl show "$service" --no-pager \
+  if ! command -v systemctl >/dev/null 2>&1; then
-    -p ActiveState -p SubState -p MainPID -p NRestarts -p ExecMainStartTimestamp
+    printf 'Available=false\nReason=systemctl_unavailable\n'
+    return 0
+  fi
+  local output
+  if output="$(systemctl show "$service" --no-pager \
+    -p ActiveState -p SubState -p MainPID -p NRestarts -p ExecMainStartTimestamp 2>&1)"; then
+    printf 'Available=true\n%s\n' "$output"
+  else
+    printf 'Available=false\nReason=service_readback_failed\n'
+  fi
 }
 
 mkdir -p "$workdir" "$(dirname "$output")"
@@ -73,17 +103,30 @@ source_counts_before="$(source_counts)"
 service_before="$(service_state)"
 
 docker run -d --name "$container" \
+  --network none \
+  --tmpfs /var/lib/postgresql/data:rw,nosuid,size=512m \
   -e POSTGRES_PASSWORD="isolated-postgres-${RANDOM}-${RANDOM}" \
   -e POSTGRES_DB=teleo \
   "$image" >/dev/null
 
-for _ in $(seq 1 40); do
+ready_streak=0
+for _ in $(seq 1 80); do
   if docker exec "$container" pg_isready -U postgres -d teleo >/dev/null 2>&1; then
+    ready_streak=$((ready_streak + 1))
+    if [[ "$ready_streak" -ge 2 ]]; then
       break
     fi
+  else
+    ready_streak=0
+  fi
   sleep 0.25
 done
-docker exec "$container" pg_isready -U postgres -d teleo >/dev/null
+if [[ "$ready_streak" -lt 2 ]]; then
+  echo "disposable Postgres did not become stably ready" >&2
+  exit 1
+fi
+container_network_mode="$(docker inspect -f '{{.HostConfig.NetworkMode}}' "$container")"
+container_mounts_json="$(docker inspect -f '{{json .Mounts}}' "$container")"
 
 docker exec "$source_container" pg_dump -U postgres -d "$source_db" \
   --schema-only --no-owner --no-privileges >"$workdir/schema.sql"
@@ -173,6 +216,11 @@ CONTAINER_NAME_READBACK="$container_name_readback" \
 CONTAINER_NAME_READBACK_RC="$container_name_readback_rc" \
 OUTER_WORKDIR_LEXISTS="$outer_workdir_lexists" \
 CANARY_RC="$canary_rc" \
+SOURCE_TIER="$source_tier" \
+SOURCE_NETWORK_MODE="$source_network_mode" \
+SOURCE_CANARY_LABEL="$source_canary_label" \
+CANARY_NETWORK_MODE="$container_network_mode" \
+CANARY_MOUNTS_JSON="$container_mounts_json" \
 python3 - "$output" <<'PY'
 import hashlib
 import json
@@ -254,10 +302,41 @@ after_counts = json.loads(os.environ["SOURCE_COUNTS_AFTER"])
 live_table_deltas = table_deltas(before_counts, after_counts)
 before_service = service(os.environ["SERVICE_BEFORE"])
 after_service = service(os.environ["SERVICE_AFTER"])
-service_stable = all(
+source_tier = os.environ["SOURCE_TIER"]
+source_network_mode = os.environ["SOURCE_NETWORK_MODE"]
+source_canary_label = os.environ["SOURCE_CANARY_LABEL"]
+container_network_mode = os.environ["CANARY_NETWORK_MODE"]
+container_mounts = json.loads(os.environ["CANARY_MOUNTS_JSON"])
+container_volume_mounts = [
+    mount for mount in container_mounts if mount.get("Type") == "volume"
+]
+container_isolated = (
+    container_network_mode == "none" and not container_volume_mounts
+)
+source_tier_enforced = (
+    source_tier != "isolated-local"
+    or (
+        source_network_mode == "none"
+        and source_canary_label == "working-leo-approved-source"
+    )
+)
+service_observable = (
+    before_service.get("Available") == "true"
+    and after_service.get("Available") == "true"
+)
+service_stable = (
+    all(
         before_service.get(key) == after_service.get(key)
         for key in ("ActiveState", "SubState", "MainPID", "NRestarts")
     )
+    if service_observable
+    else None
+)
+service_gate_passes = (
+    service_stable is True
+    if source_tier == "live-readonly"
+    else service_stable is not False
+)
 container_absent = (
     int(os.environ["CONTAINER_NAME_READBACK_RC"]) == 0
     and not os.environ["CONTAINER_NAME_READBACK"].strip()
@@ -277,12 +356,19 @@ inner_runtime_pass = data.get("status") == "pass" and int(os.environ["CANARY_RC"
 outer_ok = (
     before_counts == after_counts
     and all(delta == 0 for delta in live_table_deltas.values())
-    and service_stable
+    and service_gate_passes
     and container_absent
     and workdir_absent
     and source_binding_ok
     and stale_ephemeral_paths_absent
+    and container_isolated
+    and source_tier_enforced
 )
+if source_tier == "isolated-local":
+    data["required_tier"] = "T2_runtime"
+    data["required_tiers"] = ["T2_runtime"]
+    data["runtime_scope"] = "isolated_postgres_container_from_readonly_local_fixture"
+else:
     data["required_tier"] = "T3_live_readonly"
     data["required_tiers"] = ["T2_runtime", "T3_live_readonly"]
     data["runtime_scope"] = "isolated_postgres_container_from_readonly_live_schema"
@@ -295,7 +381,22 @@ data["outer_isolation"] = {
     "source_counts_unchanged": before_counts == after_counts,
     "service_before": before_service,
     "service_after": after_service,
+    "service_required": source_tier == "live-readonly",
+    "service_observable": service_observable,
     "service_stable": service_stable,
+    "service_gate_passes": service_gate_passes,
+    "source_boundary": {
+        "source_tier": source_tier,
+        "network_mode": source_network_mode,
+        "canary_label": source_canary_label,
+        "tier_enforced": source_tier_enforced,
+    },
+    "container_isolation": {
+        "network_mode": container_network_mode,
+        "volume_mounts": container_volume_mounts,
+        "network_disabled": container_network_mode == "none",
+        "no_docker_volumes": not container_volume_mounts,
+    },
     "cleanup_readback": {
         "container_remove_returncode": int(os.environ["OUTER_CONTAINER_REMOVE_RC"]),
         "container_name_query_returncode": int(
@@ -312,14 +413,24 @@ data["checks"]["outer_live_counts_exactly_unchanged"] = before_counts == after_c
 data["checks"]["outer_live_table_deltas_exactly_zero"] = all(
     delta == 0 for delta in live_table_deltas.values()
 )
-data["checks"]["outer_service_stable"] = service_stable
+data["checks"]["outer_service_observed_if_required"] = (
+    source_tier != "live-readonly" or service_observable
+)
+data["checks"]["outer_service_stable_if_observed"] = service_gate_passes
 data["checks"]["outer_container_absent_independent_readback"] = container_absent
 data["checks"]["outer_workdir_absent_independent_readback"] = workdir_absent
 data["checks"]["source_binding_independently_verified"] = source_binding_ok
 data["checks"]["stale_ephemeral_paths_absent"] = stale_ephemeral_paths_absent
+data["checks"]["outer_source_tier_enforced"] = source_tier_enforced
+data["checks"]["outer_container_network_disabled"] = (
+    container_network_mode == "none"
+)
+data["checks"]["outer_container_has_no_docker_volumes"] = not container_volume_mounts
 data["checks"]["outer_isolation_complete"] = outer_ok
 if outer_ok and inner_runtime_pass:
-    data["current_tier"] = "T3_live_readonly"
+    data["current_tier"] = (
+        "T2_runtime" if source_tier == "isolated-local" else "T3_live_readonly"
+    )
 else:
     data["status"] = "fail"
     data["current_tier"] = "T2_runtime" if inner_runtime_pass else "T1_model"

scripts/run_leo_local_ingestion_proposal_canary.py

@@ -0,0 +1,519 @@
+#!/usr/bin/env python3
+"""Run a document-ingestion-to-staged-proposal canary in disposable local Postgres."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import shutil
+import subprocess
+import sys
+import time
+import uuid
+from pathlib import Path
+from typing import Any
+
+HERE = Path(__file__).resolve().parent
+REPO_ROOT = HERE.parent
+sys.path.insert(0, str(HERE))
+
+import apply_proposal as ap  # noqa: E402
+import stage_normalized_proposal as normalized_stage  # noqa: E402
+
+SCHEMA = "livingip.workingLeoLocalIngestionProposalCanary.v1"
+FIXTURE_SCHEMA = "livingip.workingLeoDocumentIngestionFixture.v1"
+DEFAULT_FIXTURE = REPO_ROOT / "fixtures" / "working-leo" / "document-ingestion-v1.json"
+CONTAINER_LABEL = "livingip.canary=working-leo-local-ingestion"
+
+
+class CanaryError(RuntimeError):
+    pass
+
+
+def sha256_bytes(value: bytes) -> str:
+    return hashlib.sha256(value).hexdigest()
+
+
+def canonical_sha256(value: Any) -> str:
+    return sha256_bytes(json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode())
+
+
+def stable_uuid(artifact_sha256: str, label: str) -> str:
+    return str(uuid.uuid5(uuid.NAMESPACE_URL, f"livingip:working-leo-ingestion:{artifact_sha256}:{label}"))
+
+
+def load_fixture(path: Path) -> tuple[dict[str, Any], dict[str, Any]]:
+    raw = path.read_bytes()
+    try:
+        fixture = json.loads(raw)
+    except json.JSONDecodeError as exc:
+        raise CanaryError(f"fixture is not valid JSON: {exc}") from exc
+    if not isinstance(fixture, dict) or fixture.get("schema") != FIXTURE_SCHEMA:
+        raise CanaryError(f"fixture schema must be {FIXTURE_SCHEMA}")
+    source = fixture.get("source")
+    extraction = fixture.get("extraction")
+    claim = extraction.get("claim") if isinstance(extraction, dict) else None
+    evidence = extraction.get("evidence") if isinstance(extraction, dict) else None
+    if not all(isinstance(item, dict) for item in (source, claim, evidence)):
+        raise CanaryError("fixture requires source, extraction.claim, and extraction.evidence objects")
+    required_source = ("identity", "source_key", "source_type", "title", "locator", "body", "metadata")
+    required_claim = ("claim_key", "type", "confidence", "text", "body", "metadata")
+    required_evidence = ("role", "body", "metadata")
+    for label, value, required in (
+        ("source", source, required_source),
+        ("claim", claim, required_claim),
+        ("evidence", evidence, required_evidence),
+    ):
+        missing = [key for key in required if value.get(key) in (None, "")]
+        if missing:
+            raise CanaryError(f"fixture {label} is missing: {', '.join(missing)}")
+        if not isinstance(value.get("metadata"), dict):
+            raise CanaryError(f"fixture {label}.metadata must be an object")
+    if claim["body"] not in source["body"]:
+        raise CanaryError("extracted claim body must be an exact substring of source.body")
+    if evidence["body"] not in source["body"]:
+        raise CanaryError("extracted evidence body must be an exact substring of source.body")
+    artifact_sha256 = sha256_bytes(raw)
+    source_content_sha256 = sha256_bytes(source["body"].encode())
+    return fixture, {
+        "path": str(path.resolve()),
+        "artifact_sha256": artifact_sha256,
+        "source_content_sha256": source_content_sha256,
+        "source_identity": source["identity"],
+    }
+
+
+def build_row_ids(artifact_sha256: str) -> dict[str, str]:
+    return {
+        "source_capture_id": stable_uuid(artifact_sha256, "source-capture"),
+        "claim_extraction_id": stable_uuid(artifact_sha256, "claim-extraction"),
+        "evidence_extraction_id": stable_uuid(artifact_sha256, "evidence-extraction"),
+        "parent_proposal_id": stable_uuid(artifact_sha256, "rich-parent-proposal"),
+    }
+
+
+def build_parent_packet(
+    fixture: dict[str, Any], input_receipt: dict[str, Any], row_ids: dict[str, str]
+) -> dict[str, Any]:
+    source = fixture["source"]
+    claim = fixture["extraction"]["claim"]
+    evidence = fixture["extraction"]["evidence"]
+    parent_source_ref = (
+        f"local-ingestion:{input_receipt['artifact_sha256']}:"
+        f"source={row_ids['source_capture_id']}:claim={row_ids['claim_extraction_id']}:"
+        f"evidence={row_ids['evidence_extraction_id']}"
+    )
+    return {
+        "id": row_ids["parent_proposal_id"],
+        "proposal_type": "attach_evidence",
+        "status": "pending_review",
+        "proposed_by_handle": "leo",
+        "proposed_by_agent_id": None,
+        "channel": "local_ingestion_canary",
+        "source_ref": parent_source_ref,
+        "rationale": "Stage one hash-bound claim/evidence extraction from a captured local document fixture.",
+        "payload": {
+            "claim_candidates": [
+                {
+                    "claim_key": claim["claim_key"],
+                    "type": claim["type"],
+                    "confidence": claim["confidence"],
+                    "text": claim["text"],
+                    "body": claim["body"],
+                    "evidence_tier": "isolated_fixture",
+                    "needs_research": claim["metadata"].get("needs_research", False),
+                    "evidence": [{"source_key": source["source_key"], "role": evidence["role"]}],
+                    "edges": [],
+                }
+            ],
+            "source_candidates": [
+                {
+                    "source_key": source["source_key"],
+                    "source_type": source["source_type"],
+                    "title": source["title"],
+                    "storage_path": input_receipt["path"],
+                    "url": None,
+                    "source_quality": "local realistic ingestion fixture",
+                    "key_excerpt": evidence["body"],
+                    "content_sha256": input_receipt["source_content_sha256"],
+                }
+            ],
+            "dedupe_conflict_assessment": {
+                "database_search_query": claim["claim_key"].replace("_", " "),
+                "potential_existing_claim_ids": [],
+                "conflicts": [],
+            },
+        },
+    }
+
+
+def build_schema_sql() -> str:
+    return r"""\set ON_ERROR_STOP on
+create schema kb_canary;
+create schema kb_stage;
+create table kb_canary.source_captures (
+  id uuid primary key,
+  source_identity text not null unique,
+  source_type text not null,
+  title text not null,
+  locator text not null,
+  artifact_path text not null,
+  artifact_sha256 text not null check (artifact_sha256 ~ '^[0-9a-f]{64}$'),
+  content_sha256 text not null check (content_sha256 ~ '^[0-9a-f]{64}$'),
+  body text not null,
+  metadata jsonb not null
+);
+create table kb_canary.claim_extractions (
+  id uuid primary key,
+  source_capture_id uuid not null references kb_canary.source_captures(id),
+  claim_key text not null,
+  body text not null,
+  metadata jsonb not null
+);
+create table kb_canary.evidence_extractions (
+  id uuid primary key,
+  claim_extraction_id uuid not null references kb_canary.claim_extractions(id),
+  source_capture_id uuid not null references kb_canary.source_captures(id),
+  role text not null,
+  body text not null,
+  metadata jsonb not null
+);
+create table kb_stage.kb_proposals (
+  id uuid primary key,
+  proposal_type text not null,
+  status text not null,
+  proposed_by_handle text,
+  proposed_by_agent_id uuid,
+  channel text not null,
+  source_ref text not null,
+  rationale text not null,
+  payload jsonb not null,
+  reviewed_by_handle text,
+  reviewed_by_agent_id uuid,
+  reviewed_at timestamptz,
+  review_note text,
+  applied_by_handle text,
+  applied_by_agent_id uuid,
+  applied_at timestamptz
+);
+"""
+
+
+def build_capture_sql(fixture: dict[str, Any], input_receipt: dict[str, Any], row_ids: dict[str, str]) -> str:
+    source = fixture["source"]
+    claim = fixture["extraction"]["claim"]
+    evidence = fixture["extraction"]["evidence"]
+    q = ap.sql_literal
+    claim_metadata = {
+        **claim["metadata"],
+        "claim_type": claim["type"],
+        "confidence": claim["confidence"],
+        "text": claim["text"],
+    }
+    return rf"""\set ON_ERROR_STOP on
+begin;
+insert into kb_canary.source_captures
+  (id, source_identity, source_type, title, locator, artifact_path,
+   artifact_sha256, content_sha256, body, metadata)
+values
+  ({q(row_ids["source_capture_id"])}::uuid, {q(source["identity"])}, {q(source["source_type"])},
+   {q(source["title"])}, {q(source["locator"])}, {q(input_receipt["path"])},
+   {q(input_receipt["artifact_sha256"])}, {q(input_receipt["source_content_sha256"])},
+   {q(source["body"])}, {q(json.dumps(source["metadata"], sort_keys=True))}::jsonb);
+insert into kb_canary.claim_extractions
+  (id, source_capture_id, claim_key, body, metadata)
+values
+  ({q(row_ids["claim_extraction_id"])}::uuid, {q(row_ids["source_capture_id"])}::uuid,
+   {q(claim["claim_key"])}, {q(claim["body"])}, {q(json.dumps(claim_metadata, sort_keys=True))}::jsonb);
+insert into kb_canary.evidence_extractions
+  (id, claim_extraction_id, source_capture_id, role, body, metadata)
+values
+  ({q(row_ids["evidence_extraction_id"])}::uuid, {q(row_ids["claim_extraction_id"])}::uuid,
+   {q(row_ids["source_capture_id"])}::uuid, {q(evidence["role"])}, {q(evidence["body"])},
+   {q(json.dumps(evidence["metadata"], sort_keys=True))}::jsonb);
+commit;
+"""
+
+
+def build_readback_sql(row_ids: dict[str, str], proposal_id: str) -> str:
+    q = ap.sql_literal
+    return rf"""\set ON_ERROR_STOP on
+begin transaction read only;
+select jsonb_build_object(
+  'source_capture', (select to_jsonb(s) from kb_canary.source_captures s
+    where id = {q(row_ids["source_capture_id"])}::uuid),
+  'claim_extraction', (select to_jsonb(c) from kb_canary.claim_extractions c
+    where id = {q(row_ids["claim_extraction_id"])}::uuid),
+  'evidence_extraction', (select to_jsonb(e) from kb_canary.evidence_extractions e
+    where id = {q(row_ids["evidence_extraction_id"])}::uuid),
+  'staged_proposal', (select to_jsonb(p) from kb_stage.kb_proposals p
+    where id = {q(proposal_id)}::uuid),
+  'counts', jsonb_build_object(
+    'source_captures', (select count(*) from kb_canary.source_captures),
+    'claim_extractions', (select count(*) from kb_canary.claim_extractions),
+    'evidence_extractions', (select count(*) from kb_canary.evidence_extractions),
+    'staged_proposals', (select count(*) from kb_stage.kb_proposals)
+  )
+)::text;
+rollback;
+"""
+
+
+class DockerPostgres:
+    def __init__(self, name: str) -> None:
+        self.name = name
+        self.docker = shutil.which("docker")
+        if not self.docker:
+            raise CanaryError("docker executable is unavailable")
+        self.started = False
+        self.volume_mounts: list[dict[str, Any]] = []
+
+    def _run(
+        self, args: list[str], *, input_text: str | None = None, check: bool = True
+    ) -> subprocess.CompletedProcess[str]:
+        result = subprocess.run([self.docker, *args], input=input_text, text=True, capture_output=True, check=False)
+        if check and result.returncode != 0:
+            detail = (result.stderr or result.stdout).strip()
+            raise CanaryError(f"docker {' '.join(args[:3])} failed ({result.returncode}): {detail}")
+        return result
+
+    def start(self) -> dict[str, Any]:
+        result = self._run(
+            [
+                "run",
+                "--rm",
+                "-d",
+                "--name",
+                self.name,
+                "--network",
+                "none",
+                "--label",
+                CONTAINER_LABEL,
+                "--tmpfs",
+                "/var/lib/postgresql/data:rw,nosuid,size=256m",
+                "-e",
+                "POSTGRES_HOST_AUTH_METHOD=trust",
+                "-e",
+                "POSTGRES_DB=teleo",
+                "postgres:16-alpine",
+            ]
+        )
+        self.started = True
+        container_id = result.stdout.strip()
+        deadline = time.monotonic() + 30
+        ready_streak = 0
+        while time.monotonic() < deadline:
+            ready = self._run(["exec", self.name, "pg_isready", "-U", "postgres", "-d", "teleo"], check=False)
+            if ready.returncode == 0:
+                ready_streak += 1
+                if ready_streak >= 2:
+                    break
+            else:
+                ready_streak = 0
+            time.sleep(0.25)
+        else:
+            raise CanaryError("disposable Postgres did not become ready within 30 seconds")
+        inspect = json.loads(self._run(["inspect", self.name]).stdout)[0]
+        if inspect["HostConfig"]["NetworkMode"] != "none":
+            raise CanaryError("disposable Postgres must use network mode none")
+        if inspect["Config"]["Labels"].get("livingip.canary") != "working-leo-local-ingestion":
+            raise CanaryError("disposable Postgres is missing its canary label")
+        self.volume_mounts = [
+            {
+                "type": mount.get("Type"),
+                "name": mount.get("Name"),
+                "destination": mount.get("Destination"),
+            }
+            for mount in inspect.get("Mounts", [])
+            if mount.get("Type") == "volume"
+        ]
+        if self.volume_mounts:
+            raise CanaryError("disposable Postgres unexpectedly created a Docker volume")
+        return {
+            "container_id": container_id,
+            "name": self.name,
+            "image": inspect["Config"]["Image"],
+            "network_mode": inspect["HostConfig"]["NetworkMode"],
+            "tmpfs_data_dir": "/var/lib/postgresql/data" in inspect["HostConfig"].get("Tmpfs", {}),
+            "docker_volume_mounts": self.volume_mounts,
+            "canary_label": inspect["Config"]["Labels"].get("livingip.canary"),
+        }
+
+    def psql(self, sql: str) -> str:
+        result = self._run(
+            ["exec", "-i", self.name, "psql", "-U", "postgres", "-d", "teleo", "-Atq"],
+            input_text=sql,
+        )
+        return result.stdout.strip()
+
+    def psql_json(self, sql: str) -> dict[str, Any]:
+        output = self.psql(sql)
+        lines = [line for line in output.splitlines() if line.strip()]
+        if not lines:
+            raise CanaryError("Postgres readback returned no JSON")
+        try:
+            value = json.loads(lines[-1])
+        except json.JSONDecodeError as exc:
+            raise CanaryError(f"Postgres readback was not JSON: {lines[-1]}") from exc
+        if not isinstance(value, dict):
+            raise CanaryError("Postgres readback must be a JSON object")
+        return value
+
+    def cleanup(self) -> dict[str, Any]:
+        remove = self._run(["rm", "-f", self.name], check=False) if self.started else None
+        inspect = self._run(["inspect", self.name], check=False)
+        return {
+            "remove_attempted": self.started,
+            "remove_returncode": remove.returncode if remove else None,
+            "container_absent": inspect.returncode != 0,
+            "docker_volume_mounts_observed": self.volume_mounts,
+            "anonymous_or_named_volume_created": bool(self.volume_mounts),
+        }
+
+
+def proposal_links(readback: dict[str, Any], input_receipt: dict[str, Any]) -> dict[str, Any]:
+    proposal = readback["staged_proposal"]
+    apply_payload = proposal["payload"]["apply_payload"]
+    source_rows = apply_payload["sources"]
+    evidence_rows = apply_payload["evidence"]
+    return {
+        "proposal_id": proposal["id"],
+        "proposal_status": proposal["status"],
+        "proposal_source_ref": proposal["source_ref"],
+        "planned_claim_ids": [row["id"] for row in apply_payload["claims"]],
+        "planned_source_ids": [row["id"] for row in source_rows],
+        "planned_evidence_keys": [
+            {"claim_id": row["claim_id"], "source_id": row["source_id"], "role": row["role"]} for row in evidence_rows
+        ],
+        "source_content_hash_matches": any(
+            row.get("hash") == input_receipt["source_content_sha256"] for row in source_rows
+        ),
+        "parent_source_ref_embedded": any(
+            readback["source_capture"]["artifact_sha256"] in str(row.get("excerpt"))
+            and readback["source_capture"]["id"] in str(row.get("excerpt"))
+            and readback["claim_extraction"]["id"] in str(row.get("excerpt"))
+            and readback["evidence_extraction"]["id"] in str(row.get("excerpt"))
+            for row in source_rows
+        ),
+    }
+
+
+def evaluate_checks(
+    fixture: dict[str, Any], input_receipt: dict[str, Any], row_ids: dict[str, str], readback: dict[str, Any]
+) -> dict[str, bool]:
+    source = readback.get("source_capture") or {}
+    claim = readback.get("claim_extraction") or {}
+    evidence = readback.get("evidence_extraction") or {}
+    proposal = readback.get("staged_proposal") or {}
+    links = proposal_links(readback, input_receipt)
+    return {
+        "input_artifact_hash_persisted": source.get("artifact_sha256") == input_receipt["artifact_sha256"],
+        "source_content_hash_persisted": source.get("content_sha256") == input_receipt["source_content_sha256"],
+        "source_identity_persisted": source.get("source_identity") == input_receipt["source_identity"],
+        "claim_links_to_source_capture": claim.get("source_capture_id") == row_ids["source_capture_id"],
+        "claim_body_and_metadata_persisted": (
+            claim.get("body") == fixture["extraction"]["claim"]["body"]
+            and claim.get("metadata", {}).get("text") == fixture["extraction"]["claim"]["text"]
+        ),
+        "evidence_links_to_claim_and_source": (
+            evidence.get("claim_extraction_id") == row_ids["claim_extraction_id"]
+            and evidence.get("source_capture_id") == row_ids["source_capture_id"]
+        ),
+        "evidence_body_and_metadata_persisted": (
+            evidence.get("body") == fixture["extraction"]["evidence"]["body"]
+            and evidence.get("metadata") == fixture["extraction"]["evidence"]["metadata"]
+        ),
+        "one_row_per_ingestion_stage": readback.get("counts")
+        == {"source_captures": 1, "claim_extractions": 1, "evidence_extractions": 1, "staged_proposals": 1},
+        "proposal_is_pending_review": proposal.get("status") == "pending_review",
+        "proposal_has_source_ref": bool(proposal.get("source_ref")),
+        "proposal_payload_contains_input_hash": links["source_content_hash_matches"],
+        "proposal_payload_contains_capture_row_linkage": links["parent_source_ref_embedded"],
+        "no_review_or_apply_metadata": all(
+            proposal.get(key) is None
+            for key in ("reviewed_by_handle", "reviewed_at", "applied_by_handle", "applied_at")
+        ),
+    }
+
+
+def run_canary(fixture_path: Path, output: Path | None = None) -> dict[str, Any]:
+    fixture, input_receipt = load_fixture(fixture_path)
+    row_ids = build_row_ids(input_receipt["artifact_sha256"])
+    parent = build_parent_packet(fixture, input_receipt, row_ids)
+    child = normalized_stage.prepare_normalized_child(parent)
+    container_name = f"working-leo-ingest-{uuid.uuid4().hex[:12]}"
+    postgres = DockerPostgres(container_name)
+    receipt: dict[str, Any] = {
+        "schema": SCHEMA,
+        "status": "fail",
+        "required_tier": "isolated_realistic_ingestion_to_proposal",
+        "input": input_receipt,
+        "row_ids": row_ids,
+        "production_mutation_attempted": False,
+        "canonical_apply_attempted": False,
+    }
+    try:
+        receipt["environment"] = postgres.start()
+        postgres.psql(build_schema_sql())
+        postgres.psql(build_capture_sql(fixture, input_receipt, row_ids))
+        stage_result = postgres.psql_json(normalized_stage.build_stage_sql(child))
+        readback = postgres.psql_json(build_readback_sql(row_ids, child["id"]))
+        checks = evaluate_checks(fixture, input_receipt, row_ids, readback)
+        receipt.update(
+            {
+                "stage_command_result": stage_result,
+                "rows": readback,
+                "row_link_fields_proven": proposal_links(readback, input_receipt),
+                "checks": checks,
+                "status": "pass" if all(checks.values()) else "fail",
+            }
+        )
+    except (CanaryError, OSError, ValueError, KeyError) as exc:
+        receipt["error"] = {"type": type(exc).__name__, "message": str(exc)}
+    finally:
+        receipt["cleanup"] = postgres.cleanup()
+        receipt["cleanup"]["network_mode_none"] = (
+            receipt.get("environment", {}).get("network_mode") == "none"
+        )
+        receipt["cleanup"]["no_production_target_referenced"] = receipt[
+            "cleanup"
+        ]["network_mode_none"]
+        if not all(
+            (
+                receipt["cleanup"]["container_absent"],
+                receipt["cleanup"]["network_mode_none"],
+            )
+        ):
+            receipt["status"] = "fail"
+        receipt["strongest_claim"] = (
+            "A realistic local document fixture was hash-captured, replayed through deterministic claim/evidence "
+            "extraction, normalized by the existing rich-proposal normalizer, and staged/read back as one "
+            "pending_review proposal in disposable networkless Postgres."
+        )
+        receipt["residual_gap"] = (
+            "This does not prove live generative extraction, arbitrary document coverage, canonical apply, or any "
+            "hosted/production runtime."
+        )
+        if output:
+            output.parent.mkdir(parents=True, exist_ok=True)
+            output.write_text(json.dumps(receipt, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+    return receipt
+
+
+def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--fixture", type=Path, default=DEFAULT_FIXTURE)
+    parser.add_argument("--output", type=Path)
+    return parser.parse_args(argv)
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = parse_args(argv)
+    receipt = run_canary(args.fixture.resolve(), args.output.resolve() if args.output else None)
+    print(json.dumps(receipt, sort_keys=True, separators=(",", ":")))
+    return 0 if receipt["status"] == "pass" else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_run_approve_claim_clone_canary.py

@@ -0,0 +1,79 @@
+"""Focused tests for the strict approved-change clone receipt."""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / "scripts"))
+
+import run_approve_claim_clone_canary as canary  # noqa: E402
+
+
+def _passing_result(expected_rows: dict, expected_deltas: dict) -> dict:
+    proposal_id = "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa"
+    return {
+        "status": "pass",
+        "fixture_ids": {"proposal_id": proposal_id},
+        "apply_transition": {
+            "applied_readback": {
+                "id": proposal_id,
+                "status": "applied",
+                "applied_at": "2026-07-12 12:00:00+00",
+                "applied_by_handle": "kb-apply",
+            }
+        },
+        "row_readback": expected_rows,
+        "clone_counts_before_apply": {key: 0 for key in expected_deltas},
+        "clone_counts_after_apply": expected_deltas,
+        "clone_apply_table_deltas": expected_deltas,
+        "clone_dropped": True,
+        "cleanup_readback": {"leftover_clone_database_count": 0},
+        "source_database_mutated": False,
+        "checks": {"conflict_transaction_rolled_back_exactly": True},
+    }
+
+
+def test_service_state_records_unavailable_systemctl(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(
+        canary,
+        "_run",
+        lambda *_args, **_kwargs: (_ for _ in ()).throw(FileNotFoundError()),
+    )
+
+    assert canary._service_state("unused.service") == {
+        "available": False,
+        "returncode": 127,
+        "reason": "systemctl_unavailable",
+    }
+
+
+def test_strict_receipt_requires_exact_ids_counts_applied_at_and_cleanup() -> None:
+    expected_rows = canary._expected_bundle_rows(canary.build_fixture("unit")["apply_payload"])
+    expected_deltas = canary._expected_table_deltas(expected_rows)
+    result = _passing_result(expected_rows, expected_deltas)
+
+    receipt = canary._strict_receipt(result, expected_rows, expected_deltas)
+
+    assert receipt["pass"] is True
+    assert receipt["proposal"]["status"] == "applied"
+    assert receipt["proposal"]["applied_at"]
+    assert receipt["canonical_row_keys"]["actual"] == receipt["canonical_row_keys"]["expected"]
+    assert receipt["row_counts"]["actual_deltas"] == expected_deltas
+    assert receipt["rollback_cleanup"]["clone_database_removed"] is True
+
+    result["apply_transition"]["applied_readback"]["applied_at"] = None
+    result["row_readback"] = {**expected_rows, "claims": []}
+    result["clone_apply_table_deltas"] = {**expected_deltas, "public.claims": 0}
+    result["clone_dropped"] = False
+
+    failed = canary._strict_receipt(result, expected_rows, expected_deltas)
+
+    assert failed["pass"] is False
+    assert failed["checks"]["proposal_applied_at_present"] is False
+    assert failed["checks"]["canonical_row_keys_exact"] is False
+    assert failed["checks"]["table_deltas_exact"] is False
+    assert failed["checks"]["clone_database_removed"] is False

tests/test_run_approve_claim_clone_canary_macos_path.py

@@ -0,0 +1,102 @@
+"""Platform-path regression tests for the approved-change clone canary."""
+
+from __future__ import annotations
+
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / "scripts"))
+WRAPPER = ROOT / "scripts" / "run_approve_claim_isolated_container_canary.sh"
+
+import apply_proposal as apply  # noqa: E402
+import run_approve_claim_clone_canary as canary  # noqa: E402
+
+
+def test_role_psql_resolves_docker_before_scrubbing_path(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    captured: dict[str, object] = {}
+
+    monkeypatch.setattr(
+        canary.shutil,
+        "which",
+        lambda executable: "/opt/homebrew/bin/docker"
+        if executable == "docker"
+        else None,
+    )
+
+    def fake_run(command: list[str], **kwargs: object) -> subprocess.CompletedProcess[str]:
+        captured["command"] = command
+        captured["env"] = kwargs["env"]
+        return subprocess.CompletedProcess(command, 0, "", "")
+
+    monkeypatch.setattr(canary, "_run", fake_run)
+
+    canary._role_psql(
+        "local-postgres",
+        "teleo",
+        "select 1;",
+        role="kb_apply",
+        password="test-only-password",
+    )
+
+    assert captured["command"][0] == "/opt/homebrew/bin/docker"
+    assert captured["env"] == {
+        "PGPASSWORD": "test-only-password",
+        "PATH": "/usr/bin:/bin:/usr/local/bin",
+    }
+
+
+def test_apply_tool_resolves_docker_before_scrubbing_path(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    captured: dict[str, object] = {}
+    monkeypatch.setattr(
+        apply.shutil,
+        "which",
+        lambda executable: "/opt/homebrew/bin/docker"
+        if executable == "docker"
+        else None,
+    )
+
+    def fake_run(command: list[str], **kwargs: object) -> subprocess.CompletedProcess[str]:
+        captured["command"] = command
+        captured["env"] = kwargs["env"]
+        return subprocess.CompletedProcess(command, 0, "ok", "")
+
+    monkeypatch.setattr(apply.subprocess, "run", fake_run)
+    args = type(
+        "Args",
+        (),
+        {
+            "container": "local-postgres",
+            "role": "kb_apply",
+            "host": "127.0.0.1",
+            "db": "teleo",
+        },
+    )()
+
+    assert apply.run_psql(args, "select 1;", "test-only-password") == "ok"
+    assert captured["command"][0] == "/opt/homebrew/bin/docker"
+    assert captured["env"] == {
+        "PGPASSWORD": "test-only-password",
+        "PATH": "/usr/bin:/bin:/usr/local/bin",
+    }
+
+
+def test_outer_wrapper_enforces_local_source_and_networkless_tmpfs() -> None:
+    text = WRAPPER.read_text(encoding="utf-8")
+
+    assert 'source_canary_label" != "working-leo-approved-source"' in text
+    assert "--network none" in text
+    assert "--tmpfs /var/lib/postgresql/data:rw,nosuid,size=512m" in text
+    assert 'CANARY_NETWORK_MODE="$container_network_mode"' in text
+    assert 'CANARY_MOUNTS_JSON="$container_mounts_json"' in text
+    assert 'data["checks"]["outer_source_tier_enforced"]' in text
+    assert 'data["checks"]["outer_container_network_disabled"]' in text
+    assert 'data["checks"]["outer_container_has_no_docker_volumes"]' in text
+    assert 'source_tier != "live-readonly" or service_observable' in text

tests/test_run_leo_local_ingestion_proposal_canary.py

@@ -0,0 +1,164 @@
+from __future__ import annotations
+
+import copy
+import json
+import sys
+from pathlib import Path
+
+import pytest
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+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()
+    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
+
+
+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)
+    )
+    sql = "\n".join(
+        (
+            canary.build_schema_sql(),
+            canary.build_capture_sql(fixture, input_receipt, row_ids),
+            canary.normalized_stage.build_stage_sql(child),
+        )
+    ).lower()
+
+    assert "insert into public." not in sql
+    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
+
+
+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)
+    )
+    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,
+        },
+    }
+
+    checks = canary.evaluate_checks(fixture, input_receipt, row_ids, readback)
+    assert all(checks.values())
+    broken = copy.deepcopy(readback)
+    broken["evidence_extraction"]["claim_extraction_id"] = canary.stable_uuid("0" * 64, "wrong")
+    assert (
+        canary.evaluate_checks(fixture, input_receipt, row_ids, broken)["evidence_links_to_claim_and_source"] is False
+    )