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

Add deterministic genesis plus ledger rebuild

Browse source
This commit is contained in:
twentyOne2x 2026-07-15 00:33:29 +02:00
parent 4088c9c8f5
commit 1a6b7d51a4
3 changed files with 2205 additions and 7 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

95
docs/kb-rebuild-and-recompile.md
View file

@ -141,9 +141,89 @@ Normal applies mark the SQL hash as `exact_executed_sql`. A later
pretends the current engine hash is historical proof of the originally executed pretends the current engine hash is historical proof of the originally executed
program. program.
This closes replay-receipt loss for new strict applies. It does not reconstruct This closes replay-receipt loss for new strict applies. The receipt alone does
legacy freeform applies, and it does not yet execute the complete not retain every column of the proposal ledger, so exact reconstruction also
genesis-plus-ledger blank-database replay. needs the full approved proposal row, immutable approval snapshot, and final
applied proposal row.
## Genesis Plus Strict Ledger: Working Insert-Only Slice
`ops/run_local_genesis_ledger_rebuild.py` now executes the first exact
genesis-plus-ledger slice in one command:
```bash
.venv/bin/python ops/run_local_genesis_ledger_rebuild.py \
--genesis-dump /private/path/genesis.dump \
--genesis-manifest /private/path/genesis-manifest.jsonl \
--ledger /private/path/ledger.json \
--ledger-sha256 "$LEDGER_SHA256" \
--output /tmp/genesis-ledger-rebuild-receipt.json
```
The v1 ledger pins the genesis dump and manifest, final parity manifest,
reconstruction/restore/guard/apply/replay/parity engines, and every ordered
private material file. Each material file contains one existing `kb_apply_replay_receipt`, the
exact full proposal row immediately before apply, its immutable
`kb_proposal_approvals` row, and the exact full proposal row after apply. These
files can contain claim text or source excerpts and must remain private.
The ledger shape is:
```json
{
"artifact": "teleo_genesis_plus_ledger",
"contract_version": 1,
"engine": {
"reconstruction_command_sha256": "<sha256>",
"base_rebuild_engine_sha256": "<sha256>",
"apply_engine_sha256": "<sha256>",
"replay_receipt_engine_sha256": "<sha256>",
"guard_prerequisites_sha256": "<sha256>",
"parity_sql_sha256": "<sha256>"
},
"genesis": {
"dump_sha256": "<sha256>",
"parity_manifest_sha256": "<sha256>"
},
"entries": [{
"sequence": 1,
"material": "private/0001.json",
"sha256": "<material-file-sha256>",
"replay_material_sha256": "<receipt-replay-material-sha256>"
}],
"final_parity": {
"manifest": "final-manifest.jsonl",
"sha256": "<sha256>"
}
}
```
Each referenced material object has exact top-level keys
`artifact`, `contract_version`, `sequence`, `approved_proposal`,
`approval_snapshot`, `applied_proposal`, and `replay_receipt`. Proposal objects
must contain every current `kb_stage.kb_proposals` column; partial envelopes
are rejected.
The command verifies every hash before starting Docker, restores a tmpfs-backed
Postgres with `--network none`, reapplies the current guarded prerequisites,
and proves genesis parity. For each entry it seeds the receipt's exact row IDs
and timestamps in the disposable clone, executes the existing `kb_apply`
payload-bound guard and ledger transition, checks exact proposal and canonical
row readbacks, then verifies the complete final parity manifest. Its public
mode-`0600` receipt contains hashes, IDs, types, counts, parity summaries, and
cleanup proof, but no payloads, rows, SQL, source paths, or command errors.
The exact v1 claim ceiling is intentionally narrow:
- `add_edge`, `attach_evidence`, and `approve_claim` strict receipts execute;
- sequence gaps, hash drift, engine drift, duplicate proposals, and legacy or
freeform payloads fail before container start;
- `revise_strategy` fails closed because its current receipt omits the prior
strategies and nodes that the apply engine deactivates or retires;
- full proposal before/after rows are mandatory because the current receipt
envelope does not retain proposal origin fields or exact `updated_at`;
- this proves only an isolated local reconstruction. It does not touch or prove
VPS, GCP, Telegram, a live database, or blank-schema source recompilation.
The source compiler now turns one raw artifact, its strict UTF-8 extraction, The source compiler now turns one raw artifact, its strict UTF-8 extraction,
and a reviewed extraction manifest into a deterministic, hash-bound and a reviewed extraction manifest into a deterministic, hash-bound
@ -190,10 +270,11 @@ neither command applies canonical rows.
The existing full-data clone canary separately proves that a reviewed packet The existing full-data clone canary separately proves that a reviewed packet
can create source, claim, evidence, and edge rows and affect later reasoning. can create source, claim, evidence, and edge rows and affect later reasoning.
The remaining recompilation capability is to join genesis, accepted packets, The remaining reconstruction work is to retain complete deltas for mutating
and their row-level receipts into a complete corpus runner. One-document contracts such as `revise_strategy`, backfill or explicitly reject legacy
preparation, staging, and receipt capture do not yet prove a clean database can freeform applies, and extend beyond genesis recovery to a blank-schema source
be rebuilt semantically from every source. compiler. The insert-only ledger runner does not prove that every historical
canonical row can be rebuilt semantically from retained sources.
## Definition Of Working ## Definition Of Working

1284
ops/run_local_genesis_ledger_rebuild.py Normal file
View file

File diff suppressed because it is too large Load diff

833
tests/test_run_local_genesis_ledger_rebuild.py Normal file
View file

@ -0,0 +1,833 @@
from __future__ import annotations
import argparse
import copy
import hashlib
import json
import shutil
import stat
import subprocess
import sys
import time
import uuid
from collections.abc import Iterator
from pathlib import Path
import pytest
from ops import run_local_genesis_ledger_rebuild as rebuild
CLAIM_A = "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa"
CLAIM_B = "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb"
REVIEWER_ID = "11111111-1111-4111-8111-111111111111"
SERVICE_ID = "44444444-4444-4444-4444-444444444444"
PROPOSAL_ID = "99999999-9999-4999-8999-999999999999"
EDGE_ID = "eeeeeeee-eeee-4eee-8eee-eeeeeeeeeeee"
PRIVATE_MARKER = "PRIVATE-REPLAY-MATERIAL-MUST-NOT-LEAK"
def sha256_file(path: Path) -> str:
digest = hashlib.sha256()
digest.update(path.read_bytes())
return digest.hexdigest()
def write_json(path: Path, value: object) -> Path:
path.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8")
return path
def manifest_rows(database: str = "teleo_fixture") -> list[dict]:
role = {
"name": "kb_apply",
"can_login": True,
"superuser": False,
"create_db": False,
"create_role": False,
"inherit": False,
"replication": False,
"bypass_rls": False,
}
rows = [
{
"kind": "identity",
"database": database,
"server_version_num": 160010,
"transaction_read_only": "on",
"server_address": None,
"server_port": None,
"ssl": False,
"ssl_version": None,
"database_collation": "C",
"database_ctype": "C",
},
{"kind": "schemas", "items": ["kb_stage", "public"]},
{"kind": "extensions", "items": []},
{"kind": "application_roles", "items": [role]},
]
rows.extend(
{"kind": kind, "items": []}
for kind in (
"columns",
"constraints",
"indexes",
"sequences",
"views",
"functions",
"triggers",
"types",
"policies",
)
)
rows.extend(
[
{
"kind": "table",
"schema": "public",
"table": "claims",
"row_count": 2,
"rowset_md5": "fixture",
},
{
"kind": "performance",
"query": "count_claims",
"elapsed_ms": 1.0,
"observed_rows": 2,
},
]
)
return rows
def write_manifest(path: Path) -> Path:
path.write_text("".join(json.dumps(row) + "\n" for row in manifest_rows()), encoding="utf-8")
return path
def proposal_rows() -> tuple[dict, dict, dict]:
payload = {
"apply_payload": {
"from_claim": CLAIM_A,
"to_claim": CLAIM_B,
"edge_type": "supports",
"weight": 0.8,
},
"private_title": PRIVATE_MARKER,
}
approved = {
"id": PROPOSAL_ID,
"proposal_type": "add_edge",
"status": "approved",
"proposed_by_handle": "fixture-agent",
"proposed_by_agent_id": REVIEWER_ID,
"channel": "reconstruction_fixture",
"source_ref": "private://fixture/source",
"rationale": PRIVATE_MARKER,
"payload": payload,
"reviewed_by_handle": "m3ta",
"reviewed_by_agent_id": REVIEWER_ID,
"reviewed_at": "2026-07-14T01:00:00+00:00",
"review_note": "Reviewed exact strict fixture payload.",
"applied_by_handle": None,
"applied_by_agent_id": None,
"applied_at": None,
"created_at": "2026-07-14T00:59:00+00:00",
"updated_at": "2026-07-14T01:00:00+00:00",
}
applied = {
**approved,
"status": "applied",
"applied_by_handle": "kb-apply",
"applied_by_agent_id": SERVICE_ID,
"applied_at": "2026-07-14T01:01:00+00:00",
"updated_at": "2026-07-14T01:01:00.000001+00:00",
}
approval = {
"proposal_id": PROPOSAL_ID,
"proposal_type": "add_edge",
"payload": payload,
"reviewed_by_handle": "m3ta",
"reviewed_by_agent_id": REVIEWER_ID,
"reviewed_by_db_role": "kb_review",
"reviewed_at": "2026-07-14T01:00:00+00:00",
"review_note": "Reviewed exact strict fixture payload.",
}
return approved, applied, approval
def applied_envelope(applied: dict) -> dict:
fields = (
"id",
"proposal_type",
"status",
"payload",
"reviewed_by_handle",
"reviewed_by_agent_id",
"reviewed_at",
"review_note",
"applied_by_handle",
"applied_by_agent_id",
"applied_at",
)
return {field: applied[field] for field in fields}
def replay_material() -> dict:
approved, applied, approval = proposal_rows()
proposal = applied_envelope(applied)
edge = {
"id": EDGE_ID,
"from_claim": CLAIM_A,
"to_claim": CLAIM_B,
"edge_type": "supports",
"weight": 0.8,
"created_by": None,
"created_at": "2026-07-14T01:00:30+00:00",
}
receipt = rebuild.replay_receipt.build_replay_receipt(
proposal,
{"claim_edges": [edge]},
apply_sql_sha256=rebuild.replay_receipt.sha256_text("fixture original guarded apply SQL"),
apply_sql_source="exact_executed_sql",
exported_at_utc="2026-07-14T01:02:00+00:00",
)
return {
"artifact": rebuild.MATERIAL_ARTIFACT,
"contract_version": rebuild.MATERIAL_CONTRACT_VERSION,
"sequence": 1,
"approved_proposal": approved,
"approval_snapshot": approval,
"applied_proposal": applied,
"replay_receipt": receipt,
}
def write_bundle(tmp_path: Path, *, material: dict | None = None) -> argparse.Namespace:
dump = tmp_path / "genesis.dump"
dump.write_bytes(b"PGDMPfixture")
genesis_manifest = write_manifest(tmp_path / "genesis-manifest.jsonl")
final_manifest = write_manifest(tmp_path / "final-manifest.jsonl")
material = copy.deepcopy(material or replay_material())
material_path = write_json(tmp_path / "entry-0001.json", material)
receipt_hash = material["replay_receipt"]["hashes"]["replay_material_sha256"]
ledger = {
"artifact": rebuild.LEDGER_ARTIFACT,
"contract_version": rebuild.LEDGER_CONTRACT_VERSION,
"engine": rebuild._engine_hashes(),
"genesis": {
"dump_sha256": sha256_file(dump),
"parity_manifest_sha256": sha256_file(genesis_manifest),
},
"entries": [
{
"sequence": 1,
"material": material_path.name,
"sha256": sha256_file(material_path),
"replay_material_sha256": receipt_hash,
}
],
"final_parity": {
"manifest": final_manifest.name,
"sha256": sha256_file(final_manifest),
},
}
ledger_path = write_json(tmp_path / "ledger.json", ledger)
return argparse.Namespace(
genesis_dump=dump,
genesis_manifest=genesis_manifest,
ledger=ledger_path,
ledger_sha256=sha256_file(ledger_path),
database="teleo",
image=rebuild.canonical_rebuild.DEFAULT_IMAGE,
container_prefix="teleo-genesis-ledger-test",
tmpfs_mb=256,
startup_timeout=30.0,
restore_timeout=60.0,
apply_timeout=60.0,
manifest_timeout=60.0,
command_timeout=20.0,
poll_interval=0.05,
max_target_query_ms=5000.0,
max_target_source_ratio=100.0,
docker_bin="docker",
output=tmp_path / "receipt.json",
)
def rewrite_material_and_ledger(args: argparse.Namespace, material: dict) -> None:
material_path = args.ledger.parent / "entry-0001.json"
write_json(material_path, material)
ledger = json.loads(args.ledger.read_text(encoding="utf-8"))
ledger["entries"][0]["sha256"] = sha256_file(material_path)
write_json(args.ledger, ledger)
args.ledger_sha256 = sha256_file(args.ledger)
def test_load_bundle_validates_every_pin_and_current_apply_engine(tmp_path: Path) -> None:
args = write_bundle(tmp_path)
bundle = rebuild.load_bundle(args)
assert bundle.genesis_dump_sha256 == sha256_file(args.genesis_dump)
assert bundle.genesis_manifest_sha256 == sha256_file(args.genesis_manifest)
assert bundle.final_manifest_sha256 == sha256_file(tmp_path / "final-manifest.jsonl")
assert bundle.engine_hashes == rebuild._engine_hashes()
assert len(bundle.entries) == 1
entry = bundle.entries[0]
assert entry.applied_proposal["proposal_type"] == "add_edge"
assert entry.current_apply_sql_sha256 == hashlib.sha256(entry.current_apply_sql.encode("utf-8")).hexdigest()
assert "kb_stage.assert_approved_proposal" in entry.current_apply_sql
def test_dump_hash_mismatch_fails_before_container_start(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
args = write_bundle(tmp_path)
args.genesis_dump.write_bytes(b"PGDMPtampered")
commands: list[list[str]] = []
def absent_docker(command: list[str], *, timeout: float):
commands.append(command)
if command[1:3] == ["rm", "--force"]:
return subprocess.CompletedProcess(command, 1, "", "No such container")
if command[1:3] == ["container", "inspect"]:
return subprocess.CompletedProcess(command, 1, "", "No such object")
raise AssertionError(f"unexpected command after failed preflight: {command}")
monkeypatch.setattr(rebuild.canonical_rebuild, "_run", absent_docker)
receipt = rebuild.run_reconstruction(args)
assert receipt["status"] == "fail"
assert receipt["error"]["code"] == "hash_pin_mismatch"
assert receipt["cleanup"]["container_absent"] is True
assert not any(len(command) > 1 and command[1] == "run" for command in commands)
def test_revise_strategy_receipt_fails_closed_before_receipt_rows_are_used(tmp_path: Path) -> None:
material = replay_material()
material["replay_receipt"]["proposal"]["proposal_type"] = "revise_strategy"
material["applied_proposal"]["proposal_type"] = "revise_strategy"
material["approved_proposal"]["proposal_type"] = "revise_strategy"
material["approval_snapshot"]["proposal_type"] = "revise_strategy"
args = write_bundle(tmp_path, material=material)
with pytest.raises(rebuild.ReconstructionError, match="prior strategy") as exc_info:
rebuild.load_bundle(args)
assert exc_info.value.code == "unsupported_mutating_receipt"
def test_legacy_freeform_receipt_fails_closed(tmp_path: Path) -> None:
material = replay_material()
material["replay_receipt"]["proposal"]["payload"] = {"legacy": PRIVATE_MARKER}
material["approved_proposal"]["payload"] = {"legacy": PRIVATE_MARKER}
material["applied_proposal"]["payload"] = {"legacy": PRIVATE_MARKER}
material["approval_snapshot"]["payload"] = {"legacy": PRIVATE_MARKER}
args = write_bundle(tmp_path, material=material)
with pytest.raises(rebuild.ReconstructionError) as exc_info:
rebuild.load_bundle(args)
assert exc_info.value.code == "invalid_replay_receipt"
assert PRIVATE_MARKER not in exc_info.value.safe_message
def test_proposal_metadata_outside_apply_transition_is_rejected(tmp_path: Path) -> None:
material = replay_material()
material["applied_proposal"]["rationale"] = "silently changed after review"
args = write_bundle(tmp_path)
rewrite_material_and_ledger(args, material)
with pytest.raises(rebuild.ReconstructionError) as exc_info:
rebuild.load_bundle(args)
assert exc_info.value.code == "material_mismatch"
assert "rationale" in exc_info.value.safe_message
def test_output_cannot_overwrite_a_fixed_guard_or_engine_file(tmp_path: Path) -> None:
args = write_bundle(tmp_path)
args.output = rebuild.GUARD_PREREQUISITES_PATH
with pytest.raises(rebuild.ReconstructionError) as exc_info:
rebuild.load_bundle(args)
assert exc_info.value.code == "unsafe_output_path"
def test_unknown_private_field_name_is_not_echoed_in_public_error(tmp_path: Path) -> None:
material = replay_material()
material[PRIVATE_MARKER] = "private value"
args = write_bundle(tmp_path, material=material)
with pytest.raises(rebuild.ReconstructionError) as exc_info:
rebuild.load_bundle(args)
assert exc_info.value.code == "invalid_contract"
assert PRIVATE_MARKER not in exc_info.value.safe_message
def test_seed_sql_preserves_exact_rows_and_uses_only_fixed_tables(tmp_path: Path) -> None:
entry = rebuild.load_bundle(write_bundle(tmp_path)).entries[0]
sql = rebuild.build_seed_sql(entry)
assert "jsonb_populate_record" in sql
assert "kb_stage.kb_proposals" in sql
assert "kb_stage.kb_proposal_approvals" in sql
assert "public.claim_edges" in sql
assert "on conflict do nothing" in sql
assert PRIVATE_MARKER in sql
assert "public.strategies" not in sql
def test_private_psql_failure_output_is_replaced_with_a_fingerprint(
monkeypatch: pytest.MonkeyPatch,
) -> None:
captured: dict[str, object] = {}
def fail(command: list[str], *, stdin: str, timeout: float):
captured.update(command=command, stdin=stdin, timeout=timeout)
return subprocess.CompletedProcess(command, 1, "", f"ERROR: {PRIVATE_MARKER}")
monkeypatch.setattr(rebuild, "_run_stdin", fail)
args = argparse.Namespace(docker_bin="docker", database="teleo")
with pytest.raises(rebuild.ReconstructionError) as exc_info:
rebuild._psql(
args,
"fixture-container",
f"select '{PRIVATE_MARKER}';",
role="postgres",
timeout=3.0,
code="fixture_failure",
action="private fixture SQL",
)
assert PRIVATE_MARKER not in exc_info.value.safe_message
assert "diagnostic_sha256=" in exc_info.value.safe_message
assert PRIVATE_MARKER not in " ".join(captured["command"])
assert PRIVATE_MARKER in captured["stdin"]
def test_entry_summary_is_secret_safe(tmp_path: Path) -> None:
entry = rebuild.load_bundle(write_bundle(tmp_path)).entries[0]
encoded = json.dumps(rebuild._entry_summary(entry), sort_keys=True)
assert PRIVATE_MARKER not in encoded
assert entry.applied_proposal["id"] in encoded
assert entry.replay_material_sha256 in encoded
BOOTSTRAP_SQL = f"""
create role kb_apply login noinherit;
create role kb_review login noinherit;
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,
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()
);
create table public.strategies (
id uuid primary key,
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()
);
create table public.strategy_nodes (
id uuid primary key,
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
('{REVIEWER_ID}', 'm3ta', 'human');
insert into public.claims (id, type, text, status, confidence, tags, created_by) values
('{CLAIM_A}', 'structural', 'Fixture claim A', 'open', 0.8, array['fixture'], '{REVIEWER_ID}'),
('{CLAIM_B}', 'structural', 'Fixture claim B', 'open', 0.8, array['fixture'], '{REVIEWER_ID}');
"""
def docker_psql(
container: str,
database: str,
sql: str,
*,
role: str = "postgres",
check: bool = True,
) -> subprocess.CompletedProcess[str]:
completed = subprocess.run(
[
"docker",
"exec",
"-i",
container,
"psql",
"-X",
"-U",
role,
"-h",
"127.0.0.1",
"-d",
database,
"-Atq",
"-v",
"ON_ERROR_STOP=1",
],
input=sql,
text=True,
capture_output=True,
check=False,
)
if check and completed.returncode != 0:
pytest.fail(
f"fixture psql failed ({completed.returncode})\nstdout:\n{completed.stdout}\nstderr:\n{completed.stderr}"
)
return completed
def wait_for_postgres(container: str, database: str) -> None:
for _ in range(120):
completed = docker_psql(container, database, "select 1;", check=False)
if completed.returncode == 0:
return
time.sleep(0.25)
pytest.fail("fixture PostgreSQL did not become ready")
def capture_manifest(container: str, database: str, destination: Path) -> None:
copy_result = subprocess.run(
["docker", "cp", str(rebuild.PARITY_SQL_PATH), f"{container}:/tmp/parity.sql"],
text=True,
capture_output=True,
check=False,
)
if copy_result.returncode != 0:
pytest.fail(f"could not copy parity SQL: {copy_result.stderr}")
completed = subprocess.run(
[
"docker",
"exec",
container,
"psql",
"-X",
"-U",
"postgres",
"-d",
database,
"-Atq",
"-v",
"ON_ERROR_STOP=1",
"-f",
"/tmp/parity.sql",
],
text=True,
capture_output=True,
check=False,
)
if completed.returncode != 0:
pytest.fail(f"could not capture parity manifest: {completed.stderr}")
destination.write_text(completed.stdout, encoding="utf-8")
@pytest.fixture(scope="module")
def source_postgres() -> Iterator[tuple[str, str]]:
if shutil.which("docker") is None:
pytest.skip("Docker is required for the genesis-plus-ledger lifecycle test")
container = f"genesis-ledger-source-{uuid.uuid4().hex[:12]}"
database = "teleo_fixture"
started = subprocess.run(
[
"docker",
"run",
"--rm",
"--detach",
"--name",
container,
"--env",
f"POSTGRES_DB={database}",
"--env",
"POSTGRES_HOST_AUTH_METHOD=trust",
rebuild.canonical_rebuild.DEFAULT_IMAGE,
],
text=True,
capture_output=True,
check=False,
)
if started.returncode != 0:
pytest.fail(f"could not start fixture PostgreSQL: {started.stderr}")
try:
wait_for_postgres(container, database)
docker_psql(container, database, BOOTSTRAP_SQL)
copy_result = subprocess.run(
[
"docker",
"cp",
str(rebuild.GUARD_PREREQUISITES_PATH),
f"{container}:/tmp/kb-apply-prereqs.sql",
],
text=True,
capture_output=True,
check=False,
)
if copy_result.returncode != 0:
pytest.fail(f"could not copy apply prerequisites: {copy_result.stderr}")
migration = subprocess.run(
[
"docker",
"exec",
container,
"psql",
"-X",
"-U",
"postgres",
"-d",
database,
"-Atq",
"-v",
"ON_ERROR_STOP=1",
"-f",
"/tmp/kb-apply-prereqs.sql",
],
text=True,
capture_output=True,
check=False,
)
if migration.returncode != 0:
pytest.fail(f"could not apply guarded prerequisites: {migration.stderr}")
yield container, database
finally:
subprocess.run(
["docker", "rm", "--force", container],
text=True,
capture_output=True,
check=False,
)
def test_live_genesis_plus_ledger_command_rebuilds_exactly_and_proves_cleanup(
tmp_path: Path,
source_postgres: tuple[str, str],
) -> None:
source_container, database = source_postgres
genesis_dump = tmp_path / "genesis.dump"
dump_result = subprocess.run(
[
"docker",
"exec",
source_container,
"pg_dump",
"-U",
"postgres",
"-d",
database,
"--format=custom",
"--file=/tmp/genesis.dump",
],
text=True,
capture_output=True,
check=False,
)
assert dump_result.returncode == 0, dump_result.stderr
copy_result = subprocess.run(
["docker", "cp", f"{source_container}:/tmp/genesis.dump", str(genesis_dump)],
text=True,
capture_output=True,
check=False,
)
assert copy_result.returncode == 0, copy_result.stderr
genesis_manifest = tmp_path / "genesis-manifest.jsonl"
capture_manifest(source_container, database, genesis_manifest)
material = replay_material()
material_path = write_json(tmp_path / "entry-0001.json", material)
entry = rebuild._validate_entry_material(
material,
expected_sequence=1,
expected_replay_hash=material["replay_receipt"]["hashes"]["replay_material_sha256"],
material_path=material_path,
material_sha256=sha256_file(material_path),
)
docker_psql(source_container, database, rebuild.build_seed_sql(entry))
docker_psql(
source_container,
database,
entry.current_apply_sql,
role="kb_apply",
)
docker_psql(
source_container,
database,
rebuild.build_applied_timestamp_normalization_sql(entry),
)
final_manifest = tmp_path / "final-manifest.jsonl"
capture_manifest(source_container, database, final_manifest)
ledger = {
"artifact": rebuild.LEDGER_ARTIFACT,
"contract_version": rebuild.LEDGER_CONTRACT_VERSION,
"engine": rebuild._engine_hashes(),
"genesis": {
"dump_sha256": sha256_file(genesis_dump),
"parity_manifest_sha256": sha256_file(genesis_manifest),
},
"entries": [
{
"sequence": 1,
"material": material_path.name,
"sha256": sha256_file(material_path),
"replay_material_sha256": entry.replay_material_sha256,
}
],
"final_parity": {
"manifest": final_manifest.name,
"sha256": sha256_file(final_manifest),
},
}
ledger_path = write_json(tmp_path / "ledger.json", ledger)
output = tmp_path / "reconstruction-receipt.json"
command = [
sys.executable,
str(Path(rebuild.__file__).resolve()),
"--genesis-dump",
str(genesis_dump),
"--genesis-manifest",
str(genesis_manifest),
"--ledger",
str(ledger_path),
"--ledger-sha256",
sha256_file(ledger_path),
"--database",
database,
"--container-prefix",
"teleo-genesis-ledger-live-test",
"--tmpfs-mb",
"256",
"--max-target-query-ms",
"5000",
"--max-target-source-ratio",
"100",
"--output",
str(output),
]
completed = subprocess.run(
command,
cwd=rebuild.REPO_ROOT,
text=True,
capture_output=True,
check=False,
timeout=180,
)
assert completed.returncode == 0, completed.stderr + completed.stdout
receipt = json.loads(output.read_text(encoding="utf-8"))
assert receipt["status"] == "pass"
assert receipt["genesis_parity"]["status"] == "pass"
assert receipt["ledger"]["entries"][0]["status"] == "pass"
assert receipt["ledger"]["entries"][0]["guarded_apply_executed"] is True
assert receipt["ledger"]["entries"][0]["canonical_rows_exact"] is True
assert receipt["final_parity"]["status"] == "pass"
assert receipt["cleanup"]["container_absent"] is True
assert receipt["safety"]["network_access_available_to_container"] is False
assert receipt["safety"]["private_replay_material_emitted"] is False
assert PRIVATE_MARKER not in completed.stdout
assert PRIVATE_MARKER not in output.read_text(encoding="utf-8")
assert stat.S_IMODE(output.stat().st_mode) == 0o600
inspect = subprocess.run(
["docker", "container", "inspect", receipt["container"]["name"]],
text=True,
capture_output=True,
check=False,
)
assert inspect.returncode != 0