diff --git a/scripts/run_leo_integrated_learning_lifecycle.py b/scripts/run_leo_integrated_learning_lifecycle.py
new file mode 100644
index 0000000..13e401d
--- /dev/null
+++ b/scripts/run_leo_integrated_learning_lifecycle.py
@@ -0,0 +1,900 @@
+#!/usr/bin/env python3
+"""Run a real source -> review -> apply -> recompile lifecycle in disposable Postgres.
+
+The harness composes the repository's existing source compiler, normalized
+proposal staging, review packet, guarded approval, guarded apply, and replay
+receipt surfaces.  It starts its own networkless tmpfs Postgres container and
+never accepts an existing database target, so the explicit review/apply actions
+cannot reach VPS, GCP, or another persistent database.
+
+The source packet must contain:
+
+* ``prepared/extracted-text.txt``
+* ``prepared/source-manifest.json``
+* ``proposal-packet.json``
+
+The extracted text is used as both the hash-bound source artifact and its strict
+UTF-8 extraction.  ``proposal-packet.json`` is an independently retained
+compiler expectation: the harness refuses to continue unless a fresh compile
+matches it byte-for-byte under canonical JSON.
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+import secrets
+import shutil
+import subprocess
+import sys
+import tempfile
+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 compile_kb_source_packet as compiler  # noqa: E402
+import kb_proposal_review_packet as review_packets  # noqa: E402
+import proposal_apply_lifecycle as apply_lifecycle  # noqa: E402
+import stage_normalized_proposal as stage  # noqa: E402
+
+RECEIPT_SCHEMA = "livingip.leoIntegratedLearningLifecycleReceipt.v1"
+CONTAINER_LABEL = "livingip.canary=leo-integrated-learning-lifecycle"
+DATABASE = "teleo"
+REVIEWER_HANDLE = "m3ta"
+REVIEWER_ID = "11111111-1111-4111-8111-111111111111"
+DEFAULT_REVIEW_NOTE = (
+    "Approved the exact hash-bound candidate rows for this disposable integrated-learning lifecycle only."
+)
+POSTGRES_IMAGE = "postgres:16-alpine"
+
+
+class LifecycleError(RuntimeError):
+    """Raised when the isolated lifecycle cannot prove an exact transition."""
+
+
+def canonical_json_bytes(value: Any) -> bytes:
+    return json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=True).encode("utf-8")
+
+
+def canonical_sha256(value: Any) -> str:
+    return hashlib.sha256(canonical_json_bytes(value)).hexdigest()
+
+
+def sha256_text(value: str) -> str:
+    return hashlib.sha256(value.encode("utf-8")).hexdigest()
+
+
+def sha256_file(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as handle:
+        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def parse_last_json_line(output: str) -> Any:
+    for line in reversed(output.splitlines()):
+        line = line.strip()
+        if not line:
+            continue
+        try:
+            return json.loads(line)
+        except json.JSONDecodeError:
+            continue
+    return None
+
+
+def _safe_error_text(value: str, *, limit: int = 1200) -> str:
+    compact = " ".join(value.split())
+    return compact[-limit:]
+
+
+def command_receipt(result: subprocess.CompletedProcess[str], *, expected_marker: str | None = None) -> dict[str, Any]:
+    parsed = parse_last_json_line(result.stdout)
+    if isinstance(parsed, dict):
+        parsed = {key: value for key, value in parsed.items() if key not in {"receipt_path"}}
+    return {
+        "returncode": result.returncode,
+        "stdout_sha256": sha256_text(result.stdout),
+        "stderr_sha256": sha256_text(result.stderr),
+        "stdout_json": parsed if isinstance(parsed, dict) else None,
+        "expected_marker": expected_marker,
+        "expected_marker_present": expected_marker in result.stdout if expected_marker else None,
+    }
+
+
+def run_tool(arguments: list[str], *, action: str) -> subprocess.CompletedProcess[str]:
+    result = subprocess.run(arguments, text=True, capture_output=True, check=False)
+    if result.returncode != 0:
+        detail = _safe_error_text(result.stderr or result.stdout)
+        raise LifecycleError(f"{action} failed ({result.returncode}): {detail}")
+    return result
+
+
+class DisposablePostgres:
+    def __init__(self, name: str, admin_password: str) -> None:
+        self.name = name
+        self.admin_password = admin_password
+        self.docker = shutil.which("docker")
+        if not self.docker:
+            raise LifecycleError("docker executable is unavailable")
+        self.started = False
+        self.environment: dict[str, Any] = {}
+
+    def _run(
+        self,
+        arguments: list[str],
+        *,
+        input_text: str | None = None,
+        check: bool = True,
+    ) -> subprocess.CompletedProcess[str]:
+        result = subprocess.run(
+            [self.docker, *arguments],
+            input=input_text,
+            text=True,
+            capture_output=True,
+            check=False,
+        )
+        if check and result.returncode != 0:
+            detail = _safe_error_text(result.stderr or result.stdout)
+            raise LifecycleError(f"docker {arguments[0]} 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,
+                "--label",
+                f"livingip.canary.instance={self.name}",
+                "--tmpfs",
+                "/var/lib/postgresql/data:rw,nosuid,size=384m",
+                "-e",
+                f"POSTGRES_PASSWORD={self.admin_password}",
+                "-e",
+                f"POSTGRES_DB={DATABASE}",
+                POSTGRES_IMAGE,
+            ]
+        )
+        self.started = True
+        deadline = time.monotonic() + 45
+        ready_streak = 0
+        while time.monotonic() < deadline:
+            ready = self._run(
+                ["exec", self.name, "pg_isready", "-U", "postgres", "-d", DATABASE],
+                check=False,
+            )
+            if ready.returncode == 0:
+                ready_streak += 1
+                if ready_streak >= 2:
+                    break
+            else:
+                ready_streak = 0
+            time.sleep(0.25)
+        else:
+            raise LifecycleError("disposable Postgres did not become stably ready")
+
+        inspect = json.loads(self._run(["inspect", self.name]).stdout)[0]
+        mounts = [
+            {
+                "type": mount.get("Type"),
+                "name": mount.get("Name"),
+                "destination": mount.get("Destination"),
+            }
+            for mount in inspect.get("Mounts", [])
+        ]
+        labels = inspect.get("Config", {}).get("Labels", {})
+        self.environment = {
+            "container_id": result.stdout.strip(),
+            "container_name": self.name,
+            "image": inspect.get("Config", {}).get("Image"),
+            "network_mode": inspect.get("HostConfig", {}).get("NetworkMode"),
+            "tmpfs_data_dir": "/var/lib/postgresql/data" in inspect.get("HostConfig", {}).get("Tmpfs", {}),
+            "mounts": mounts,
+            "canary_label": labels.get("livingip.canary"),
+            "instance_label": labels.get("livingip.canary.instance"),
+        }
+        if self.environment["network_mode"] != "none":
+            raise LifecycleError("disposable Postgres is not networkless")
+        if not self.environment["tmpfs_data_dir"] or mounts:
+            raise LifecycleError("disposable Postgres must use tmpfs without Docker volumes")
+        if self.environment["canary_label"] != "leo-integrated-learning-lifecycle":
+            raise LifecycleError("disposable Postgres canary label is missing")
+        if self.environment["instance_label"] != self.name:
+            raise LifecycleError("disposable Postgres instance label is missing")
+        return self.environment
+
+    def psql(self, sql: str) -> str:
+        result = self._run(
+            [
+                "exec",
+                "-i",
+                self.name,
+                "psql",
+                "-X",
+                "-U",
+                "postgres",
+                "-d",
+                DATABASE,
+                "-v",
+                "ON_ERROR_STOP=1",
+                "-Atq",
+            ],
+            input_text=sql,
+        )
+        return result.stdout.strip()
+
+    def psql_json(self, sql: str) -> Any:
+        output = self.psql(sql)
+        parsed = parse_last_json_line(output)
+        if parsed is None:
+            raise LifecycleError("Postgres readback did not contain JSON")
+        return parsed
+
+    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)
+        names = self._run(
+            ["container", "ls", "-a", "--filter", f"name=^/{self.name}$", "--format", "{{.Names}}"],
+            check=False,
+        )
+        labels = self._run(
+            [
+                "container",
+                "ls",
+                "-a",
+                "--filter",
+                "label=livingip.canary=leo-integrated-learning-lifecycle",
+                "--filter",
+                f"label=livingip.canary.instance={self.name}",
+                "--format",
+                "{{.Names}}",
+            ],
+            check=False,
+        )
+        return {
+            "remove_attempted": self.started,
+            "remove_returncode": remove.returncode if remove else None,
+            "container_absent": inspect.returncode != 0,
+            "name_readback_clear": names.returncode == 0 and not names.stdout.strip(),
+            "instance_label_readback_clear": labels.returncode == 0 and not labels.stdout.strip(),
+            "network_mode_was_none": self.environment.get("network_mode") == "none",
+            "tmpfs_data_dir_was_present": self.environment.get("tmpfs_data_dir") is True,
+            "docker_volume_mounts_observed": self.environment.get("mounts", []),
+        }
+
+
+def bootstrap_sql(review_password: str, apply_password: str) -> str:
+    return f"""\\set ON_ERROR_STOP on
+create role kb_apply login noinherit password {ap.sql_literal(apply_password)};
+create role kb_review login noinherit password {ap.sql_literal(review_password)};
+create schema kb_stage;
+create type evidence_role as enum ('grounds', 'illustrates', 'contradicts');
+create type edge_type as enum (
+  'supports', 'challenges', 'requires', 'relates', 'contradicts', 'supersedes',
+  'derives_from', 'cites', 'causes', 'constrains', 'accelerates'
+);
+
+create table public.agents (
+  id uuid primary key,
+  handle text not null unique,
+  kind text not null
+);
+create table public.claims (
+  id uuid primary key,
+  type text not null,
+  text text not null,
+  status text not null,
+  confidence numeric,
+  tags text[] not null default '{{}}',
+  created_by uuid references public.agents(id),
+  superseded_by uuid references public.claims(id),
+  created_at timestamptz not null default now(),
+  updated_at timestamptz not null default now()
+);
+create table public.sources (
+  id uuid primary key,
+  source_type text not null,
+  url text,
+  storage_path text,
+  excerpt text,
+  hash text not null,
+  created_by uuid references public.agents(id),
+  captured_at timestamptz,
+  created_at timestamptz not null default now()
+);
+create table public.claim_evidence (
+  id uuid primary key default gen_random_uuid(),
+  claim_id uuid not null references public.claims(id),
+  source_id uuid not null references public.sources(id),
+  role evidence_role not null,
+  weight numeric,
+  created_by uuid references public.agents(id),
+  created_at timestamptz not null default now()
+);
+create table public.claim_edges (
+  id uuid primary key,
+  from_claim uuid not null references public.claims(id),
+  to_claim uuid not null references public.claims(id),
+  edge_type edge_type not null,
+  weight numeric,
+  created_by uuid references public.agents(id),
+  created_at timestamptz not null default now()
+);
+create table public.reasoning_tools (
+  id uuid primary key,
+  agent_id uuid references public.agents(id),
+  name text not null,
+  description text not null,
+  category text,
+  created_at timestamptz not null default now(),
+  updated_at timestamptz not null default now()
+);
+create table public.strategies (
+  id uuid primary key default gen_random_uuid(),
+  agent_id uuid not null references public.agents(id),
+  diagnosis text,
+  guiding_policy text,
+  proximate_objectives jsonb,
+  version integer not null default 1,
+  active boolean not null default true,
+  created_at timestamptz not null default now(),
+  unique (agent_id, version)
+);
+create table public.strategy_nodes (
+  id uuid primary key default gen_random_uuid(),
+  agent_id uuid references public.agents(id),
+  node_type text,
+  title text,
+  body text,
+  rank integer,
+  status text not null default 'active',
+  horizon text,
+  measure text,
+  source_ref text,
+  metadata jsonb not null default '{{}}',
+  created_at timestamptz not null default now(),
+  updated_at timestamptz not null default now()
+);
+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 references public.agents(id),
+  channel text,
+  source_ref text,
+  rationale text,
+  payload jsonb not null,
+  reviewed_by_handle text,
+  reviewed_by_agent_id uuid references public.agents(id),
+  reviewed_at timestamptz,
+  review_note text,
+  applied_by_handle text,
+  applied_by_agent_id uuid references public.agents(id),
+  applied_at timestamptz,
+  created_at timestamptz not null default now(),
+  updated_at timestamptz not null default now()
+);
+insert into public.agents (id, handle, kind)
+values ({ap.sql_literal(REVIEWER_ID)}::uuid, {ap.sql_literal(REVIEWER_HANDLE)}, 'human');
+"""
+
+
+def compile_source_packet(source_packet: Path) -> tuple[dict[str, Any], dict[str, Any], dict[str, Path]]:
+    paths = {
+        "artifact": source_packet / "prepared" / "extracted-text.txt",
+        "text": source_packet / "prepared" / "extracted-text.txt",
+        "manifest": source_packet / "prepared" / "source-manifest.json",
+        "expected": source_packet / "proposal-packet.json",
+    }
+    missing = [str(path) for path in paths.values() if not path.is_file()]
+    if missing:
+        raise LifecycleError(f"source packet is missing required files: {missing}")
+    expected = json.loads(paths["expected"].read_text(encoding="utf-8"))
+    if not isinstance(expected, dict):
+        raise LifecycleError("proposal-packet.json must contain one JSON object")
+    compiled = compiler.compile_source_packet(paths["artifact"], paths["text"], paths["manifest"])
+    if canonical_json_bytes(compiled) != canonical_json_bytes(expected):
+        raise LifecycleError("fresh compiler output does not match retained proposal-packet.json")
+    return compiled, expected, paths
+
+
+def proposal_readback(postgres: DisposablePostgres, proposal_id: str) -> dict[str, Any] | None:
+    sql = f"""select jsonb_build_object(
+  'id', id::text,
+  'proposal_type', proposal_type,
+  'status', status,
+  'proposed_by_handle', proposed_by_handle,
+  'proposed_by_agent_id', proposed_by_agent_id::text,
+  'channel', channel,
+  'source_ref', source_ref,
+  'rationale', rationale,
+  'payload', payload,
+  'reviewed_by_handle', reviewed_by_handle,
+  'reviewed_by_agent_id', reviewed_by_agent_id::text,
+  'reviewed_at', reviewed_at::text,
+  'review_note', review_note,
+  'applied_by_handle', applied_by_handle,
+  'applied_by_agent_id', applied_by_agent_id::text,
+  'applied_at', applied_at::text,
+  'created_at', created_at::text,
+  'updated_at', updated_at::text
+)::text
+from kb_stage.kb_proposals
+where id = {ap.sql_literal(proposal_id)}::uuid;"""
+    parsed = parse_last_json_line(postgres.psql(sql))
+    if parsed is not None and not isinstance(parsed, dict):
+        raise LifecycleError("proposal readback was not an object")
+    return parsed
+
+
+def approval_readback(postgres: DisposablePostgres, proposal_id: str) -> dict[str, Any] | None:
+    sql = f"""select jsonb_build_object(
+  'proposal_id', proposal_id::text,
+  'proposal_type', proposal_type,
+  'payload', payload,
+  'reviewed_by_handle', reviewed_by_handle,
+  'reviewed_by_agent_id', reviewed_by_agent_id::text,
+  'reviewed_by_db_role', reviewed_by_db_role::text,
+  'reviewed_at', reviewed_at::text,
+  'review_note', review_note
+)::text
+from kb_stage.kb_proposal_approvals
+where proposal_id = {ap.sql_literal(proposal_id)}::uuid;"""
+    parsed = parse_last_json_line(postgres.psql(sql))
+    if parsed is not None and not isinstance(parsed, dict):
+        raise LifecycleError("approval readback was not an object")
+    return parsed
+
+
+def probe_applied_proposal(child: dict[str, Any]) -> dict[str, Any]:
+    return {
+        **child,
+        "status": "applied",
+        "applied_by_handle": ap.SERVICE_AGENT_HANDLE,
+        "applied_at": "2000-01-01T00:00:00+00:00",
+    }
+
+
+def target_rows_readback(postgres: DisposablePostgres, child: dict[str, Any]) -> dict[str, list[dict[str, Any]]]:
+    sql = ap.replay_receipt.build_postflight_sql(probe_applied_proposal(child))
+    parsed = postgres.psql_json(sql)
+    if not isinstance(parsed, dict) or any(not isinstance(rows, list) for rows in parsed.values()):
+        raise LifecycleError("canonical target-row readback was malformed")
+    return parsed
+
+
+def agents_readback(postgres: DisposablePostgres) -> list[dict[str, Any]]:
+    parsed = postgres.psql_json(
+        "select coalesce(jsonb_agg(to_jsonb(a) order by a.id::text), '[]'::jsonb)::text from public.agents a;"
+    )
+    if not isinstance(parsed, list):
+        raise LifecycleError("agent readback was malformed")
+    return parsed
+
+
+def database_state(postgres: DisposablePostgres, child: dict[str, Any]) -> dict[str, Any]:
+    proposal_id = str(child["id"])
+    canonical_rows = target_rows_readback(postgres, child)
+    state = {
+        "agents": agents_readback(postgres),
+        "canonical_rows": canonical_rows,
+        "proposal": proposal_readback(postgres, proposal_id),
+        "immutable_approval": approval_readback(postgres, proposal_id),
+    }
+    return {
+        "database_fingerprint": canonical_sha256(state),
+        "canonical_fingerprint": canonical_sha256(canonical_rows),
+        "state": state,
+    }
+
+
+def exact_identities(
+    child: dict[str, Any], canonical_rows: dict[str, list[dict[str, Any]]] | None = None
+) -> dict[str, Any]:
+    apply_payload = child["payload"]["apply_payload"]
+    evidence_rows = canonical_rows.get("claim_evidence", []) if canonical_rows else []
+    return {
+        "proposal_id": child["id"],
+        "claim_ids": [row["id"] for row in apply_payload.get("claims", [])],
+        "source_ids": [row["id"] for row in apply_payload.get("sources", [])],
+        "planned_evidence_keys": [
+            {
+                "claim_id": row["claim_id"],
+                "source_id": row["source_id"],
+                "role": row.get("role") or "grounds",
+            }
+            for row in apply_payload.get("evidence", [])
+        ],
+        "evidence_ids": [row["id"] for row in evidence_rows if row.get("id")],
+        "edge_ids": [row["id"] for row in canonical_rows.get("claim_edges", [])] if canonical_rows else [],
+    }
+
+
+def review_packet(proposal: dict[str, Any]) -> dict[str, Any]:
+    packet = review_packets.classify_proposal(proposal)
+    apply_payload = proposal["payload"]["apply_payload"]
+    packet["candidate_rows"] = {
+        family: apply_payload.get(family, [])
+        for family in ("claims", "sources", "evidence", "edges", "reasoning_tools")
+    }
+    packet["packet_sha256"] = canonical_sha256(packet)
+    return packet
+
+
+def write_secret_file(path: Path, key: str, value: str) -> None:
+    path.write_text(f"{key}={value}\n", encoding="utf-8")
+    path.chmod(0o600)
+
+
+def _tool_base(script: Path, proposal_id: str, secrets_file: Path, role: str, container: str) -> list[str]:
+    return [
+        sys.executable,
+        str(script),
+        proposal_id,
+        "--secrets-file",
+        str(secrets_file),
+        "--container",
+        container,
+        "--db",
+        DATABASE,
+        "--host",
+        "127.0.0.1",
+        "--role",
+        role,
+    ]
+
+
+def _all_target_rows_empty(rows: dict[str, list[dict[str, Any]]]) -> bool:
+    return set(rows) == set(apply_lifecycle.ROW_TABLES) and all(not table_rows for table_rows in rows.values())
+
+
+def _row_counts(rows: dict[str, list[dict[str, Any]]]) -> dict[str, int]:
+    return {family: len(table_rows) for family, table_rows in rows.items()}
+
+
+def normalized_canonical_rows(rows: dict[str, list[dict[str, Any]]]) -> dict[str, list[dict[str, Any]]]:
+    return {family: sorted(table_rows, key=canonical_json_bytes) for family, table_rows in sorted(rows.items())}
+
+
+def run_lifecycle(
+    source_packet: Path,
+    output: Path,
+    *,
+    review_action: str,
+    reviewed_by: str = REVIEWER_HANDLE,
+    review_note: str = DEFAULT_REVIEW_NOTE,
+) -> dict[str, Any]:
+    if review_action != "approve":
+        raise LifecycleError("the only supported explicit review action is 'approve'")
+    if reviewed_by != REVIEWER_HANDLE:
+        raise LifecycleError(f"disposable reviewer must be {REVIEWER_HANDLE!r}")
+    if not review_note.strip():
+        raise LifecycleError("review note must be non-empty")
+
+    source_packet = source_packet.resolve()
+    output = output.resolve()
+    temp_root = Path(tempfile.mkdtemp(prefix="leo-integrated-learning-"))
+    private_apply_receipt_path = temp_root / "apply-receipt.json"
+    container_name = f"leo-integrated-learning-{os.getpid()}-{uuid.uuid4().hex[:10]}"
+    postgres = DisposablePostgres(container_name, secrets.token_urlsafe(32))
+    receipt: dict[str, Any] = {
+        "schema": RECEIPT_SCHEMA,
+        "status": "fail",
+        "required_tier": "realistic_isolated_local_container_end_to_end",
+        "current_tier": "not_proven",
+        "source_packet": str(source_packet),
+        "review_action": review_action,
+        "safety_boundary": {
+            "self_created_container_only": True,
+            "existing_database_target_argument_supported": False,
+            "vps_or_gcp_target_supported": False,
+            "telegram_send_performed": False,
+        },
+        "checks": {},
+    }
+
+    try:
+        compiled, expected, paths = compile_source_packet(source_packet)
+        child = compiled["strict_child_proposal"]
+        parent = compiled["parent_proposal"]
+        proposal_id = str(child["id"])
+        apply_payload = child["payload"]["apply_payload"]
+        manifest = json.loads(paths["manifest"].read_text(encoding="utf-8"))
+        receipt["source"] = {
+            "source_identity": manifest["source"]["identity"],
+            "source_locator": manifest["source"]["locator"],
+            "artifact_sha256": sha256_file(paths["artifact"]),
+            "manifest_canonical_sha256": canonical_sha256(manifest),
+            "retained_proposal_packet_canonical_sha256": canonical_sha256(expected),
+        }
+        receipt["compiled"] = {
+            "status": compiled["status"],
+            "bundle_sha256": canonical_sha256(compiled),
+            "parent_proposal_id": parent["id"],
+            "strict_proposal_id": proposal_id,
+            "candidate_counts": {
+                family: len(apply_payload.get(family, []))
+                for family in ("claims", "sources", "evidence", "edges", "reasoning_tools")
+            },
+        }
+        compiled_identities = exact_identities(child)
+        receipt["identities"] = compiled_identities
+
+        environment = postgres.start()
+        receipt["environment"] = environment
+        review_password = secrets.token_urlsafe(32)
+        apply_password = secrets.token_urlsafe(32)
+        review_secrets = temp_root / "kb-review.env"
+        apply_secrets = temp_root / "kb-apply.env"
+        preflight_path = temp_root / "fresh-preflight.json"
+        write_secret_file(review_secrets, "KB_REVIEW_PASSWORD", review_password)
+        write_secret_file(apply_secrets, "KB_APPLY_PASSWORD", apply_password)
+
+        postgres.psql(bootstrap_sql(review_password, apply_password))
+        postgres.psql((HERE / "kb_apply_prereqs.sql").read_text(encoding="utf-8"))
+
+        initial = database_state(postgres, child)
+        stage_result = postgres.psql_json(stage.build_stage_sql(child))
+        pending = proposal_readback(postgres, proposal_id)
+        if pending is None:
+            raise LifecycleError("staging did not persist the exact proposal")
+        staged = database_state(postgres, child)
+        packet = review_packet(pending)
+        receipt["phases"] = {
+            "initial": initial,
+            "staged_pending_review": {
+                "database_fingerprint": staged["database_fingerprint"],
+                "canonical_fingerprint": staged["canonical_fingerprint"],
+                "proposal": pending,
+                "stage_result": stage_result,
+                "review_packet": packet,
+            },
+        }
+
+        approve_base = [
+            *_tool_base(
+                HERE / "approve_proposal.py",
+                proposal_id,
+                review_secrets,
+                "kb_review",
+                container_name,
+            ),
+            "--reviewed-by",
+            reviewed_by,
+            "--review-note",
+            review_note,
+        ]
+        approve_dry_run = run_tool([*approve_base, "--dry-run"], action="approval dry-run")
+        if "kb_stage.approve_strict_proposal" not in approve_dry_run.stdout:
+            raise LifecycleError("approval dry-run did not bind the guarded approval function")
+        approve_execute = run_tool(approve_base, action="explicit review approval")
+        approved = proposal_readback(postgres, proposal_id)
+        immutable_approval = approval_readback(postgres, proposal_id)
+        if approved is None or immutable_approval is None:
+            raise LifecycleError("explicit review did not create both approved state and immutable approval")
+        approved_state = database_state(postgres, child)
+        receipt["phases"]["approved_unapplied"] = {
+            "database_fingerprint": approved_state["database_fingerprint"],
+            "canonical_fingerprint": approved_state["canonical_fingerprint"],
+            "proposal": approved,
+            "immutable_approval": immutable_approval,
+            "review_action": {
+                "reviewed_by": reviewed_by,
+                "review_note": review_note,
+                "dry_run": command_receipt(
+                    approve_dry_run,
+                    expected_marker="kb_stage.approve_strict_proposal",
+                ),
+                "execute": command_receipt(approve_execute),
+            },
+        }
+
+        target_rows_before = target_rows_readback(postgres, child)
+        preflight = apply_lifecycle.build_fresh_preflight(approved, target_rows_before)
+        preflight_path.write_text(json.dumps(preflight, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+        preflight_path.chmod(0o600)
+        apply_base = [
+            *_tool_base(
+                HERE / "apply_proposal.py",
+                proposal_id,
+                apply_secrets,
+                "kb_apply",
+                container_name,
+            ),
+            "--fresh-preflight",
+            str(preflight_path),
+            "--receipt-out",
+            str(private_apply_receipt_path),
+        ]
+        apply_dry_run = run_tool([*apply_base, "--dry-run"], action="guarded apply dry-run")
+        apply_markers = (
+            "kb_stage.assert_approved_proposal",
+            "kb_stage.finish_approved_proposal",
+        )
+        if any(marker not in apply_dry_run.stdout for marker in apply_markers):
+            raise LifecycleError("apply dry-run did not bind both approval guards")
+        apply_execute = run_tool(apply_base, action="guarded canonical apply")
+        if not private_apply_receipt_path.is_file():
+            raise LifecycleError("guarded apply did not retain its private replay receipt")
+        apply_receipt = json.loads(private_apply_receipt_path.read_text(encoding="utf-8"))
+        applied = proposal_readback(postgres, proposal_id)
+        if applied is None:
+            raise LifecycleError("applied proposal readback is missing")
+        final_rows = target_rows_readback(postgres, child)
+        applied_state = database_state(postgres, child)
+
+        recompiled, _expected_again, _paths_again = compile_source_packet(source_packet)
+        recompiled_child = recompiled["strict_child_proposal"]
+        final_identities = exact_identities(recompiled_child, final_rows)
+        receipt["identities"] = final_identities
+        receipt["phases"]["applied"] = {
+            "database_fingerprint": applied_state["database_fingerprint"],
+            "canonical_fingerprint": applied_state["canonical_fingerprint"],
+            "proposal": applied,
+            "canonical_rows": final_rows,
+            "apply_action": {
+                "dry_run": {
+                    **command_receipt(apply_dry_run),
+                    "approval_guards_present": all(marker in apply_dry_run.stdout for marker in apply_markers),
+                },
+                "execute": command_receipt(apply_execute),
+            },
+            "apply_receipt": apply_receipt,
+        }
+        receipt["phases"]["recompiled_readback"] = {
+            "bundle_sha256": canonical_sha256(recompiled),
+            "matches_initial_compile": canonical_json_bytes(recompiled) == canonical_json_bytes(compiled),
+            "proposal_id_matches": recompiled_child["id"] == proposal_id,
+            "claim_ids_match": final_identities["claim_ids"] == compiled_identities["claim_ids"],
+            "source_ids_match": final_identities["source_ids"] == compiled_identities["source_ids"],
+            "canonical_rows_sha256": canonical_sha256(normalized_canonical_rows(final_rows)),
+            "apply_receipt_rows_sha256": canonical_sha256(
+                normalized_canonical_rows(apply_receipt.get("canonical_rows"))
+            ),
+        }
+
+        expected_counts = ap.replay_receipt.expected_row_counts(applied)
+        actual_counts = _row_counts(final_rows)
+        checks = {
+            "fresh_compile_matches_retained_packet": canonical_json_bytes(compiled) == canonical_json_bytes(expected),
+            "compiler_emits_pending_review": compiled.get("status") == "pending_review",
+            "initial_target_rows_absent": _all_target_rows_empty(initial["state"]["canonical_rows"]),
+            "staged_proposal_exact": pending.get("id") == proposal_id and pending.get("payload") == child["payload"],
+            "staged_status_pending_review": pending.get("status") == "pending_review",
+            "review_packet_requires_human_review": packet.get("review_state") == "needs_human_review",
+            "stage_did_not_change_canonical_rows": staged["canonical_fingerprint"] == initial["canonical_fingerprint"],
+            "explicit_review_status_approved": approved.get("status") == "approved",
+            "approved_is_not_applied": approved.get("applied_at") is None,
+            "immutable_approval_matches_payload": immutable_approval.get("payload") == child["payload"],
+            "immutable_approval_matches_reviewer": immutable_approval.get("reviewed_by_handle") == reviewed_by,
+            "review_did_not_change_canonical_rows": approved_state["canonical_fingerprint"]
+            == initial["canonical_fingerprint"],
+            "apply_status_applied": applied.get("status") == "applied" and bool(applied.get("applied_at")),
+            "apply_changed_canonical_fingerprint": applied_state["canonical_fingerprint"]
+            != initial["canonical_fingerprint"],
+            "apply_row_counts_exact": actual_counts == expected_counts,
+            "apply_receipt_replay_ready": apply_receipt.get("replay_ready") is True,
+            "apply_receipt_rows_exact": normalized_canonical_rows(apply_receipt.get("canonical_rows"))
+            == normalized_canonical_rows(final_rows),
+            "recompile_matches_initial": canonical_json_bytes(recompiled) == canonical_json_bytes(compiled),
+            "recompile_proposal_id_exact": recompiled_child["id"] == proposal_id,
+            "database_fingerprints_distinguish_all_transitions": len(
+                {
+                    initial["database_fingerprint"],
+                    staged["database_fingerprint"],
+                    approved_state["database_fingerprint"],
+                    applied_state["database_fingerprint"],
+                }
+            )
+            == 4,
+        }
+        receipt["checks"] = checks
+        if not all(checks.values()):
+            failed = [name for name, passed in checks.items() if not passed]
+            raise LifecycleError(f"lifecycle checks failed: {failed}")
+    except (LifecycleError, OSError, ValueError, KeyError, TypeError, json.JSONDecodeError) as exc:
+        receipt["error"] = {"type": type(exc).__name__, "message": str(exc)}
+    finally:
+        cleanup = postgres.cleanup()
+        shutil.rmtree(temp_root, ignore_errors=True)
+        cleanup["temporary_secret_root_absent"] = not temp_root.exists()
+        cleanup["private_apply_receipt_removed_with_secret_root"] = not private_apply_receipt_path.exists()
+        cleanup["all"] = all(
+            (
+                cleanup.get("container_absent") is True,
+                cleanup.get("name_readback_clear") is True,
+                cleanup.get("instance_label_readback_clear") is True,
+                cleanup.get("network_mode_was_none") is True,
+                cleanup.get("tmpfs_data_dir_was_present") is True,
+                cleanup.get("docker_volume_mounts_observed") == [],
+                cleanup.get("temporary_secret_root_absent") is True,
+                cleanup.get("private_apply_receipt_removed_with_secret_root") is True,
+            )
+        )
+        receipt["cleanup"] = cleanup
+        receipt["checks"]["container_and_private_temp_cleanup_proven"] = cleanup["all"]
+        receipt["status"] = (
+            "pass" if "error" not in receipt and receipt["checks"] and all(receipt["checks"].values()) else "fail"
+        )
+        receipt["current_tier"] = (
+            "realistic_isolated_local_container_end_to_end"
+            if receipt["status"] == "pass"
+            else "runtime_verification_failed"
+        )
+        receipt["strongest_claim"] = (
+            "A retained real source packet was freshly compiled, staged, explicitly reviewed, guarded-applied "
+            "into networkless disposable Postgres, deterministically recompiled, and read back with exact rows; "
+            "the container and private temporary files were then independently absent."
+            if receipt["status"] == "pass"
+            else "The integrated-learning lifecycle did not pass all fail-closed checks."
+        )
+        receipt["not_proven"] = [
+            "VPS or GCP canonical apply",
+            "live Telegram send or reply",
+            "human acceptance of these candidate claims for production",
+        ]
+        output.parent.mkdir(parents=True, exist_ok=True)
+        temporary_output = output.with_name(f".{output.name}.{uuid.uuid4().hex}.tmp")
+        temporary_output.write_text(json.dumps(receipt, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+        os.replace(temporary_output, output)
+    return receipt
+
+
+def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--source-packet", required=True, type=Path)
+    parser.add_argument("--output", required=True, type=Path)
+    parser.add_argument(
+        "--review-action",
+        required=True,
+        choices=("approve",),
+        help="explicit disposable-only review action; no default is accepted",
+    )
+    parser.add_argument("--reviewed-by", default=REVIEWER_HANDLE)
+    parser.add_argument("--review-note", default=DEFAULT_REVIEW_NOTE)
+    return parser.parse_args(argv)
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = parse_args(argv)
+    receipt = run_lifecycle(
+        args.source_packet,
+        args.output,
+        review_action=args.review_action,
+        reviewed_by=args.reviewed_by,
+        review_note=args.review_note,
+    )
+    print(
+        json.dumps(
+            {
+                "status": receipt["status"],
+                "receipt": str(args.output.resolve()),
+                "proposal_id": receipt.get("identities", {}).get("proposal_id"),
+                "current_tier": receipt["current_tier"],
+                "cleanup": receipt["cleanup"].get("all"),
+            },
+            sort_keys=True,
+        )
+    )
+    return 0 if receipt["status"] == "pass" else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/tests/test_run_leo_integrated_learning_lifecycle.py b/tests/test_run_leo_integrated_learning_lifecycle.py
new file mode 100644
index 0000000..ffee6de
--- /dev/null
+++ b/tests/test_run_leo_integrated_learning_lifecycle.py
@@ -0,0 +1,133 @@
+from __future__ import annotations
+
+import json
+import shutil
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(REPO_ROOT / "scripts"))
+
+import compile_kb_source_packet as compiler  # noqa: E402
+import run_leo_integrated_learning_lifecycle as lifecycle  # noqa: E402
+
+
+def _write_source_packet(tmp_path: Path) -> Path:
+    packet = tmp_path / "source-packet"
+    prepared = packet / "prepared"
+    prepared.mkdir(parents=True)
+    artifact = REPO_ROOT / "fixtures" / "working-leo" / "document-ingestion-v1.json"
+    manifest = REPO_ROOT / "fixtures" / "working-leo" / "source-compiler-manifest-v1.json"
+    extracted = prepared / "extracted-text.txt"
+    manifest_target = prepared / "source-manifest.json"
+    shutil.copyfile(artifact, extracted)
+    shutil.copyfile(manifest, manifest_target)
+    expected = compiler.compile_source_packet(extracted, extracted, manifest_target)
+    (packet / "proposal-packet.json").write_text(
+        json.dumps(expected, indent=2, sort_keys=True) + "\n",
+        encoding="utf-8",
+    )
+    return packet
+
+
+def _docker_available() -> bool:
+    if not shutil.which("docker"):
+        return False
+    result = subprocess.run(
+        ["docker", "info", "--format", "{{.ServerVersion}}"],
+        text=True,
+        capture_output=True,
+        check=False,
+    )
+    return result.returncode == 0
+
+
+def test_source_packet_recompiles_to_retained_expectation(tmp_path: Path) -> None:
+    packet = _write_source_packet(tmp_path)
+
+    compiled, expected, paths = lifecycle.compile_source_packet(packet)
+
+    assert compiled == expected
+    assert lifecycle.sha256_file(paths["artifact"]) == expected["validated_manifest"]["artifact_sha256"]
+    assert compiled["status"] == "pending_review"
+    assert compiled["strict_child_proposal"]["status"] == "pending_review"
+
+
+def test_source_packet_refuses_retained_packet_drift(tmp_path: Path) -> None:
+    packet = _write_source_packet(tmp_path)
+    expected_path = packet / "proposal-packet.json"
+    expected = json.loads(expected_path.read_text(encoding="utf-8"))
+    expected["strict_child_proposal"]["status"] = "approved"
+    expected_path.write_text(json.dumps(expected), encoding="utf-8")
+
+    with pytest.raises(lifecycle.LifecycleError, match="does not match retained"):
+        lifecycle.compile_source_packet(packet)
+
+
+def test_exact_identities_adds_observed_evidence_ids(tmp_path: Path) -> None:
+    packet = _write_source_packet(tmp_path)
+    compiled, _expected, _paths = lifecycle.compile_source_packet(packet)
+    child = compiled["strict_child_proposal"]
+    apply_payload = child["payload"]["apply_payload"]
+    planned = apply_payload["evidence"][0]
+    evidence_id = "eeeeeeee-eeee-4eee-8eee-eeeeeeeeeeee"
+
+    identities = lifecycle.exact_identities(
+        child,
+        {
+            "claim_evidence": [{"id": evidence_id, **planned}],
+            "claim_edges": [],
+        },
+    )
+
+    assert identities["proposal_id"] == child["id"]
+    assert identities["claim_ids"] == [row["id"] for row in apply_payload["claims"]]
+    assert identities["source_ids"] == [row["id"] for row in apply_payload["sources"]]
+    assert identities["planned_evidence_keys"] == [
+        {
+            "claim_id": row["claim_id"],
+            "source_id": row["source_id"],
+            "role": row["role"],
+        }
+        for row in apply_payload["evidence"]
+    ]
+    assert identities["evidence_ids"] == [evidence_id]
+
+
+def test_lifecycle_requires_explicit_approve_action(tmp_path: Path) -> None:
+    with pytest.raises(lifecycle.LifecycleError, match="only supported explicit review action"):
+        lifecycle.run_lifecycle(
+            tmp_path / "unused",
+            tmp_path / "receipt.json",
+            review_action="inspect",
+        )
+
+
+@pytest.mark.skipif(not _docker_available(), reason="Docker daemon is required for the isolated lifecycle")
+def test_disposable_postgres_lifecycle_end_to_end(tmp_path: Path) -> None:
+    packet = _write_source_packet(tmp_path)
+    output = tmp_path / "integrated-learning-receipt.json"
+
+    receipt = lifecycle.run_lifecycle(
+        packet,
+        output,
+        review_action="approve",
+        review_note="Approved exact fixture payload for isolated lifecycle regression coverage.",
+    )
+
+    assert receipt["status"] == "pass", receipt.get("error")
+    assert receipt["current_tier"] == "realistic_isolated_local_container_end_to_end"
+    assert receipt["phases"]["staged_pending_review"]["proposal"]["status"] == "pending_review"
+    assert receipt["phases"]["approved_unapplied"]["proposal"]["status"] == "approved"
+    assert receipt["phases"]["approved_unapplied"]["proposal"]["applied_at"] is None
+    assert receipt["phases"]["applied"]["proposal"]["status"] == "applied"
+    assert receipt["phases"]["applied"]["apply_receipt"]["replay_ready"] is True
+    assert receipt["phases"]["recompiled_readback"]["matches_initial_compile"] is True
+    assert receipt["identities"]["evidence_ids"]
+    assert len(receipt["checks"]) >= 20
+    assert all(receipt["checks"].values())
+    assert receipt["cleanup"]["all"] is True
+    assert json.loads(output.read_text(encoding="utf-8")) == receipt