diff --git a/docs/gcp-kb-cloudsql-restore-runbook.md b/docs/gcp-kb-cloudsql-restore-runbook.md index 0a1ad09..2015807 100644 --- a/docs/gcp-kb-cloudsql-restore-runbook.md +++ b/docs/gcp-kb-cloudsql-restore-runbook.md @@ -61,6 +61,24 @@ retains a private custom dump, dump SHA-256, object TOC, catalog/data manifest, and before/after service state. It never restarts Leo or writes to the source database. +Prove that this exact snapshot can rebuild a blank Postgres target before using +it for GCP: + +```bash +.venv/bin/python ops/run_local_canonical_postgres_rebuild.py \ + --dump /teleo-canonical.dump \ + --source-manifest /source-manifest.jsonl \ + --output /tmp/teleo-canonical-rebuild-receipt.json +``` + +The local runner starts a uniquely named `postgres:16-alpine` container with +network mode `none` and tmpfs-only database storage. It waits for an actual +`psql` connection to the named database, restores with `pg_restore +--no-owner --no-privileges --exit-on-error`, compares the full parity manifest, +then removes the container and proves it is absent. A passing local receipt is +the exact-recovery preflight; it is not semantic recompilation from raw source +documents. + Run `ops/postgres_parity_manifest.sql` against the isolated restored target, then compare source and target: @@ -85,7 +103,7 @@ drill into `teleo`, `teleo_kb`, or `teleo_canonical`. Database isolation does not isolate cluster-global roles or extensions, so verify those separately and do not run the Docker-only gate bootstrap against the shared Cloud SQL instance. -After the parity verifier passes, run the no-send Cory composition replay from +After the parity verifier passes, run the no-send operator composition replay from staging compute against that generated database. Only then delete the generated database and uploaded import object and retain cleanup readback. diff --git a/docs/kb-rebuild-and-recompile.md b/docs/kb-rebuild-and-recompile.md new file mode 100644 index 0000000..f4498fe --- /dev/null +++ b/docs/kb-rebuild-and-recompile.md @@ -0,0 +1,124 @@ +# Rebuilding Leo's Knowledge Database + +## Outcome + +Leo should improve by compiling durable source material and reviewed changes +into Postgres. Repeatedly changing prompts or retraining the chat behavior is +not the knowledge system. + +There are two different rebuilds: + +1. **Exact recovery** restores the current canonical database from a verified + snapshot. This is working now. +2. **Semantic recompilation** starts from the retained source corpus and the + reviewed change ledger, then reproduces the canonical rows. This is partly + recoverable but is not yet complete. + +## Exact Recovery: Working + +Run: + +```bash +.venv/bin/python ops/run_local_canonical_postgres_rebuild.py \ + --dump /private/path/teleo-canonical.dump \ + --source-manifest /private/path/source-manifest.jsonl \ + --output /tmp/teleo-canonical-rebuild-receipt.json +``` + +The 2026-07-13 live canary restored a fresh, network-isolated Postgres in 3.81 +seconds and then removed it. The restored target matched the source across all +39 manifest tables and all 52,164 rows, with no schema, data, constraint, role, +or performance mismatch. Key rows included: + +- 1,837 claims; +- 4,145 sources; +- 4,670 claim-evidence links; +- 4,916 claim edges; +- 17 reasoning tools; +- 26 proposals. + +This is the fastest disaster-recovery path. It does not require Leo to +re-extract or relearn the corpus. + +## Source Recompilation: Current Evidence + +Read-only VPS inspection found two retained June import runs. Both point to the +Forgejo-era workspace at `/opt/teleo-eval/workspaces/main` and retained +inventory JSONL files under `/opt/teleo-eval/kb-import/`. + +The database still retains: + +- `kb_stage.import_runs` for the two inventories; +- staged claims, sources, claim-source links, and claim edges; +- `kb_stage.canonical_mappings` from legacy keys to canonical UUIDs; +- mappings for 1,807 of 1,837 canonical claims; +- mappings for all 4,145 canonical sources. + +For the 1,807 mapped claims, current canonical type, text, status, confidence, +tags, and creator match the retained staged rows exactly. Creation timestamps +also follow a recoverable rule: use the legacy timestamp when present and the +mapping timestamp for the eight rows that had no legacy timestamp. + +The simple retained-row joins currently reproduce: + +- 4,254 of 4,670 canonical evidence links; +- 4,878 of 4,916 canonical edge rows can be accounted for by a staged relation; + historical duplicate multiplicity still needs an explicit replay rule. + +The remaining gaps are concrete rather than mysterious: + +- 30 claims were created after or outside the mapped import; +- 284 mapped source rows do not have a matching retained `staged_sources` row + and need the original source-synthesis rule or an explicit genesis record; +- 416 evidence links need source-synthesis or later-change provenance; +- 38 edge rows need later-change receipts or explicit replay records; +- old applied proposal rows do not describe every historical canonical write. + +This proves that most of the initial database came from the retained file-KB +import path. It does not yet prove a clean blank-database recompile. + +## Target Compiler + +The durable rebuild model is: + +```text +immutable source corpus + file hashes +-> deterministic inventory and classification +-> staged claims, sources, evidence links, and edges +-> stable canonical ID mapping +-> review decisions +-> append-only accepted apply payloads and receipts +-> canonical Postgres +-> render/sync/restart +-> answer benchmark +``` + +Use the current verified snapshot as **genesis epoch 1**. Preserve its dump, +manifest, source commit, inventory files, mappings, and aggregate rebuild +receipt. Every accepted change after that epoch must carry a replayable strict +apply payload and row-level postflight receipt. This prevents the historical +gap from growing while the old import rules are reconstructed. + +The existing local ingestion canary already proves one hash-bound document +fixture can become a normalized `pending_review` proposal without touching +canonical rows. The existing full-data clone canary proves a reviewed packet +can create source, claim, evidence, and edge rows and affect later reasoning. +The next capability is to join those into a corpus runner and replay ledger. + +## Definition Of Working + +Semantic recompilation is complete only when all of these pass: + +1. A command creates a blank database from the retained source corpus plus the + reviewed ledger. +2. Schema, constraints, roles, table counts, row hashes, and key query results + match the canonical manifest. +3. Every canonical row traces to a genesis import record or a reviewed apply + receipt. +4. A new document can be hash-captured, extracted into grounded candidates, + deduplicated, staged, reviewed, applied in a disposable clone, and read back. +5. After render/sync and restart, Leo answers the related broad question using + the new rows and cites the source chain. + +Until then, exact snapshot recovery is production recovery; source +recompilation is an active build capability, not a finished claim. diff --git a/docs/reports/leo-working-state-20260709/local-canonical-postgres-rebuild-canary-20260713.md b/docs/reports/leo-working-state-20260709/local-canonical-postgres-rebuild-canary-20260713.md new file mode 100644 index 0000000..a5904a9 --- /dev/null +++ b/docs/reports/leo-working-state-20260709/local-canonical-postgres-rebuild-canary-20260713.md @@ -0,0 +1,60 @@ +# Local Canonical Postgres Rebuild Canary + +Generated UTC: `2026-07-13T08:34:54.016399+00:00` + +Status: **pass** + +## Result + +- Total command time: `3.809726s` +- Named database ready: `0.781077s` +- `pg_restore`: `1.026398s` +- Manifest capture and parity comparison: `1.229484s` +- Cleanup: `0.088110s` +- Dump bytes: `13,462,116` +- Dump SHA-256: `8a217e32bf6c83cb2b629e5ca5f91fc57f759e5010d9c40babe0d13c9a450822` +- Source manifest SHA-256: `d077f859b5f8ece908c61958a90914272ddc6fb9930df3b97d2ad2ac85b8e39a` + +## Parity + +- Manifest tables: `39/39` +- Total rows: `52,164/52,164` +- Table mismatches: `0` +- Column, constraint, function, index, policy, schema, sequence, trigger, + type, and view hash mismatches: `0` +- Application role mismatches: `0` +- Performance threshold failures: `0` + +Key restored rows: + +| Object | Rows | +|---|---:| +| `public.claims` | 1,837 | +| `public.sources` | 4,145 | +| `public.claim_evidence` | 4,670 | +| `public.claim_edges` | 4,916 | +| `public.reasoning_tools` | 17 | +| `kb_stage.kb_proposals` | 26 | + +## Isolation And Cleanup + +- Docker image: `postgres:16-alpine` +- Network mode: `none` +- Postgres data directory: `tmpfs`, no persistent Docker volume +- Named-database readiness used an actual `psql -d teleo` connection +- Production database touched: `false` +- Container absence after cleanup: `true` +- Independent label-filtered orphan check: no containers found + +## Command + +```bash +.venv/bin/python ops/run_local_canonical_postgres_rebuild.py \ + --dump /private/path/teleo-canonical.dump \ + --source-manifest /private/path/source-manifest.jsonl \ + --output /tmp/teleo-canonical-rebuild-receipt.json +``` + +This proves fast exact recovery from the canonical snapshot. It does not claim +that the whole database can yet be semantically recompiled from raw documents +and reviewed change records alone. diff --git a/ops/run_local_canonical_postgres_rebuild.py b/ops/run_local_canonical_postgres_rebuild.py new file mode 100755 index 0000000..a5d3d10 --- /dev/null +++ b/ops/run_local_canonical_postgres_rebuild.py @@ -0,0 +1,819 @@ +#!/usr/bin/env python3 +"""Restore and verify a canonical PostgreSQL dump in an isolated Docker canary.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import re +import signal +import subprocess +import tempfile +import time +import uuid +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +try: + from .verify_postgres_parity_manifest import compare_manifests, load_manifest +except ImportError: # pragma: no cover - direct script execution + from verify_postgres_parity_manifest import compare_manifests, load_manifest + + +DEFAULT_IMAGE = "postgres:16-alpine" +DEFAULT_MANIFEST_SQL = Path(__file__).with_name("postgres_parity_manifest.sql") +DATA_DIR = "/var/lib/postgresql/data" +DUMP_DESTINATION = "/tmp/teleo-canonical.dump" +MANIFEST_DESTINATION = "/tmp/postgres-parity-manifest.sql" +CANARY_LABEL = "com.livingip.teleo.canonical-rebuild-canary" +DATABASE_NAME_RE = re.compile(r"[A-Za-z_][A-Za-z0-9_]{0,62}\Z") +CONTAINER_PREFIX_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,39}\Z") +ALLOWED_APPLICATION_ROLES = frozenset({"kb_apply", "kb_review"}) +ROLE_BOOLEAN_FIELDS = ( + "can_login", + "superuser", + "create_db", + "create_role", + "inherit", + "replication", + "bypass_rls", +) +KEY_TABLES = ( + "public.claims", + "public.sources", + "public.claim_evidence", + "public.claim_edges", + "public.reasoning_tools", + "kb_stage.kb_proposals", +) + + +class TerminationRequested(RuntimeError): + """Raised after SIGINT/SIGTERM so the normal cleanup path still runs.""" + + +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 validate_custom_dump(path: Path) -> None: + if not path.is_file(): + raise ValueError(f"custom dump does not exist or is not a regular file: {path}") + if path.stat().st_size <= 5: + raise ValueError("custom dump is empty or truncated") + with path.open("rb") as handle: + signature = handle.read(5) + if signature != b"PGDMP": + raise ValueError("dump is not PostgreSQL custom format (missing PGDMP signature)") + + +def new_container_name(prefix: str) -> str: + timestamp = datetime.now(UTC).strftime("%Y%m%d%H%M%S") + return f"{prefix}-{timestamp}-{uuid.uuid4().hex[:12]}" + + +def _run( + command: list[str], + *, + timeout: float, +) -> subprocess.CompletedProcess[str]: + return subprocess.run( + command, + capture_output=True, + check=False, + text=True, + timeout=timeout, + ) + + +def _bounded_error(completed: subprocess.CompletedProcess[str]) -> str: + message = (completed.stderr or completed.stdout or "no command output").strip() + return message[-2000:] + + +def _require_success(completed: subprocess.CompletedProcess[str], action: str) -> str: + if completed.returncode != 0: + raise RuntimeError(f"{action} failed (exit {completed.returncode}): {_bounded_error(completed)}") + return completed.stdout + + +def _absence_result(completed: subprocess.CompletedProcess[str]) -> tuple[bool | None, str | None]: + if completed.returncode == 0: + return False, None + message = (completed.stderr or completed.stdout or "").strip() + lowered = message.lower() + if "no such object" in lowered or "no such container" in lowered: + return True, None + return None, message[-2000:] or f"docker inspect exited {completed.returncode}" + + +def inspect_container_absence( + docker_bin: str, + container: str, + *, + timeout: float, +) -> tuple[bool | None, str | None]: + completed = _run([docker_bin, "container", "inspect", container], timeout=timeout) + return _absence_result(completed) + + +def inspect_container_isolation( + docker_bin: str, + container: str, + *, + expected_image: str, + expected_label: str, + timeout: float, +) -> dict[str, Any]: + completed = _run([docker_bin, "container", "inspect", container], timeout=timeout) + raw = _require_success(completed, "container isolation inspection") + try: + rows = json.loads(raw) + observed = rows[0] + host_config = observed["HostConfig"] + config = observed["Config"] + except (IndexError, KeyError, TypeError, json.JSONDecodeError) as exc: + raise RuntimeError("container isolation inspection returned invalid JSON") from exc + + network_mode = host_config.get("NetworkMode") + tmpfs = host_config.get("Tmpfs") or {} + labels = config.get("Labels") or {} + image = config.get("Image") + running = bool((observed.get("State") or {}).get("Running")) + problems = [] + if network_mode != "none": + problems.append(f"network_mode={network_mode!r}") + if DATA_DIR not in tmpfs: + problems.append("postgres_data_tmpfs_missing") + if labels.get(CANARY_LABEL) != expected_label: + problems.append("canary_label_mismatch") + if image != expected_image: + problems.append(f"image={image!r}") + if not running: + problems.append("container_not_running") + if problems: + raise RuntimeError("container isolation check failed: " + ", ".join(problems)) + return { + "network_mode": network_mode, + "postgres_data_tmpfs": tmpfs[DATA_DIR], + "image": image, + "label": labels[CANARY_LABEL], + "running": running, + } + + +def _psql_command(container: str, database: str, *arguments: str, docker_bin: str) -> list[str]: + return [ + docker_bin, + "exec", + container, + "psql", + "-X", + "-U", + "postgres", + "-d", + database, + "-Atq", + "-v", + "ON_ERROR_STOP=1", + *arguments, + ] + + +def wait_for_named_database( + docker_bin: str, + container: str, + database: str, + *, + startup_timeout: float, + command_timeout: float, + poll_interval: float, +) -> int: + deadline = time.monotonic() + startup_timeout + attempts = 0 + last_error = "no psql attempt completed" + while True: + attempts += 1 + remaining = max(0.1, deadline - time.monotonic()) + completed = _run( + _psql_command( + container, + database, + "-c", + "select current_database();", + docker_bin=docker_bin, + ), + timeout=min(command_timeout, remaining), + ) + if completed.returncode == 0 and completed.stdout.strip() == database: + return attempts + last_error = _bounded_error(completed) + if time.monotonic() >= deadline: + raise RuntimeError( + f"named database did not accept a psql connection within {startup_timeout:g}s: {last_error}" + ) + time.sleep(min(poll_interval, max(0.0, deadline - time.monotonic()))) + + +def _role_keyword(value: bool, enabled: str, disabled: str) -> str: + return enabled if value else disabled + + +def application_role_sql(source_manifest: dict[str, Any]) -> str: + items = source_manifest["singleton"]["application_roles"].get("items", []) + if not isinstance(items, list): + raise ValueError("source manifest application_roles.items must be a list") + statements = [] + seen: set[str] = set() + for item in items: + if not isinstance(item, dict): + raise ValueError("source manifest application role must be an object") + name = item.get("name") + if name not in ALLOWED_APPLICATION_ROLES: + raise ValueError(f"source manifest contains non-allowlisted application role: {name!r}") + if name in seen: + raise ValueError(f"source manifest contains duplicate application role: {name}") + seen.add(name) + for field in ROLE_BOOLEAN_FIELDS: + if type(item.get(field)) is not bool: + raise ValueError(f"source manifest application role {name} has invalid {field}") + attributes = [ + _role_keyword(item["can_login"], "LOGIN", "NOLOGIN"), + _role_keyword(item["superuser"], "SUPERUSER", "NOSUPERUSER"), + _role_keyword(item["create_db"], "CREATEDB", "NOCREATEDB"), + _role_keyword(item["create_role"], "CREATEROLE", "NOCREATEROLE"), + _role_keyword(item["inherit"], "INHERIT", "NOINHERIT"), + _role_keyword(item["replication"], "REPLICATION", "NOREPLICATION"), + _role_keyword(item["bypass_rls"], "BYPASSRLS", "NOBYPASSRLS"), + ] + statements.append(f'CREATE ROLE "{name}" {" ".join(attributes)};') + return "\n".join(statements) + + +def execute_application_role_bootstrap( + docker_bin: str, + container: str, + database: str, + source_manifest: dict[str, Any], + *, + timeout: float, +) -> list[str]: + sql = application_role_sql(source_manifest) + if not sql: + return [] + completed = _run( + _psql_command(container, database, "-c", sql, docker_bin=docker_bin), + timeout=timeout, + ) + _require_success(completed, "application role bootstrap") + return sorted(item["name"] for item in source_manifest["singleton"]["application_roles"].get("items", [])) + + +def query_scalar( + docker_bin: str, + container: str, + database: str, + sql: str, + *, + timeout: float, +) -> str: + completed = _run( + _psql_command(container, database, "-c", sql, docker_bin=docker_bin), + timeout=timeout, + ) + return _require_success(completed, "PostgreSQL readback").strip() + + +def collect_key_counts( + docker_bin: str, + container: str, + database: str, + *, + timeout: float, +) -> tuple[dict[str, int], int]: + counts: dict[str, int] = {} + for table in KEY_TABLES: + exists = query_scalar( + docker_bin, + container, + database, + f"select to_regclass('{table}') is not null;", + timeout=timeout, + ) + if exists != "t": + raise RuntimeError(f"required canonical table is missing after restore: {table}") + schema, name = table.split(".", 1) + raw_count = query_scalar( + docker_bin, + container, + database, + f'select count(*) from "{schema}"."{name}";', + timeout=timeout, + ) + try: + counts[table] = int(raw_count) + except ValueError as exc: + raise RuntimeError(f"invalid row count for {table}: {raw_count!r}") from exc + raw_table_count = query_scalar( + docker_bin, + container, + database, + "select count(*) from pg_tables where schemaname in ('public', 'kb_stage');", + timeout=timeout, + ) + try: + table_count = int(raw_table_count) + except ValueError as exc: + raise RuntimeError(f"invalid canonical schema table count: {raw_table_count!r}") from exc + return counts, table_count + + +def run_parity_check( + docker_bin: str, + container: str, + database: str, + *, + source_manifest_path: Path, + source_manifest: dict[str, Any], + manifest_sql: Path, + command_timeout: float, + max_target_query_ms: float, + max_target_source_ratio: float, +) -> dict[str, Any]: + copy_result = _run( + [docker_bin, "cp", str(manifest_sql), f"{container}:{MANIFEST_DESTINATION}"], + timeout=command_timeout, + ) + _require_success(copy_result, "parity manifest copy") + completed = _run( + _psql_command( + container, + database, + "-f", + MANIFEST_DESTINATION, + docker_bin=docker_bin, + ), + timeout=command_timeout, + ) + target_payload = _require_success(completed, "target parity manifest capture") + with tempfile.TemporaryDirectory(prefix="teleo-canonical-parity-") as temp_dir: + target_path = Path(temp_dir) / "target-manifest.jsonl" + target_path.write_text(target_payload, encoding="utf-8") + target_manifest = load_manifest(target_path) + problems, details = compare_manifests( + source_manifest, + target_manifest, + max_target_query_ms=max_target_query_ms, + max_target_source_ratio=max_target_source_ratio, + ) + target_receipt = { + "bytes": target_path.stat().st_size, + "sha256": sha256_file(target_path), + "line_count": sum(1 for line in target_payload.splitlines() if line.strip()), + } + return { + "requested": True, + "status": "pass" if not problems else "fail", + "verifier": "ops.verify_postgres_parity_manifest.compare_manifests", + "source_manifest": { + "path": str(source_manifest_path), + "bytes": source_manifest_path.stat().st_size, + "sha256": sha256_file(source_manifest_path), + }, + "target_manifest": target_receipt, + "problems": problems, + "details": details, + } + + +def cleanup_container( + docker_bin: str, + container: str, + *, + command_timeout: float, + poll_interval: float, +) -> dict[str, Any]: + cleanup: dict[str, Any] = { + "attempted": True, + "container": container, + "force_remove_attempted": True, + "container_absent": None, + "status": "unproven", + } + try: + removed = _run([docker_bin, "rm", "--force", container], timeout=command_timeout) + cleanup["force_remove_exit_code"] = removed.returncode + if removed.returncode != 0: + cleanup["force_remove_message"] = _bounded_error(removed) + except (OSError, subprocess.TimeoutExpired) as exc: + cleanup["force_remove_error"] = str(exc) + except (KeyboardInterrupt, TerminationRequested) as exc: + cleanup["interrupted"] = str(exc) + cleanup["force_remove_error"] = str(exc) + + last_error = None + for attempt in range(1, 6): + cleanup["absence_inspection_attempts"] = attempt + try: + absent, error = inspect_container_absence( + docker_bin, + container, + timeout=command_timeout, + ) + except (OSError, subprocess.TimeoutExpired) as exc: + absent, error = None, str(exc) + except (KeyboardInterrupt, TerminationRequested) as exc: + cleanup["interrupted"] = str(exc) + absent, error = None, str(exc) + if absent is True: + cleanup["container_absent"] = True + cleanup["status"] = "pass" + cleanup.pop("absence_inspection_error", None) + return cleanup + cleanup["container_absent"] = absent + last_error = error + if error and "interrupted" not in cleanup: + cleanup["absence_inspection_error"] = error + break + if attempt < 5: + try: + time.sleep(poll_interval) + except (KeyboardInterrupt, TerminationRequested) as exc: + cleanup["interrupted"] = str(exc) + if last_error: + cleanup["absence_inspection_error"] = last_error + cleanup["status"] = "fail" if cleanup["container_absent"] is False else "unproven" + return cleanup + + +def _initial_receipt(args: argparse.Namespace, container: str) -> dict[str, Any]: + return { + "artifact": "local_canonical_postgres_rebuild_canary", + "generated_at_utc": datetime.now(UTC).isoformat(), + "status": "running", + "database": args.database, + "dump": { + "path": str(args.dump.resolve()), + "bytes": None, + "sha256": None, + "custom_format_signature_valid": False, + }, + "container": { + "name": container, + "image": args.image, + "network_mode_required": "none", + "postgres_data_tmpfs_required": f"rw,nosuid,nodev,noexec,size={args.tmpfs_mb}m", + "preexisting_absence_proven": False, + "started": False, + "isolation": None, + }, + "application_roles_bootstrapped": [], + "key_counts": {}, + "canonical_schema_table_count": None, + "parity": { + "requested": args.source_manifest is not None, + "status": "pending" if args.source_manifest is not None else "not_requested", + "reason": None if args.source_manifest is not None else "--source-manifest was not supplied", + }, + "timings_seconds": {}, + "cleanup": { + "attempted": False, + "container": container, + "container_absent": None, + "status": "pending", + }, + "safety": { + "production_database_touched": False, + "network_access_available_to_container": False, + "persistent_postgres_storage_used": False, + }, + "error": None, + } + + +def run_canary(args: argparse.Namespace) -> dict[str, Any]: + started_at = time.monotonic() + container = new_container_name(args.container_prefix) + receipt = _initial_receipt(args, container) + phase = "validation" + source_manifest: dict[str, Any] | None = None + primary_error: dict[str, str] | None = None + + try: + hash_started = time.monotonic() + validate_custom_dump(args.dump) + receipt["dump"].update( + { + "bytes": args.dump.stat().st_size, + "sha256": sha256_file(args.dump), + "custom_format_signature_valid": True, + } + ) + receipt["timings_seconds"]["dump_validation_and_hash"] = round(time.monotonic() - hash_started, 6) + + if args.source_manifest is not None: + phase = "source_manifest_validation" + source_manifest = load_manifest(args.source_manifest) + + phase = "container_name_preflight" + absent, absence_error = inspect_container_absence( + args.docker_bin, + container, + timeout=args.command_timeout, + ) + if absent is not True: + detail = absence_error or "generated container name unexpectedly already exists" + raise RuntimeError(f"could not prove generated container name is unused: {detail}") + receipt["container"]["preexisting_absence_proven"] = True + + phase = "container_start" + startup_started = time.monotonic() + tmpfs_options = f"{DATA_DIR}:rw,nosuid,nodev,noexec,size={args.tmpfs_mb}m" + run_result = _run( + [ + args.docker_bin, + "run", + "--detach", + "--rm", + "--network", + "none", + "--name", + container, + "--label", + f"{CANARY_LABEL}={container}", + "--tmpfs", + tmpfs_options, + "--env", + f"POSTGRES_DB={args.database}", + "--env", + "POSTGRES_HOST_AUTH_METHOD=trust", + args.image, + ], + timeout=args.command_timeout, + ) + container_id = _require_success(run_result, "networkless PostgreSQL container start").strip() + if not container_id: + raise RuntimeError("networkless PostgreSQL container start returned no container id") + receipt["container"]["started"] = True + receipt["container"]["id_prefix"] = container_id[:12] + receipt["container"]["isolation"] = inspect_container_isolation( + args.docker_bin, + container, + expected_image=args.image, + expected_label=container, + timeout=args.command_timeout, + ) + + phase = "named_database_readiness" + readiness_attempts = wait_for_named_database( + args.docker_bin, + container, + args.database, + startup_timeout=args.startup_timeout, + command_timeout=args.command_timeout, + poll_interval=args.poll_interval, + ) + receipt["container"]["named_database_psql_ready"] = True + receipt["container"]["named_database_psql_attempts"] = readiness_attempts + receipt["timings_seconds"]["container_start_and_named_database_ready"] = round( + time.monotonic() - startup_started, + 6, + ) + + if source_manifest is not None: + phase = "application_role_bootstrap" + receipt["application_roles_bootstrapped"] = execute_application_role_bootstrap( + args.docker_bin, + container, + args.database, + source_manifest, + timeout=args.command_timeout, + ) + + phase = "dump_copy" + copy_result = _run( + [args.docker_bin, "cp", str(args.dump.resolve()), f"{container}:{DUMP_DESTINATION}"], + timeout=args.command_timeout, + ) + _require_success(copy_result, "custom dump copy") + + phase = "restore" + restore_started = time.monotonic() + restore_result = _run( + [ + args.docker_bin, + "exec", + container, + "pg_restore", + "-U", + "postgres", + "-d", + args.database, + "--no-owner", + "--no-privileges", + "--exit-on-error", + DUMP_DESTINATION, + ], + timeout=args.restore_timeout, + ) + _require_success(restore_result, "canonical PostgreSQL restore") + receipt["timings_seconds"]["pg_restore"] = round(time.monotonic() - restore_started, 6) + + phase = "key_count_readback" + count_started = time.monotonic() + key_counts, table_count = collect_key_counts( + args.docker_bin, + container, + args.database, + timeout=args.command_timeout, + ) + receipt["key_counts"] = key_counts + receipt["canonical_schema_table_count"] = table_count + receipt["timings_seconds"]["key_count_readback"] = round(time.monotonic() - count_started, 6) + + if source_manifest is not None: + phase = "parity_manifest_and_compare" + parity_started = time.monotonic() + receipt["parity"] = run_parity_check( + args.docker_bin, + container, + args.database, + source_manifest_path=args.source_manifest, + source_manifest=source_manifest, + manifest_sql=args.manifest_sql, + command_timeout=args.manifest_timeout, + max_target_query_ms=args.max_target_query_ms, + max_target_source_ratio=args.max_target_source_ratio, + ) + receipt["timings_seconds"]["manifest_capture_and_parity_compare"] = round( + time.monotonic() - parity_started, + 6, + ) + if receipt["parity"]["status"] != "pass": + raise RuntimeError("canonical parity verification failed") + receipt["status"] = "pass" + except ( + OSError, + ValueError, + RuntimeError, + subprocess.TimeoutExpired, + json.JSONDecodeError, + KeyboardInterrupt, + TerminationRequested, + ) as exc: + primary_error = { + "phase": phase, + "type": type(exc).__name__, + "message": str(exc)[-2000:], + } + receipt["status"] = "fail" + receipt["error"] = primary_error + finally: + cleanup_started = time.monotonic() + cleanup = cleanup_container( + args.docker_bin, + container, + command_timeout=args.command_timeout, + poll_interval=args.poll_interval, + ) + receipt["cleanup"] = cleanup + receipt["timings_seconds"]["cleanup"] = round(time.monotonic() - cleanup_started, 6) + receipt["timings_seconds"]["total"] = round(time.monotonic() - started_at, 6) + if cleanup.get("container_absent") is not True: + receipt["status"] = "fail" + cleanup_error = { + "phase": "cleanup", + "type": "CleanupProofError", + "message": "container absence was not independently proven", + } + if primary_error is None: + receipt["error"] = cleanup_error + else: + receipt["cleanup_error"] = cleanup_error + if cleanup.get("interrupted"): + receipt["status"] = "fail" + interruption_error = { + "phase": "cleanup", + "type": "TerminationRequested", + "message": cleanup["interrupted"], + } + if primary_error is None: + receipt["error"] = interruption_error + else: + receipt["cleanup_interruption"] = interruption_error + return receipt + + +def write_receipt(path: Path, payload: dict[str, Any]) -> None: + path = path.expanduser() + if not path.is_absolute(): + path = Path.cwd() / path + path.parent.mkdir(parents=True, exist_ok=True) + path = path.parent.resolve() / path.name + encoded = (json.dumps(payload, indent=2, sort_keys=True) + "\n").encode() + temporary = path.parent / f".{path.name}.{uuid.uuid4().hex}.tmp" + descriptor = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + try: + with os.fdopen(descriptor, "wb") as handle: + handle.write(encoded) + handle.flush() + os.fsync(handle.fileno()) + os.replace(temporary, path) + os.chmod(path, 0o600) + finally: + try: + temporary.unlink() + except FileNotFoundError: + pass + + +def parse_args(argv: list[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--dump", required=True, type=Path, help="PostgreSQL custom-format pg_dump") + parser.add_argument("--database", default="teleo") + parser.add_argument("--image", default=DEFAULT_IMAGE) + parser.add_argument("--container-prefix", default="teleo-canonical-rebuild") + parser.add_argument("--tmpfs-mb", default=1024, type=int) + parser.add_argument("--source-manifest", type=Path) + parser.add_argument("--manifest-sql", default=DEFAULT_MANIFEST_SQL, type=Path) + parser.add_argument("--max-target-query-ms", default=2000.0, type=float) + parser.add_argument("--max-target-source-ratio", default=20.0, type=float) + parser.add_argument("--startup-timeout", default=60.0, type=float) + parser.add_argument("--restore-timeout", default=300.0, type=float) + parser.add_argument("--manifest-timeout", default=180.0, type=float) + parser.add_argument("--command-timeout", default=30.0, type=float) + parser.add_argument("--poll-interval", default=0.25, type=float) + parser.add_argument("--docker-bin", default="docker") + parser.add_argument("--output", required=True, type=Path) + args = parser.parse_args(argv) + + if not DATABASE_NAME_RE.fullmatch(args.database): + parser.error("--database must be a safe PostgreSQL identifier up to 63 characters") + if not CONTAINER_PREFIX_RE.fullmatch(args.container_prefix): + parser.error("--container-prefix must be 1-40 Docker-safe characters") + if not args.image or any(character.isspace() for character in args.image): + parser.error("--image must be a non-empty Docker image reference without whitespace") + if not args.docker_bin or any(character.isspace() for character in args.docker_bin): + parser.error("--docker-bin must be one executable name or path without whitespace") + if not 64 <= args.tmpfs_mb <= 65536: + parser.error("--tmpfs-mb must be between 64 and 65536") + for value, flag in ( + (args.startup_timeout, "--startup-timeout"), + (args.restore_timeout, "--restore-timeout"), + (args.manifest_timeout, "--manifest-timeout"), + (args.command_timeout, "--command-timeout"), + (args.poll_interval, "--poll-interval"), + (args.max_target_query_ms, "--max-target-query-ms"), + (args.max_target_source_ratio, "--max-target-source-ratio"), + ): + if value <= 0: + parser.error(f"{flag} must be positive") + if not args.dump.is_file(): + parser.error("--dump must be an existing regular file") + if args.source_manifest is not None: + if not args.source_manifest.is_file(): + parser.error("--source-manifest must be an existing regular file") + if not args.manifest_sql.is_file(): + parser.error("--manifest-sql must be an existing regular file when parity is requested") + protected_inputs = {args.dump.resolve(), args.manifest_sql.resolve()} + if args.source_manifest is not None: + protected_inputs.add(args.source_manifest.resolve()) + if args.output.resolve() in protected_inputs: + parser.error("--output must not overwrite a dump or manifest input") + return args + + +def _termination_handler(signum: int, _frame: Any) -> None: + raise TerminationRequested(f"received {signal.Signals(signum).name}") + + +def main(argv: list[str] | None = None) -> int: + args = parse_args(argv) + previous_handlers = {} + for watched_signal in (signal.SIGINT, signal.SIGTERM): + previous_handlers[watched_signal] = signal.getsignal(watched_signal) + signal.signal(watched_signal, _termination_handler) + try: + payload = run_canary(args) + finally: + for watched_signal, previous in previous_handlers.items(): + signal.signal(watched_signal, previous) + + try: + write_receipt(args.output, payload) + except OSError as exc: + payload["status"] = "fail" + payload["receipt_write_error"] = str(exc) + print(json.dumps(payload, indent=2, sort_keys=True)) + return 0 if payload["status"] == "pass" else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_run_local_canonical_postgres_rebuild.py b/tests/test_run_local_canonical_postgres_rebuild.py new file mode 100644 index 0000000..4bd4860 --- /dev/null +++ b/tests/test_run_local_canonical_postgres_rebuild.py @@ -0,0 +1,382 @@ +import argparse +import json +import stat +import subprocess +from pathlib import Path + +import pytest + +from ops import run_local_canonical_postgres_rebuild as rebuild + +COUNTS = { + "public.claims": 1837, + "public.sources": 4145, + "public.claim_evidence": 4670, + "public.claim_edges": 4916, + "public.reasoning_tools": 1, + "kb_stage.kb_proposals": 26, +} + + +def completed(command: list[str], returncode: int = 0, stdout: str = "", stderr: str = ""): + return subprocess.CompletedProcess(command, returncode, stdout, stderr) + + +class FakeDocker: + def __init__( + self, + *, + restore_fails: bool = False, + preserve_on_remove: bool = False, + manifest_text: str = "", + ) -> None: + self.restore_fails = restore_fails + self.preserve_on_remove = preserve_on_remove + self.manifest_text = manifest_text + self.commands: list[list[str]] = [] + self.running = False + self.container_name = "" + self.image = "" + self.label = "" + self.tmpfs = "" + + def __call__(self, command: list[str], *, timeout: float): + assert timeout > 0 + self.commands.append(command) + + if command[1:3] == ["container", "inspect"]: + if not self.running: + return completed(command, 1, stderr=f"Error: No such object: {command[-1]}") + payload = [ + { + "Config": { + "Image": self.image, + "Labels": {rebuild.CANARY_LABEL: self.label}, + }, + "HostConfig": { + "NetworkMode": "none", + "Tmpfs": {rebuild.DATA_DIR: self.tmpfs.split(":", 1)[1]}, + }, + "State": {"Running": True}, + } + ] + return completed(command, stdout=json.dumps(payload)) + + if command[1] == "run": + self.running = True + self.container_name = command[command.index("--name") + 1] + self.label = command[command.index("--label") + 1].split("=", 1)[1] + self.tmpfs = command[command.index("--tmpfs") + 1] + self.image = command[-1] + return completed(command, stdout="0123456789abcdef\n") + + if command[1] == "cp": + return completed(command) + + if command[1] == "exec" and command[3] == "pg_restore": + if self.restore_fails: + return completed(command, 1, stderr="pg_restore: error: fixture restore failure") + return completed(command) + + if command[1] == "exec" and command[3] == "psql": + if "-f" in command: + return completed(command, stdout=self.manifest_text) + sql = command[command.index("-c") + 1] + if sql == "select current_database();": + return completed(command, stdout="teleo\n") + if sql.startswith("CREATE ROLE"): + return completed(command) + if sql.startswith("select to_regclass("): + return completed(command, stdout="t\n") + if "from pg_tables" in sql: + return completed(command, stdout="47\n") + for table, count in COUNTS.items(): + schema, name = table.split(".", 1) + if f'from "{schema}"."{name}"' in sql: + return completed(command, stdout=f"{count}\n") + raise AssertionError(f"unexpected psql query: {sql}") + + if command[1:3] == ["rm", "--force"]: + if not self.preserve_on_remove: + self.running = False + return completed(command, stdout=f"{command[-1]}\n") + return completed(command, 1, stderr="fixture refused removal") + + raise AssertionError(f"unexpected command: {command}") + + +def write_dump(path: Path, payload: bytes = b"fixture") -> Path: + path.write_bytes(b"PGDMP" + payload) + return path + + +def manifest_rows() -> list[dict]: + role = { + "name": "kb_apply", + "can_login": True, + "superuser": False, + "create_db": False, + "create_role": False, + "inherit": True, + "replication": False, + "bypass_rls": False, + } + rows = [ + { + "kind": "identity", + "database": "teleo", + "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": 1837, + "rowset_md5": "same-row-hash", + }, + { + "kind": "performance", + "query": "count_claims", + "elapsed_ms": 1.0, + "observed_rows": 1837, + }, + ] + ) + 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 args_for(tmp_path: Path, **overrides) -> argparse.Namespace: + values = { + "dump": write_dump(tmp_path / "canonical.dump"), + "database": "teleo", + "image": rebuild.DEFAULT_IMAGE, + "container_prefix": "teleo-canonical-rebuild-test", + "tmpfs_mb": 256, + "source_manifest": None, + "manifest_sql": Path("ops/postgres_parity_manifest.sql").resolve(), + "max_target_query_ms": 2000.0, + "max_target_source_ratio": 20.0, + "startup_timeout": 2.0, + "restore_timeout": 5.0, + "manifest_timeout": 5.0, + "command_timeout": 2.0, + "poll_interval": 0.001, + "docker_bin": "docker", + "output": tmp_path / "receipt.json", + } + values.update(overrides) + return argparse.Namespace(**values) + + +def test_success_uses_networkless_tmpfs_named_database_and_required_restore_flags( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + fake = FakeDocker() + monkeypatch.setattr(rebuild, "_run", fake) + + result = rebuild.run_canary(args_for(tmp_path)) + + assert result["status"] == "pass" + assert result["container"]["name"].startswith("teleo-canonical-rebuild-test-") + assert result["container"]["named_database_psql_ready"] is True + assert result["container"]["isolation"]["network_mode"] == "none" + assert result["key_counts"] == COUNTS + assert result["canonical_schema_table_count"] == 47 + assert result["cleanup"]["status"] == "pass" + assert result["cleanup"]["container_absent"] is True + + run_command = next(command for command in fake.commands if command[1] == "run") + assert run_command[run_command.index("--network") + 1] == "none" + assert run_command[run_command.index("--tmpfs") + 1] == (f"{rebuild.DATA_DIR}:rw,nosuid,nodev,noexec,size=256m") + assert "--rm" in run_command + + readiness_index = next( + index for index, command in enumerate(fake.commands) if "select current_database();" in command + ) + restore_index = next( + index for index, command in enumerate(fake.commands) if len(command) > 3 and command[3] == "pg_restore" + ) + assert readiness_index < restore_index + readiness_command = fake.commands[readiness_index] + assert readiness_command[readiness_command.index("-d") + 1] == "teleo" + assert not any("pg_isready" in part for command in fake.commands for part in command) + + restore_command = fake.commands[restore_index] + assert "--no-owner" in restore_command + assert "--no-privileges" in restore_command + assert "--exit-on-error" in restore_command + + +def test_optional_manifest_bootstraps_allowlisted_roles_and_uses_existing_verifier( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + source_manifest = write_manifest(tmp_path / "source-manifest.jsonl") + fake = FakeDocker(manifest_text=source_manifest.read_text()) + monkeypatch.setattr(rebuild, "_run", fake) + + result = rebuild.run_canary(args_for(tmp_path, source_manifest=source_manifest)) + + assert result["status"] == "pass" + assert result["application_roles_bootstrapped"] == ["kb_apply"] + assert result["parity"]["status"] == "pass" + assert result["parity"]["verifier"] == "ops.verify_postgres_parity_manifest.compare_manifests" + assert result["parity"]["problems"] == [] + assert result["parity"]["details"]["source_total_rows"] == 1837 + assert result["parity"]["target_manifest"]["line_count"] == len(manifest_rows()) + + role_command = next( + command + for command in fake.commands + if command[1] == "exec" and command[3] == "psql" and any("CREATE ROLE" in part for part in command) + ) + role_sql = role_command[role_command.index("-c") + 1] + assert 'CREATE ROLE "kb_apply" LOGIN NOSUPERUSER' in role_sql + assert "PASSWORD" not in role_sql + assert any(command[1] == "cp" and rebuild.MANIFEST_DESTINATION in command[-1] for command in fake.commands) + + +def test_restore_failure_still_forces_removal_and_proves_absence( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + fake = FakeDocker(restore_fails=True) + monkeypatch.setattr(rebuild, "_run", fake) + + result = rebuild.run_canary(args_for(tmp_path)) + + assert result["status"] == "fail" + assert result["error"]["phase"] == "restore" + assert "fixture restore failure" in result["error"]["message"] + assert result["cleanup"]["force_remove_exit_code"] == 0 + assert result["cleanup"]["container_absent"] is True + assert result["cleanup"]["status"] == "pass" + assert any(command[1:3] == ["rm", "--force"] for command in fake.commands) + assert fake.running is False + + +def test_unproven_cleanup_overrides_success_and_fails_closed( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + fake = FakeDocker(preserve_on_remove=True) + monkeypatch.setattr(rebuild, "_run", fake) + monkeypatch.setattr(rebuild.time, "sleep", lambda _seconds: None) + + result = rebuild.run_canary(args_for(tmp_path)) + + assert result["status"] == "fail" + assert result["error"]["phase"] == "cleanup" + assert result["error"]["type"] == "CleanupProofError" + assert result["cleanup"]["container_absent"] is False + assert result["cleanup"]["status"] == "fail" + assert result["cleanup"]["absence_inspection_attempts"] == 5 + + +def test_cleanup_interruption_is_receipted_and_does_not_bypass_absence_proof( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + fake = FakeDocker() + interrupted = False + + def interrupt_cleanup(command: list[str], *, timeout: float): + nonlocal interrupted + if command[1:3] == ["rm", "--force"] and not interrupted: + interrupted = True + fake.running = False + raise rebuild.TerminationRequested("received SIGTERM") + return fake(command, timeout=timeout) + + monkeypatch.setattr(rebuild, "_run", interrupt_cleanup) + + result = rebuild.run_canary(args_for(tmp_path)) + + assert result["status"] == "fail" + assert result["error"] == { + "phase": "cleanup", + "type": "TerminationRequested", + "message": "received SIGTERM", + } + assert result["cleanup"]["container_absent"] is True + assert result["cleanup"]["status"] == "pass" + + +def test_invalid_dump_fails_before_start_but_still_proves_name_absent( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + invalid = tmp_path / "plain.sql" + invalid.write_text("select 1;") + fake = FakeDocker() + monkeypatch.setattr(rebuild, "_run", fake) + + result = rebuild.run_canary(args_for(tmp_path, dump=invalid)) + + assert result["status"] == "fail" + assert result["error"]["phase"] == "validation" + assert "PGDMP" in result["error"]["message"] + assert not any(command[1] == "run" for command in fake.commands) + assert result["cleanup"]["container_absent"] is True + + +def test_application_role_sql_rejects_roles_outside_exact_allowlist() -> None: + source = {"singleton": {"application_roles": {"items": [{"name": "postgres"}]}}} + + with pytest.raises(ValueError, match="non-allowlisted"): + rebuild.application_role_sql(source) + + +def test_container_names_are_unique_and_bounded() -> None: + first = rebuild.new_container_name("teleo-rebuild") + second = rebuild.new_container_name("teleo-rebuild") + + assert first != second + assert first.startswith("teleo-rebuild-") + assert len(first) <= 63 + + +def test_receipt_write_is_atomic_private_and_replaces_symlink_not_target(tmp_path: Path) -> None: + target = tmp_path / "unrelated.json" + target.write_text("keep") + receipt = tmp_path / "receipt.json" + receipt.symlink_to(target) + + rebuild.write_receipt(receipt, {"status": "pass"}) + + assert json.loads(receipt.read_text()) == {"status": "pass"} + assert target.read_text() == "keep" + assert stat.S_IMODE(receipt.stat().st_mode) == 0o600