teleo/teleo-infrastructure
8
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions 1

Merge pull request #174 from living-ip/codex/leo-integrated-learning-loop-w2-20260715

Browse source 
Add isolated Leo integrated-learning lifecycle
This commit is contained in:
twentyOne2x 2026-07-16 12:06:44 +02:00 committed by GitHub
commit c8f6481fc7
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 1033 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

900
scripts/run_leo_integrated_learning_lifecycle.py Normal file
View file

@ -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())

133
tests/test_run_leo_integrated_learning_lifecycle.py Normal file
View file

@ -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