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

Add one-command canonical KB rebuild with parity

Browse source
This commit is contained in:
twentyOne2x 2026-07-13 10:50:11 +02:00
parent b570db63b0
commit 466bf34347
5 changed files with 1404 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

20
docs/gcp-kb-cloudsql-restore-runbook.md
View file

@ -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 <private-output-dir>/teleo-canonical.dump \
--source-manifest <private-output-dir>/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.

124
docs/kb-rebuild-and-recompile.md Normal file
View file

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

60
docs/reports/leo-working-state-20260709/local-canonical-postgres-rebuild-canary-20260713.md Normal file
View file

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

819
ops/run_local_canonical_postgres_rebuild.py Executable file
View file

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

382
tests/test_run_local_canonical_postgres_rebuild.py Normal file
View file

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