2383 lines
94 KiB
Python
2383 lines
94 KiB
Python
#!/usr/bin/env python3
|
|
"""Deterministically rebuild a disposable Postgres from genesis plus a strict ledger.
|
|
|
|
The v1 replay boundary supports strict ``add_edge``, ``attach_evidence``,
|
|
V2/V3 ``approve_claim``, and ``revise_strategy`` receipts. Insert-only actions seed
|
|
their receipt-pinned rows before the guarded transition. ``revise_strategy``
|
|
instead captures the exact prestate identities, executes the exact hash-matched
|
|
guarded SQL, validates the generated delta, and normalizes only that delta to the
|
|
receipt-pinned transaction timestamp, IDs, and rows. Every entry must retain the
|
|
full approved and applied proposal rows plus its immutable approval snapshot.
|
|
Legacy and freeform entries still fail before Docker starts.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import signal
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import uuid
|
|
from dataclasses import dataclass
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
try:
|
|
from . import run_local_canonical_postgres_rebuild as canonical_rebuild
|
|
from .verify_postgres_parity_manifest import load_manifest
|
|
except ImportError: # pragma: no cover - direct script execution
|
|
import run_local_canonical_postgres_rebuild as canonical_rebuild
|
|
from verify_postgres_parity_manifest import load_manifest
|
|
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
SCRIPTS_DIR = REPO_ROOT / "scripts"
|
|
if str(SCRIPTS_DIR) not in sys.path:
|
|
sys.path.insert(0, str(SCRIPTS_DIR))
|
|
|
|
import apply_proposal as apply_engine # noqa: E402, I001
|
|
import export_kb_transition_replay_material as replay_exporter # noqa: E402
|
|
import kb_apply_replay_receipt as replay_receipt # noqa: E402
|
|
|
|
|
|
LEDGER_ARTIFACT = "teleo_genesis_plus_ledger"
|
|
MATERIAL_ARTIFACT = "teleo_strict_proposal_replay_material"
|
|
ADMISSION_ARTIFACT = "teleo_ledger_admission_manifest"
|
|
LEDGER_CONTRACT_VERSION = 1
|
|
MATERIAL_CONTRACT_VERSION = 1
|
|
ADMISSION_CONTRACT_VERSION = 1
|
|
CANONICAL_REVIEWER_HANDLE = "m3taversal"
|
|
REVIEW_METADATA_PROVENANCE = "operator_supplied_explicit_review_metadata"
|
|
ADMISSION_BASES = frozenset(
|
|
{
|
|
"forgejo_primary_source",
|
|
"reviewed_telegram_steering",
|
|
"reviewed_primary_source",
|
|
}
|
|
)
|
|
ADMISSION_SCOPES = frozenset({"collective_core", "agent_identity", "knowledge_commons"})
|
|
DEFAULT_REBUILD_IMAGE = (
|
|
"docker.io/library/postgres:16-alpine@sha256:57c72fd2a128e416c7fcc499958864df5301e940bca0a56f58fddf30ffc07777"
|
|
)
|
|
SUPPORTED_PROPOSAL_TYPES = frozenset({"add_edge", "attach_evidence", "approve_claim", "revise_strategy"})
|
|
CLAIM_CEILING = (
|
|
"exact genesis plus ordered strict proposal replay; supported types are add_edge, "
|
|
"attach_evidence, V2/V3 approve_claim, and revise_strategy; V3 cutover is ordered and exact; "
|
|
"mutating strategy replay "
|
|
"captures prestate identities and normalizes the exact receipt-pinned poststate; "
|
|
"full approved/applied proposal rows and immutable approval snapshots are required"
|
|
)
|
|
NOT_PROVEN = (
|
|
"legacy or freeform proposal replay",
|
|
"source-corpus recompilation from a blank schema",
|
|
"VPS, GCP, Telegram, or any live database mutation",
|
|
)
|
|
|
|
APPLY_ENGINE_PATH = SCRIPTS_DIR / "apply_proposal.py"
|
|
REPLAY_RECEIPT_ENGINE_PATH = SCRIPTS_DIR / "kb_apply_replay_receipt.py"
|
|
GUARD_PREREQUISITES_PATH = SCRIPTS_DIR / "kb_apply_prereqs.sql"
|
|
V3_CONTRACT_PATH = REPO_ROOT / "ops" / "teleo_v3_epistemic_contract.sql"
|
|
PARITY_SQL_PATH = REPO_ROOT / "ops" / "postgres_parity_manifest.sql"
|
|
BASE_REBUILD_PATH = REPO_ROOT / "ops" / "run_local_canonical_postgres_rebuild.py"
|
|
GUARD_PREREQUISITES_DESTINATION = "/tmp/kb-apply-prereqs.sql"
|
|
V3_CONTRACT_DESTINATION = "/tmp/teleo-v3-epistemic-contract.sql"
|
|
FIXED_ENGINE_PATHS = frozenset(
|
|
{
|
|
APPLY_ENGINE_PATH.resolve(),
|
|
REPLAY_RECEIPT_ENGINE_PATH.resolve(),
|
|
GUARD_PREREQUISITES_PATH.resolve(),
|
|
V3_CONTRACT_PATH.resolve(),
|
|
PARITY_SQL_PATH.resolve(),
|
|
BASE_REBUILD_PATH.resolve(),
|
|
Path(__file__).resolve(),
|
|
}
|
|
)
|
|
|
|
SHA256_RE = re.compile(r"[0-9a-f]{64}\Z")
|
|
IMAGE_DIGEST_RE = re.compile(r"\S+@sha256:[0-9a-f]{64}\Z")
|
|
TYPED_TIMESTAMP_FIELDS = frozenset({"captured_at", "created_at", "updated_at", "reviewed_at", "applied_at"})
|
|
PROPOSAL_TRANSITION_FIELDS = frozenset(
|
|
{"status", "applied_by_handle", "applied_by_agent_id", "applied_at", "updated_at"}
|
|
)
|
|
PROPOSAL_FIELDS = frozenset(
|
|
{
|
|
"id",
|
|
"proposal_type",
|
|
"status",
|
|
"proposed_by_handle",
|
|
"proposed_by_agent_id",
|
|
"channel",
|
|
"source_ref",
|
|
"rationale",
|
|
"payload",
|
|
"reviewed_by_handle",
|
|
"reviewed_by_agent_id",
|
|
"reviewed_at",
|
|
"review_note",
|
|
"applied_by_handle",
|
|
"applied_by_agent_id",
|
|
"applied_at",
|
|
"created_at",
|
|
"updated_at",
|
|
}
|
|
)
|
|
APPROVAL_FIELDS = frozenset(
|
|
{
|
|
"proposal_id",
|
|
"proposal_type",
|
|
"payload",
|
|
"reviewed_by_handle",
|
|
"reviewed_by_agent_id",
|
|
"reviewed_by_db_role",
|
|
"reviewed_at",
|
|
"review_note",
|
|
}
|
|
)
|
|
CANONICAL_TABLE_ORDER = (
|
|
"claims",
|
|
"sources",
|
|
"reasoning_tools",
|
|
"claim_evidence",
|
|
"claim_evidence_assessments",
|
|
"claim_edges",
|
|
)
|
|
STRATEGY_REQUIRED_FIELDS = frozenset(
|
|
{
|
|
"id",
|
|
"agent_id",
|
|
"diagnosis",
|
|
"guiding_policy",
|
|
"proximate_objectives",
|
|
"version",
|
|
"active",
|
|
"created_at",
|
|
}
|
|
)
|
|
STRATEGY_NODE_REQUIRED_FIELDS = frozenset(
|
|
{
|
|
"id",
|
|
"agent_id",
|
|
"node_type",
|
|
"title",
|
|
"body",
|
|
"rank",
|
|
"status",
|
|
"horizon",
|
|
"measure",
|
|
"source_ref",
|
|
"metadata",
|
|
"created_at",
|
|
"updated_at",
|
|
}
|
|
)
|
|
|
|
|
|
class ReconstructionError(RuntimeError):
|
|
"""A fail-closed error whose message is safe for the public receipt."""
|
|
|
|
def __init__(self, code: str, safe_message: str) -> None:
|
|
super().__init__(safe_message)
|
|
self.code = code
|
|
self.safe_message = safe_message
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class LedgerEntry:
|
|
sequence: int
|
|
material_path: Path
|
|
material_sha256: str
|
|
replay_material_sha256: str
|
|
contract_version: int
|
|
approved_proposal: dict[str, Any]
|
|
approval_snapshot: dict[str, Any]
|
|
applied_proposal: dict[str, Any]
|
|
receipt: dict[str, Any]
|
|
current_apply_sql: str
|
|
current_apply_sql_sha256: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ReconstructionBundle:
|
|
ledger_sha256: str
|
|
ledger_bytes: int
|
|
genesis_dump_sha256: str
|
|
genesis_dump_bytes: int
|
|
genesis_manifest: dict[str, Any]
|
|
genesis_manifest_sha256: str
|
|
genesis_manifest_bytes: int
|
|
final_manifest_path: Path
|
|
final_manifest: dict[str, Any]
|
|
final_manifest_sha256: str
|
|
final_manifest_bytes: int
|
|
engine_hashes: dict[str, str]
|
|
entries: tuple[LedgerEntry, ...]
|
|
admission: dict[str, Any]
|
|
admission_manifest_path: Path | None
|
|
admission_manifest_sha256: str | None
|
|
admission_manifest_bytes: int | None
|
|
protected_paths: frozenset[Path]
|
|
|
|
|
|
def _canonical_json(value: Any) -> str:
|
|
return json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=True)
|
|
|
|
|
|
def _sha256_text(value: str) -> str:
|
|
return hashlib.sha256(value.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def _validate_sha256(value: Any, field: str) -> str:
|
|
if not isinstance(value, str) or not SHA256_RE.fullmatch(value):
|
|
raise ReconstructionError("invalid_hash_pin", f"{field} must be a lowercase SHA-256 digest")
|
|
return value
|
|
|
|
|
|
def _require_exact_keys(value: Any, expected: frozenset[str], field: str) -> dict[str, Any]:
|
|
if not isinstance(value, dict):
|
|
raise ReconstructionError("invalid_contract", f"{field} must be an object")
|
|
observed = frozenset(value)
|
|
if observed != expected:
|
|
raise ReconstructionError(
|
|
"invalid_contract",
|
|
f"{field} fields do not match v1 contract; "
|
|
f"missing_count={len(expected - observed)}, extra_count={len(observed - expected)}",
|
|
)
|
|
return value
|
|
|
|
|
|
def _load_json_object(path: Path, field: str) -> dict[str, Any]:
|
|
try:
|
|
raw = path.read_text(encoding="utf-8")
|
|
return replay_exporter._parse_json_object(raw, label=field)
|
|
except replay_exporter.ExportError as exc:
|
|
raise ReconstructionError(
|
|
"invalid_json_artifact",
|
|
f"{field} is not exact duplicate-free UTF-8 JSON",
|
|
) from exc
|
|
except (OSError, UnicodeError) as exc:
|
|
raise ReconstructionError(
|
|
"invalid_json_artifact",
|
|
f"{field} is not exact duplicate-free UTF-8 JSON",
|
|
) from exc
|
|
|
|
|
|
def _resolve_artifact(ledger_path: Path, declared: Any, field: str) -> Path:
|
|
if not isinstance(declared, str) or not declared.strip():
|
|
raise ReconstructionError("invalid_contract", f"{field} must be a non-empty path string")
|
|
candidate = Path(declared).expanduser()
|
|
if not candidate.is_absolute():
|
|
candidate = ledger_path.parent / candidate
|
|
candidate = candidate.resolve()
|
|
if not candidate.is_file():
|
|
raise ReconstructionError("missing_artifact", f"{field} is not an existing regular file")
|
|
return candidate
|
|
|
|
|
|
def _assert_file_hash(path: Path, expected: Any, field: str) -> str:
|
|
pin = _validate_sha256(expected, field)
|
|
actual = canonical_rebuild.sha256_file(path)
|
|
if actual != pin:
|
|
raise ReconstructionError("hash_pin_mismatch", f"{field} does not match the supplied artifact")
|
|
return actual
|
|
|
|
|
|
def _engine_hashes() -> dict[str, str]:
|
|
paths = {
|
|
"reconstruction_command_sha256": Path(__file__),
|
|
"base_rebuild_engine_sha256": BASE_REBUILD_PATH,
|
|
"apply_engine_sha256": APPLY_ENGINE_PATH,
|
|
"replay_receipt_engine_sha256": REPLAY_RECEIPT_ENGINE_PATH,
|
|
"guard_prerequisites_sha256": GUARD_PREREQUISITES_PATH,
|
|
"v3_contract_sha256": V3_CONTRACT_PATH,
|
|
"parity_sql_sha256": PARITY_SQL_PATH,
|
|
}
|
|
return {name: canonical_rebuild.sha256_file(path) for name, path in paths.items()}
|
|
|
|
|
|
def _proposal_contract_version(proposal: dict[str, Any]) -> int:
|
|
payload = proposal.get("payload")
|
|
apply_payload = payload.get("apply_payload") if isinstance(payload, dict) else None
|
|
if not isinstance(apply_payload, dict):
|
|
raise ReconstructionError("invalid_material", "proposal is missing a strict apply payload")
|
|
raw_version = apply_payload.get("contract_version", 2)
|
|
if type(raw_version) is not int or raw_version not in {2, 3}:
|
|
raise ReconstructionError("unsupported_contract_version", "proposal contract version is unsupported")
|
|
if raw_version == 3 and proposal.get("proposal_type") != "approve_claim":
|
|
raise ReconstructionError("unsupported_contract_version", "V3 replay currently supports approve_claim only")
|
|
return raw_version
|
|
|
|
|
|
def _validate_contract_order(entries: list[LedgerEntry]) -> None:
|
|
v3_started = False
|
|
for entry in entries:
|
|
if entry.contract_version == 3:
|
|
v3_started = True
|
|
elif v3_started:
|
|
raise ReconstructionError(
|
|
"contract_order_mismatch",
|
|
"V2 ledger entries cannot appear after the ordered V3 cutover",
|
|
)
|
|
|
|
|
|
def _load_admission_manifest(
|
|
args: argparse.Namespace,
|
|
*,
|
|
ledger_sha256: str,
|
|
entries: list[LedgerEntry],
|
|
) -> tuple[dict[str, Any], Path | None, str | None, int | None]:
|
|
manifest_path = getattr(args, "admission_manifest", None)
|
|
manifest_pin = getattr(args, "admission_manifest_sha256", None)
|
|
if manifest_path is None and manifest_pin is None:
|
|
return (
|
|
{
|
|
"status": "forensic_only",
|
|
"canonical_retrieval_eligible": False,
|
|
"reason": "no explicit admission manifest was supplied",
|
|
"entry_count": len(entries),
|
|
},
|
|
None,
|
|
None,
|
|
None,
|
|
)
|
|
if manifest_path is None or manifest_pin is None:
|
|
raise ReconstructionError(
|
|
"incomplete_admission_pin",
|
|
"admission requires both a manifest and its SHA-256 pin",
|
|
)
|
|
if not isinstance(manifest_path, Path) or not manifest_path.is_file():
|
|
raise ReconstructionError("missing_admission_manifest", "admission manifest is not an existing regular file")
|
|
|
|
manifest_path = manifest_path.resolve()
|
|
manifest_sha256 = _assert_file_hash(manifest_path, manifest_pin, "--admission-manifest-sha256")
|
|
manifest = _require_exact_keys(
|
|
_load_json_object(manifest_path, "admission manifest"),
|
|
frozenset(
|
|
{
|
|
"artifact",
|
|
"contract_version",
|
|
"ledger_sha256",
|
|
"reviewed_by_handle",
|
|
"reviewed_by_agent_id",
|
|
"reviewed_at",
|
|
"review_note",
|
|
"entries",
|
|
}
|
|
),
|
|
"admission manifest",
|
|
)
|
|
if manifest["artifact"] != ADMISSION_ARTIFACT or manifest["contract_version"] != ADMISSION_CONTRACT_VERSION:
|
|
raise ReconstructionError(
|
|
"unsupported_admission_contract",
|
|
"admission manifest is not the supported v1 contract",
|
|
)
|
|
if _validate_sha256(manifest["ledger_sha256"], "admission manifest ledger_sha256") != ledger_sha256:
|
|
raise ReconstructionError("admission_ledger_mismatch", "admission manifest does not select this exact ledger")
|
|
|
|
reviewer_handle = manifest["reviewed_by_handle"]
|
|
review_note = manifest["review_note"]
|
|
if reviewer_handle != CANONICAL_REVIEWER_HANDLE:
|
|
raise ReconstructionError(
|
|
"invalid_admission_review",
|
|
"admission reviewer handle must equal m3taversal exactly",
|
|
)
|
|
if not isinstance(review_note, str) or not review_note.strip():
|
|
raise ReconstructionError("invalid_admission_review", "admission review note must be non-empty")
|
|
try:
|
|
reviewer_agent_id = str(uuid.UUID(str(manifest["reviewed_by_agent_id"])))
|
|
reviewed_at = replay_receipt.canonical_utc_timestamp(
|
|
manifest["reviewed_at"],
|
|
path="admission manifest reviewed_at",
|
|
)
|
|
except (AttributeError, TypeError, ValueError) as exc:
|
|
raise ReconstructionError(
|
|
"invalid_admission_review",
|
|
"admission review requires one reviewer UUID and timezone-aware timestamp",
|
|
) from exc
|
|
|
|
raw_entries = manifest["entries"]
|
|
if not isinstance(raw_entries, list) or len(raw_entries) != len(entries):
|
|
raise ReconstructionError(
|
|
"admission_selection_mismatch",
|
|
"admission manifest must select every entry in the supplied clean-genesis ledger exactly once",
|
|
)
|
|
selected: list[dict[str, Any]] = []
|
|
for index, (raw_entry, entry) in enumerate(zip(raw_entries, entries, strict=True), 1):
|
|
selected_entry = _require_exact_keys(
|
|
raw_entry,
|
|
frozenset(
|
|
{
|
|
"sequence",
|
|
"proposal_id",
|
|
"material_sha256",
|
|
"replay_material_sha256",
|
|
"admission_scope",
|
|
"basis",
|
|
}
|
|
),
|
|
f"admission manifest entries[{index - 1}]",
|
|
)
|
|
if entry.contract_version != 3:
|
|
raise ReconstructionError(
|
|
"legacy_admission_forbidden",
|
|
"only source-backed V3 entries may enter an admitted clean genesis",
|
|
)
|
|
expected = {
|
|
"sequence": entry.sequence,
|
|
"proposal_id": entry.applied_proposal["id"],
|
|
"material_sha256": entry.material_sha256,
|
|
"replay_material_sha256": entry.replay_material_sha256,
|
|
}
|
|
if any(selected_entry[field] != value for field, value in expected.items()):
|
|
raise ReconstructionError(
|
|
"admission_selection_mismatch",
|
|
f"admission manifest entry {index} does not pin the exact ledger material",
|
|
)
|
|
scope = selected_entry["admission_scope"]
|
|
basis = selected_entry["basis"]
|
|
if scope not in ADMISSION_SCOPES or basis not in ADMISSION_BASES:
|
|
raise ReconstructionError(
|
|
"invalid_admission_classification",
|
|
f"admission manifest entry {index} has an unsupported scope or basis",
|
|
)
|
|
selected.append({**expected, "admission_scope": scope, "basis": basis})
|
|
|
|
return (
|
|
{
|
|
"status": "review_metadata_supplied",
|
|
"canonical_retrieval_eligible": False,
|
|
"reason": "operator-supplied review metadata is hash-pinned but not authenticated",
|
|
"artifact": ADMISSION_ARTIFACT,
|
|
"contract_version": ADMISSION_CONTRACT_VERSION,
|
|
"review_metadata_provenance": REVIEW_METADATA_PROVENANCE,
|
|
"cryptographically_authenticated": False,
|
|
"manifest_sha256": manifest_sha256,
|
|
"ledger_sha256": ledger_sha256,
|
|
"reviewed_by_handle": reviewer_handle.strip(),
|
|
"reviewed_by_agent_id": reviewer_agent_id,
|
|
"reviewed_at": reviewed_at,
|
|
"review_note_sha256": _sha256_text(review_note),
|
|
"entry_count": len(selected),
|
|
"entries": selected,
|
|
},
|
|
manifest_path,
|
|
manifest_sha256,
|
|
manifest_path.stat().st_size,
|
|
)
|
|
|
|
|
|
def _validate_proposal_rows(
|
|
approved: dict[str, Any],
|
|
applied: dict[str, Any],
|
|
approval: dict[str, Any],
|
|
receipt_proposal: dict[str, Any],
|
|
) -> None:
|
|
_require_exact_keys(approved, PROPOSAL_FIELDS, "material.approved_proposal")
|
|
_require_exact_keys(applied, PROPOSAL_FIELDS, "material.applied_proposal")
|
|
_require_exact_keys(approval, APPROVAL_FIELDS, "material.approval_snapshot")
|
|
|
|
try:
|
|
proposal_id = str(uuid.UUID(str(applied["id"])))
|
|
except (TypeError, ValueError, AttributeError) as exc:
|
|
raise ReconstructionError("invalid_material", "material proposal id must be a UUID") from exc
|
|
|
|
if approved["id"] != proposal_id or approval["proposal_id"] != proposal_id:
|
|
raise ReconstructionError("material_mismatch", "proposal and approval ids do not match")
|
|
if approved["status"] != "approved" or applied["status"] != "applied":
|
|
raise ReconstructionError(
|
|
"unsupported_proposal_state",
|
|
"material requires one exact approved row and one exact applied row",
|
|
)
|
|
if any(approved[field] is not None for field in ("applied_by_handle", "applied_by_agent_id", "applied_at")):
|
|
raise ReconstructionError(
|
|
"unsupported_proposal_state", "approved proposal material already contains apply metadata"
|
|
)
|
|
if not applied["applied_by_handle"] or not applied["applied_by_agent_id"] or not applied["applied_at"]:
|
|
raise ReconstructionError("invalid_material", "applied proposal material lacks exact apply metadata")
|
|
if not approved["reviewed_by_agent_id"] or not approval["reviewed_by_db_role"]:
|
|
raise ReconstructionError(
|
|
"invalid_material", "strict replay requires a reviewer agent id and reviewer database role"
|
|
)
|
|
|
|
immutable_fields = PROPOSAL_FIELDS - PROPOSAL_TRANSITION_FIELDS
|
|
changed_immutable = sorted(field for field in immutable_fields if approved[field] != applied[field])
|
|
if changed_immutable:
|
|
raise ReconstructionError(
|
|
"material_mismatch",
|
|
f"proposal changed outside the guarded apply transition: {changed_immutable}",
|
|
)
|
|
|
|
approval_projection = {
|
|
"proposal_id": approved["id"],
|
|
"proposal_type": approved["proposal_type"],
|
|
"payload": approved["payload"],
|
|
"reviewed_by_handle": approved["reviewed_by_handle"],
|
|
"reviewed_by_agent_id": approved["reviewed_by_agent_id"],
|
|
"reviewed_at": approved["reviewed_at"],
|
|
"review_note": approved["review_note"],
|
|
}
|
|
for field, expected in approval_projection.items():
|
|
if approval[field] != expected:
|
|
raise ReconstructionError(
|
|
"material_mismatch", f"approval snapshot does not match approved proposal field {field}"
|
|
)
|
|
|
|
envelope_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",
|
|
)
|
|
try:
|
|
envelope_mismatch = any(
|
|
replay_receipt.normalize_typed_value(
|
|
field,
|
|
receipt_proposal.get(field),
|
|
path=f"replay_receipt.proposal.{field}",
|
|
)
|
|
!= replay_receipt.normalize_typed_value(
|
|
field,
|
|
applied[field],
|
|
path=f"material.applied_proposal.{field}",
|
|
)
|
|
for field in envelope_fields
|
|
)
|
|
except ValueError as exc:
|
|
raise ReconstructionError(
|
|
"invalid_material", "replay receipt proposal envelope contains an invalid typed value"
|
|
) from exc
|
|
if envelope_mismatch:
|
|
raise ReconstructionError(
|
|
"material_mismatch", "replay receipt proposal envelope does not match applied proposal row"
|
|
)
|
|
|
|
|
|
def _parse_aware_timestamp(value: Any, field: str, *, code: str = "invalid_mutating_receipt") -> datetime:
|
|
if not isinstance(value, str) or not value:
|
|
raise ReconstructionError(code, f"{field} must be an ISO-8601 timestamp")
|
|
try:
|
|
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
|
|
except ValueError as exc:
|
|
raise ReconstructionError(code, f"{field} must be an ISO-8601 timestamp") from exc
|
|
if parsed.tzinfo is None:
|
|
raise ReconstructionError(code, f"{field} must include a timezone")
|
|
return parsed
|
|
|
|
|
|
def _validated_uuid_list(value: Any, field: str, *, code: str = "invalid_mutating_state") -> list[str]:
|
|
if not isinstance(value, list):
|
|
raise ReconstructionError(code, f"{field} must be a UUID list")
|
|
normalized: list[str] = []
|
|
for item in value:
|
|
try:
|
|
normalized.append(str(uuid.UUID(str(item))))
|
|
except (TypeError, ValueError, AttributeError) as exc:
|
|
raise ReconstructionError(code, f"{field} must contain only UUIDs") from exc
|
|
if len(normalized) != len(set(normalized)):
|
|
raise ReconstructionError(code, f"{field} contains duplicate UUIDs")
|
|
return normalized
|
|
|
|
|
|
def _validate_revise_strategy_receipt_rows(
|
|
proposal: dict[str, Any], canonical_rows: dict[str, Any]
|
|
) -> tuple[dict[str, Any], list[dict[str, Any]]]:
|
|
payload = proposal.get("payload")
|
|
apply_payload = payload.get("apply_payload") if isinstance(payload, dict) else None
|
|
if not isinstance(apply_payload, dict):
|
|
raise ReconstructionError("invalid_mutating_receipt", "revise_strategy receipt requires a strict apply payload")
|
|
try:
|
|
agent_id = str(uuid.UUID(str(apply_payload.get("agent_id"))))
|
|
except (TypeError, ValueError, AttributeError) as exc:
|
|
raise ReconstructionError(
|
|
"invalid_mutating_receipt", "revise_strategy receipt requires one agent UUID"
|
|
) from exc
|
|
|
|
strategies = canonical_rows.get("strategies")
|
|
nodes = canonical_rows.get("strategy_nodes")
|
|
if not isinstance(strategies, list) or len(strategies) != 1 or not isinstance(strategies[0], dict):
|
|
raise ReconstructionError("invalid_mutating_receipt", "revise_strategy receipt must pin one fresh strategy row")
|
|
expected_nodes = apply_payload.get("strategy_nodes")
|
|
if (
|
|
not isinstance(expected_nodes, list)
|
|
or not isinstance(nodes, list)
|
|
or len(nodes) != len(expected_nodes)
|
|
or any(not isinstance(row, dict) for row in nodes)
|
|
):
|
|
raise ReconstructionError(
|
|
"invalid_mutating_receipt", "revise_strategy receipt must pin every fresh strategy node row"
|
|
)
|
|
|
|
strategy = strategies[0]
|
|
if frozenset(strategy) != STRATEGY_REQUIRED_FIELDS:
|
|
raise ReconstructionError(
|
|
"invalid_mutating_receipt", "revise_strategy receipt strategy row fields are not canonical"
|
|
)
|
|
if any(frozenset(row) != STRATEGY_NODE_REQUIRED_FIELDS for row in nodes):
|
|
raise ReconstructionError(
|
|
"invalid_mutating_receipt", "revise_strategy receipt strategy node row fields are not canonical"
|
|
)
|
|
strategy_id = _validated_uuid_list([strategy.get("id")], "receipt strategy id", code="invalid_mutating_receipt")[0]
|
|
node_ids = _validated_uuid_list(
|
|
[row.get("id") for row in nodes],
|
|
"receipt strategy node ids",
|
|
code="invalid_mutating_receipt",
|
|
)
|
|
if strategy.get("agent_id") != agent_id or any(row.get("agent_id") != agent_id for row in nodes):
|
|
raise ReconstructionError(
|
|
"invalid_mutating_receipt", "revise_strategy receipt rows do not belong to the payload agent"
|
|
)
|
|
if strategy.get("active") is not True or any(row.get("status") != "active" for row in nodes):
|
|
raise ReconstructionError(
|
|
"invalid_mutating_receipt", "revise_strategy receipt must pin the fresh active poststate"
|
|
)
|
|
version = strategy.get("version")
|
|
if isinstance(version, bool) or not isinstance(version, int) or version < 1:
|
|
raise ReconstructionError(
|
|
"invalid_mutating_receipt", "revise_strategy receipt strategy version must be positive"
|
|
)
|
|
|
|
transaction_timestamp = strategy.get("created_at")
|
|
transaction_time = _parse_aware_timestamp(transaction_timestamp, "receipt strategy created_at")
|
|
for row in nodes:
|
|
if row.get("created_at") != transaction_timestamp or row.get("updated_at") != transaction_timestamp:
|
|
raise ReconstructionError(
|
|
"invalid_mutating_receipt",
|
|
"revise_strategy fresh strategy and node timestamps must share one transaction timestamp",
|
|
)
|
|
reviewed_time = _parse_aware_timestamp(proposal.get("reviewed_at"), "receipt proposal reviewed_at")
|
|
applied_time = _parse_aware_timestamp(proposal.get("applied_at"), "receipt proposal applied_at")
|
|
if reviewed_time > applied_time:
|
|
raise ReconstructionError("invalid_mutating_receipt", "revise_strategy proposal approval is after apply time")
|
|
if transaction_time < reviewed_time:
|
|
raise ReconstructionError(
|
|
"invalid_mutating_receipt", "revise_strategy transaction timestamp is before proposal approval"
|
|
)
|
|
if transaction_time > applied_time:
|
|
raise ReconstructionError(
|
|
"invalid_mutating_receipt", "revise_strategy transaction timestamp is after proposal apply time"
|
|
)
|
|
|
|
return {**strategy, "id": strategy_id}, [
|
|
{**row, "id": normalized_id} for row, normalized_id in zip(nodes, node_ids, strict=True)
|
|
]
|
|
|
|
|
|
def _validate_entry_material(
|
|
material: dict[str, Any],
|
|
*,
|
|
expected_sequence: int,
|
|
expected_replay_hash: str,
|
|
material_path: Path,
|
|
material_sha256: str,
|
|
) -> LedgerEntry:
|
|
expected_keys = frozenset(
|
|
{
|
|
"artifact",
|
|
"contract_version",
|
|
"sequence",
|
|
"approved_proposal",
|
|
"approval_snapshot",
|
|
"applied_proposal",
|
|
"replay_receipt",
|
|
}
|
|
)
|
|
_require_exact_keys(material, expected_keys, "ledger entry material")
|
|
if material["artifact"] != MATERIAL_ARTIFACT or material["contract_version"] != MATERIAL_CONTRACT_VERSION:
|
|
raise ReconstructionError("unsupported_material_contract", "ledger entry material is not v1")
|
|
if material["sequence"] != expected_sequence:
|
|
raise ReconstructionError("ledger_order_mismatch", "material sequence does not match ledger order")
|
|
|
|
receipt = material["replay_receipt"]
|
|
if not isinstance(receipt, dict):
|
|
raise ReconstructionError("invalid_material", "material.replay_receipt must be an object")
|
|
proposal = receipt.get("proposal")
|
|
if not isinstance(proposal, dict):
|
|
raise ReconstructionError("invalid_material", "replay receipt has no proposal object")
|
|
proposal_type = proposal.get("proposal_type")
|
|
if proposal_type not in SUPPORTED_PROPOSAL_TYPES:
|
|
raise ReconstructionError(
|
|
"unsupported_proposal_type", "ledger entry proposal type is not supported by v1 reconstruction"
|
|
)
|
|
if receipt.get("artifact") != "kb_apply_replay_receipt" or receipt.get("replay_ready") is not True:
|
|
raise ReconstructionError("invalid_replay_receipt", "ledger entry is not a replay-ready apply receipt")
|
|
if receipt.get("receipt_contract_version") != replay_receipt.RECEIPT_CONTRACT_VERSION:
|
|
raise ReconstructionError("unsupported_receipt_contract", "replay receipt contract version is unsupported")
|
|
|
|
apply_metadata = receipt.get("apply_engine")
|
|
canonical_rows = receipt.get("canonical_rows")
|
|
if not isinstance(apply_metadata, dict) or not isinstance(canonical_rows, dict):
|
|
raise ReconstructionError("invalid_replay_receipt", "replay receipt is missing apply or row material")
|
|
if proposal_type == "revise_strategy":
|
|
_validate_revise_strategy_receipt_rows(proposal, canonical_rows)
|
|
legacy_receipt = False
|
|
try:
|
|
rebuilt = replay_receipt.build_replay_receipt(
|
|
proposal,
|
|
canonical_rows,
|
|
apply_sql_sha256=apply_metadata.get("apply_sql_sha256"),
|
|
apply_sql_source=apply_metadata.get("source"),
|
|
exported_at_utc=receipt.get("exported_at_utc"),
|
|
)
|
|
except (KeyError, TypeError, ValueError):
|
|
legacy_receipt = True
|
|
try:
|
|
rebuilt = replay_exporter.validate_worker_replay_receipt(
|
|
receipt,
|
|
expected_proposal_id=material["applied_proposal"].get("id"),
|
|
)
|
|
except (KeyError, TypeError, ValueError, replay_exporter.ExportError) as exc:
|
|
raise ReconstructionError(
|
|
"invalid_replay_receipt", "replay receipt failed strict payload and exact-row validation"
|
|
) from exc
|
|
|
|
if not legacy_receipt and receipt.get("hashes") != rebuilt.get("hashes"):
|
|
raise ReconstructionError("replay_receipt_hash_mismatch", "replay receipt hashes are not recoverable")
|
|
if receipt.get("expected_row_counts") != rebuilt.get("expected_row_counts"):
|
|
raise ReconstructionError("replay_receipt_count_mismatch", "replay receipt expected counts are invalid")
|
|
if receipt.get("actual_row_counts") != rebuilt.get("actual_row_counts"):
|
|
raise ReconstructionError("replay_receipt_count_mismatch", "replay receipt actual counts are invalid")
|
|
receipt_hashes = receipt.get("hashes")
|
|
if not isinstance(receipt_hashes, dict):
|
|
raise ReconstructionError("replay_receipt_hash_mismatch", "replay receipt hashes are not recoverable")
|
|
actual_replay_hash = receipt_hashes.get("replay_material_sha256")
|
|
if actual_replay_hash != expected_replay_hash:
|
|
raise ReconstructionError(
|
|
"replay_material_hash_mismatch", "ledger replay-material pin does not match the private receipt"
|
|
)
|
|
receipt = rebuilt
|
|
proposal = receipt["proposal"]
|
|
|
|
approved = material["approved_proposal"]
|
|
approval = material["approval_snapshot"]
|
|
applied = material["applied_proposal"]
|
|
_validate_proposal_rows(approved, applied, approval, proposal)
|
|
try:
|
|
current_apply_sql = apply_engine.build_apply_sql(proposal, applied["applied_by_handle"])
|
|
except (KeyError, TypeError, ValueError) as exc:
|
|
raise ReconstructionError(
|
|
"current_apply_engine_rejected_material",
|
|
"current guarded apply engine rejected the strict replay material",
|
|
) from exc
|
|
current_apply_sql_sha256 = _sha256_text(current_apply_sql)
|
|
if proposal_type == "revise_strategy" and (
|
|
apply_metadata.get("source") != "exact_executed_sql"
|
|
or apply_metadata.get("apply_sql_sha256") != current_apply_sql_sha256
|
|
):
|
|
raise ReconstructionError(
|
|
"mutating_apply_engine_mismatch",
|
|
"revise_strategy replay requires the exact executed SQL to match the current guarded apply engine",
|
|
)
|
|
|
|
return LedgerEntry(
|
|
sequence=expected_sequence,
|
|
material_path=material_path,
|
|
material_sha256=material_sha256,
|
|
replay_material_sha256=actual_replay_hash,
|
|
contract_version=_proposal_contract_version(proposal),
|
|
approved_proposal=approved,
|
|
approval_snapshot=approval,
|
|
applied_proposal=applied,
|
|
receipt=receipt,
|
|
current_apply_sql=current_apply_sql,
|
|
current_apply_sql_sha256=current_apply_sql_sha256,
|
|
)
|
|
|
|
|
|
def load_bundle(args: argparse.Namespace) -> ReconstructionBundle:
|
|
canonical_rebuild.validate_custom_dump(args.genesis_dump)
|
|
ledger_sha256 = _assert_file_hash(args.ledger, args.ledger_sha256, "--ledger-sha256")
|
|
ledger = _load_json_object(args.ledger, "ledger")
|
|
top_level_keys = frozenset({"artifact", "contract_version", "engine", "genesis", "entries", "final_parity"})
|
|
_require_exact_keys(ledger, top_level_keys, "ledger")
|
|
if ledger["artifact"] != LEDGER_ARTIFACT or ledger["contract_version"] != LEDGER_CONTRACT_VERSION:
|
|
raise ReconstructionError("unsupported_ledger_contract", "ledger is not the v1 genesis-plus-ledger contract")
|
|
|
|
actual_engine_hashes = _engine_hashes()
|
|
expected_engine = _require_exact_keys(ledger["engine"], frozenset(actual_engine_hashes), "ledger.engine")
|
|
for field, actual_hash in actual_engine_hashes.items():
|
|
if _validate_sha256(expected_engine[field], f"ledger.engine.{field}") != actual_hash:
|
|
raise ReconstructionError("engine_hash_mismatch", f"ledger engine pin does not match current {field}")
|
|
|
|
genesis = _require_exact_keys(
|
|
ledger["genesis"], frozenset({"dump_sha256", "parity_manifest_sha256"}), "ledger.genesis"
|
|
)
|
|
genesis_dump_sha256 = _assert_file_hash(args.genesis_dump, genesis["dump_sha256"], "ledger.genesis.dump_sha256")
|
|
genesis_manifest_sha256 = _assert_file_hash(
|
|
args.genesis_manifest,
|
|
genesis["parity_manifest_sha256"],
|
|
"ledger.genesis.parity_manifest_sha256",
|
|
)
|
|
try:
|
|
genesis_manifest = load_manifest(args.genesis_manifest)
|
|
except (OSError, ValueError, KeyError, TypeError, json.JSONDecodeError) as exc:
|
|
raise ReconstructionError("invalid_genesis_manifest", "genesis parity manifest is invalid") from exc
|
|
|
|
final_spec = _require_exact_keys(ledger["final_parity"], frozenset({"manifest", "sha256"}), "ledger.final_parity")
|
|
final_manifest_path = _resolve_artifact(args.ledger, final_spec["manifest"], "final parity manifest")
|
|
final_manifest_sha256 = _assert_file_hash(final_manifest_path, final_spec["sha256"], "ledger.final_parity.sha256")
|
|
try:
|
|
final_manifest = load_manifest(final_manifest_path)
|
|
except (OSError, ValueError, KeyError, TypeError, json.JSONDecodeError) as exc:
|
|
raise ReconstructionError("invalid_final_manifest", "final parity manifest is invalid") from exc
|
|
|
|
raw_entries = ledger["entries"]
|
|
if not isinstance(raw_entries, list):
|
|
raise ReconstructionError("invalid_contract", "ledger.entries must be an ordered list")
|
|
entries: list[LedgerEntry] = []
|
|
protected_paths = {
|
|
args.genesis_dump.resolve(),
|
|
args.genesis_manifest.resolve(),
|
|
args.ledger.resolve(),
|
|
final_manifest_path,
|
|
*FIXED_ENGINE_PATHS,
|
|
}
|
|
seen_proposal_ids: set[str] = set()
|
|
for expected_sequence, raw_entry in enumerate(raw_entries, 1):
|
|
entry_spec = _require_exact_keys(
|
|
raw_entry,
|
|
frozenset({"sequence", "material", "sha256", "replay_material_sha256"}),
|
|
f"ledger.entries[{expected_sequence - 1}]",
|
|
)
|
|
if entry_spec["sequence"] != expected_sequence:
|
|
raise ReconstructionError(
|
|
"ledger_order_mismatch", "ledger entry sequences must be contiguous and start at 1"
|
|
)
|
|
material_path = _resolve_artifact(
|
|
args.ledger, entry_spec["material"], f"ledger entry {expected_sequence} material"
|
|
)
|
|
material_sha256 = _assert_file_hash(
|
|
material_path, entry_spec["sha256"], f"ledger entry {expected_sequence} sha256"
|
|
)
|
|
expected_replay_hash = _validate_sha256(
|
|
entry_spec["replay_material_sha256"],
|
|
f"ledger entry {expected_sequence} replay_material_sha256",
|
|
)
|
|
entry = _validate_entry_material(
|
|
_load_json_object(material_path, f"ledger entry {expected_sequence} material"),
|
|
expected_sequence=expected_sequence,
|
|
expected_replay_hash=expected_replay_hash,
|
|
material_path=material_path,
|
|
material_sha256=material_sha256,
|
|
)
|
|
proposal_id = entry.applied_proposal["id"]
|
|
if proposal_id in seen_proposal_ids:
|
|
raise ReconstructionError("duplicate_proposal", "ledger contains the same proposal more than once")
|
|
seen_proposal_ids.add(proposal_id)
|
|
entries.append(entry)
|
|
protected_paths.add(material_path)
|
|
|
|
_validate_contract_order(entries)
|
|
admission, admission_path, admission_sha256, admission_bytes = _load_admission_manifest(
|
|
args,
|
|
ledger_sha256=ledger_sha256,
|
|
entries=entries,
|
|
)
|
|
if admission_path is not None:
|
|
protected_paths.add(admission_path)
|
|
|
|
if args.output.resolve() in protected_paths:
|
|
raise ReconstructionError("unsafe_output_path", "--output must not overwrite any reconstruction input")
|
|
|
|
return ReconstructionBundle(
|
|
ledger_sha256=ledger_sha256,
|
|
ledger_bytes=args.ledger.stat().st_size,
|
|
genesis_dump_sha256=genesis_dump_sha256,
|
|
genesis_dump_bytes=args.genesis_dump.stat().st_size,
|
|
genesis_manifest=genesis_manifest,
|
|
genesis_manifest_sha256=genesis_manifest_sha256,
|
|
genesis_manifest_bytes=args.genesis_manifest.stat().st_size,
|
|
final_manifest_path=final_manifest_path,
|
|
final_manifest=final_manifest,
|
|
final_manifest_sha256=final_manifest_sha256,
|
|
final_manifest_bytes=final_manifest_path.stat().st_size,
|
|
engine_hashes=actual_engine_hashes,
|
|
entries=tuple(entries),
|
|
admission=admission,
|
|
admission_manifest_path=admission_path,
|
|
admission_manifest_sha256=admission_sha256,
|
|
admission_manifest_bytes=admission_bytes,
|
|
protected_paths=frozenset(protected_paths),
|
|
)
|
|
|
|
|
|
def _jsonb_literal(value: Any) -> str:
|
|
return apply_engine.sql_literal(_canonical_json(value)) + "::jsonb"
|
|
|
|
|
|
def _approval_seed_row(approved_proposal: dict[str, Any], approval_snapshot: dict[str, Any]) -> dict[str, Any]:
|
|
return {
|
|
**approval_snapshot,
|
|
"approved_proposal_snapshot": approved_proposal,
|
|
}
|
|
|
|
|
|
def build_seed_sql(entry: LedgerEntry) -> str:
|
|
approval_seed = _approval_seed_row(entry.approved_proposal, entry.approval_snapshot)
|
|
statements = [
|
|
"begin;",
|
|
"set local standard_conforming_strings = on;",
|
|
"insert into kb_stage.kb_proposals",
|
|
"select seeded.* from jsonb_populate_record(",
|
|
f" null::kb_stage.kb_proposals, {_jsonb_literal(entry.approved_proposal)}",
|
|
") seeded",
|
|
"on conflict (id) do nothing;",
|
|
"insert into kb_stage.kb_proposal_approvals",
|
|
"select seeded.*",
|
|
"from kb_stage.kb_proposals proposal",
|
|
"cross join lateral jsonb_populate_record(",
|
|
" null::kb_stage.kb_proposal_approvals,",
|
|
f" jsonb_set({_jsonb_literal(approval_seed)}, '{{approved_proposal_snapshot}}', to_jsonb(proposal), true)",
|
|
") seeded",
|
|
f"where proposal.id = {apply_engine.sql_literal(entry.approved_proposal['id'])}::uuid",
|
|
"on conflict (proposal_id) do nothing;",
|
|
]
|
|
if entry.contract_version != 3:
|
|
rows = entry.receipt["canonical_rows"]
|
|
for table in CANONICAL_TABLE_ORDER:
|
|
table_rows = rows.get(table)
|
|
if not table_rows:
|
|
continue
|
|
statements.extend(
|
|
[
|
|
f"insert into public.{table}",
|
|
"select seeded.* from jsonb_populate_recordset(",
|
|
f" null::public.{table}, {_jsonb_literal(table_rows)}",
|
|
") seeded",
|
|
"on conflict do nothing;",
|
|
]
|
|
)
|
|
statements.append("commit;")
|
|
return "\n".join(statements) + "\n"
|
|
|
|
|
|
def _uuid_membership_sql(column: str, values: list[str], *, negate: bool = False) -> str:
|
|
if not values:
|
|
return "true" if negate else "false"
|
|
rendered = ", ".join(f"{apply_engine.sql_literal(value)}::uuid" for value in values)
|
|
operator = "not in" if negate else "in"
|
|
return f"{column} {operator} ({rendered})"
|
|
|
|
|
|
def _uuid_array_sql(values: list[str]) -> str:
|
|
if not values:
|
|
return "array[]::uuid[]"
|
|
rendered = ", ".join(f"{apply_engine.sql_literal(value)}::uuid" for value in values)
|
|
return f"array[{rendered}]::uuid[]"
|
|
|
|
|
|
def build_revise_strategy_prestate_sql(entry: LedgerEntry) -> str:
|
|
apply_payload = entry.receipt["proposal"]["payload"]["apply_payload"]
|
|
agent = apply_engine.sql_literal(apply_payload["agent_id"])
|
|
return f"""with strategy_state as (
|
|
select coalesce(jsonb_agg(s.id::text order by s.id::text), '[]'::jsonb) as strategy_ids,
|
|
coalesce(
|
|
jsonb_agg(s.id::text order by s.id::text) filter (where s.active),
|
|
'[]'::jsonb
|
|
) as active_strategy_ids,
|
|
coalesce(max(s.version), 0) as max_strategy_version
|
|
from public.strategies s
|
|
where s.agent_id = {agent}::uuid
|
|
), node_state as (
|
|
select coalesce(jsonb_agg(n.id::text order by n.id::text), '[]'::jsonb) as strategy_node_ids,
|
|
coalesce(
|
|
jsonb_agg(n.id::text order by n.id::text) filter (where n.status <> 'retired'),
|
|
'[]'::jsonb
|
|
) as non_retired_strategy_node_ids
|
|
from public.strategy_nodes n
|
|
where n.agent_id = {agent}::uuid
|
|
)
|
|
select jsonb_build_object(
|
|
'strategy_ids', s.strategy_ids,
|
|
'active_strategy_ids', s.active_strategy_ids,
|
|
'max_strategy_version', s.max_strategy_version,
|
|
'strategy_node_ids', n.strategy_node_ids,
|
|
'non_retired_strategy_node_ids', n.non_retired_strategy_node_ids
|
|
)::text
|
|
from strategy_state s cross join node_state n;
|
|
"""
|
|
|
|
|
|
def _validate_revise_strategy_prestate(value: Any) -> dict[str, Any]:
|
|
expected = frozenset(
|
|
{
|
|
"strategy_ids",
|
|
"active_strategy_ids",
|
|
"max_strategy_version",
|
|
"strategy_node_ids",
|
|
"non_retired_strategy_node_ids",
|
|
}
|
|
)
|
|
state = _require_exact_keys(value, expected, "revise_strategy prestate")
|
|
strategy_ids = _validated_uuid_list(state["strategy_ids"], "prestate strategy ids")
|
|
active_strategy_ids = _validated_uuid_list(state["active_strategy_ids"], "prestate active strategy ids")
|
|
strategy_node_ids = _validated_uuid_list(state["strategy_node_ids"], "prestate strategy node ids")
|
|
non_retired_node_ids = _validated_uuid_list(
|
|
state["non_retired_strategy_node_ids"], "prestate non-retired strategy node ids"
|
|
)
|
|
max_version = state["max_strategy_version"]
|
|
if isinstance(max_version, bool) or not isinstance(max_version, int) or max_version < 0:
|
|
raise ReconstructionError(
|
|
"invalid_mutating_state", "prestate maximum strategy version must be a non-negative integer"
|
|
)
|
|
if len(active_strategy_ids) > 1 or not set(active_strategy_ids).issubset(strategy_ids):
|
|
raise ReconstructionError(
|
|
"invalid_mutating_state", "prestate active strategies violate the one-active invariant"
|
|
)
|
|
if not set(non_retired_node_ids).issubset(strategy_node_ids):
|
|
raise ReconstructionError(
|
|
"invalid_mutating_state", "prestate non-retired strategy nodes are not a subset of all nodes"
|
|
)
|
|
return {
|
|
"strategy_ids": strategy_ids,
|
|
"active_strategy_ids": active_strategy_ids,
|
|
"max_strategy_version": max_version,
|
|
"strategy_node_ids": strategy_node_ids,
|
|
"non_retired_strategy_node_ids": non_retired_node_ids,
|
|
}
|
|
|
|
|
|
def build_revise_strategy_generated_delta_sql(entry: LedgerEntry, prestate: dict[str, Any]) -> str:
|
|
state = _validate_revise_strategy_prestate(prestate)
|
|
apply_payload = entry.receipt["proposal"]["payload"]["apply_payload"]
|
|
agent = apply_engine.sql_literal(apply_payload["agent_id"])
|
|
strategy_filter = _uuid_membership_sql("s.id", state["strategy_ids"], negate=True)
|
|
node_filter = _uuid_membership_sql("n.id", state["strategy_node_ids"], negate=True)
|
|
return f"""select jsonb_build_object(
|
|
'strategies', coalesce((
|
|
select jsonb_agg(to_jsonb(s) order by s.id::text)
|
|
from public.strategies s
|
|
where s.agent_id = {agent}::uuid and {strategy_filter}
|
|
), '[]'::jsonb),
|
|
'strategy_nodes', coalesce((
|
|
select jsonb_agg(to_jsonb(n) order by n.id::text)
|
|
from public.strategy_nodes n
|
|
where n.agent_id = {agent}::uuid and {node_filter}
|
|
), '[]'::jsonb)
|
|
)::text;
|
|
"""
|
|
|
|
|
|
def _validate_revise_strategy_generated_delta(
|
|
entry: LedgerEntry,
|
|
prestate: dict[str, Any],
|
|
generated_rows: Any,
|
|
) -> tuple[dict[str, Any], list[dict[str, Any]]]:
|
|
state = _validate_revise_strategy_prestate(prestate)
|
|
if not isinstance(generated_rows, dict):
|
|
raise ReconstructionError("invalid_mutating_delta", "revise_strategy generated delta must be a row object")
|
|
try:
|
|
replay_receipt.build_replay_receipt(
|
|
entry.receipt["proposal"],
|
|
generated_rows,
|
|
apply_sql_sha256=entry.current_apply_sql_sha256,
|
|
apply_sql_source="exact_executed_sql",
|
|
exported_at_utc=entry.receipt.get("exported_at_utc"),
|
|
)
|
|
except (KeyError, TypeError, ValueError) as exc:
|
|
raise ReconstructionError(
|
|
"invalid_mutating_delta",
|
|
"revise_strategy generated rows do not match the strict payload",
|
|
) from exc
|
|
|
|
strategies = generated_rows.get("strategies")
|
|
nodes = generated_rows.get("strategy_nodes")
|
|
if not isinstance(strategies, list) or len(strategies) != 1 or not isinstance(strategies[0], dict):
|
|
raise ReconstructionError(
|
|
"invalid_mutating_delta", "revise_strategy apply did not generate exactly one strategy"
|
|
)
|
|
if not isinstance(nodes, list) or any(not isinstance(row, dict) for row in nodes):
|
|
raise ReconstructionError("invalid_mutating_delta", "revise_strategy apply generated invalid strategy nodes")
|
|
strategy = strategies[0]
|
|
if strategy.get("version") != state["max_strategy_version"] + 1:
|
|
raise ReconstructionError("invalid_mutating_delta", "revise_strategy generated an unexpected strategy version")
|
|
transaction_timestamp = strategy.get("created_at")
|
|
_parse_aware_timestamp(transaction_timestamp, "generated strategy created_at", code="invalid_mutating_delta")
|
|
if any(
|
|
row.get("created_at") != transaction_timestamp or row.get("updated_at") != transaction_timestamp
|
|
for row in nodes
|
|
):
|
|
raise ReconstructionError(
|
|
"invalid_mutating_delta",
|
|
"revise_strategy generated rows do not share one transaction timestamp",
|
|
)
|
|
_validated_uuid_list([strategy.get("id")], "generated strategy id", code="invalid_mutating_delta")
|
|
_validated_uuid_list(
|
|
[row.get("id") for row in nodes],
|
|
"generated strategy node ids",
|
|
code="invalid_mutating_delta",
|
|
)
|
|
return strategy, nodes
|
|
|
|
|
|
def build_revise_strategy_normalization_sql(
|
|
entry: LedgerEntry,
|
|
prestate: dict[str, Any],
|
|
generated_rows: dict[str, Any],
|
|
) -> str:
|
|
state = _validate_revise_strategy_prestate(prestate)
|
|
generated_strategy, generated_nodes = _validate_revise_strategy_generated_delta(entry, state, generated_rows)
|
|
receipt_strategy, receipt_nodes = _validate_revise_strategy_receipt_rows(
|
|
entry.receipt["proposal"], entry.receipt["canonical_rows"]
|
|
)
|
|
if receipt_strategy["version"] != state["max_strategy_version"] + 1:
|
|
raise ReconstructionError(
|
|
"mutating_receipt_version_mismatch",
|
|
"revise_strategy receipt version does not follow the observed prestate",
|
|
)
|
|
if receipt_strategy["id"] in state["strategy_ids"] or set(row["id"] for row in receipt_nodes).intersection(
|
|
state["strategy_node_ids"]
|
|
):
|
|
raise ReconstructionError(
|
|
"mutating_receipt_id_collision", "revise_strategy receipt IDs collide with the observed prestate"
|
|
)
|
|
|
|
apply_payload = entry.receipt["proposal"]["payload"]["apply_payload"]
|
|
agent = apply_engine.sql_literal(apply_payload["agent_id"])
|
|
generated_strategy_id = str(uuid.UUID(str(generated_strategy["id"])))
|
|
generated_node_ids = [str(uuid.UUID(str(row["id"]))) for row in generated_nodes]
|
|
previous_active = state["active_strategy_ids"]
|
|
changed_node_ids = state["non_retired_strategy_node_ids"]
|
|
transaction_timestamp = apply_engine.sql_literal(receipt_strategy["created_at"])
|
|
generated_node_array = _uuid_array_sql(generated_node_ids)
|
|
previous_active_array = _uuid_array_sql(previous_active)
|
|
changed_node_array = _uuid_array_sql(changed_node_ids)
|
|
|
|
prior_strategy_assertion = ""
|
|
if previous_active:
|
|
prior_strategy_assertion = f"""
|
|
if (select count(*) from public.strategies
|
|
where agent_id = {agent}::uuid and id = any(previous_active_ids) and not active) <> {len(previous_active)} then
|
|
raise exception 'revise_strategy prior active strategy transition mismatch';
|
|
end if;"""
|
|
prior_node_normalization = ""
|
|
if changed_node_ids:
|
|
prior_node_normalization = f"""
|
|
if (select count(*) from public.strategy_nodes
|
|
where agent_id = {agent}::uuid and id = any(changed_node_ids) and status = 'retired') <> {len(changed_node_ids)} then
|
|
raise exception 'revise_strategy prior node transition mismatch';
|
|
end if;
|
|
update public.strategy_nodes
|
|
set status = 'retired', updated_at = {transaction_timestamp}::timestamptz
|
|
where agent_id = {agent}::uuid and id = any(changed_node_ids);
|
|
get diagnostics changed_rows = row_count;
|
|
if changed_rows <> {len(changed_node_ids)} then
|
|
raise exception 'revise_strategy prior node normalization mismatch';
|
|
end if;"""
|
|
|
|
return f"""begin;
|
|
set local standard_conforming_strings = on;
|
|
do $reconstruction$
|
|
declare
|
|
changed_rows integer;
|
|
generated_strategy_id constant uuid := {apply_engine.sql_literal(generated_strategy_id)}::uuid;
|
|
generated_node_ids constant uuid[] := {generated_node_array};
|
|
previous_active_ids constant uuid[] := {previous_active_array};
|
|
changed_node_ids constant uuid[] := {changed_node_array};
|
|
begin{prior_strategy_assertion}
|
|
if exists (
|
|
select 1 from public.strategy_node_anchors
|
|
where from_node_id = any(generated_node_ids) or to_node_id = any(generated_node_ids)
|
|
) then
|
|
raise exception 'revise_strategy generated nodes unexpectedly have anchors';
|
|
end if;
|
|
delete from public.strategy_nodes
|
|
where agent_id = {agent}::uuid and id = any(generated_node_ids);
|
|
get diagnostics changed_rows = row_count;
|
|
if changed_rows <> {len(generated_node_ids)} then
|
|
raise exception 'revise_strategy generated node delta mismatch';
|
|
end if;
|
|
delete from public.strategies
|
|
where agent_id = {agent}::uuid and id = generated_strategy_id;
|
|
get diagnostics changed_rows = row_count;
|
|
if changed_rows <> 1 then
|
|
raise exception 'revise_strategy generated strategy delta mismatch';
|
|
end if;{prior_node_normalization}
|
|
insert into public.strategies
|
|
select seeded.* from jsonb_populate_record(
|
|
null::public.strategies, {_jsonb_literal(receipt_strategy)}
|
|
) seeded;
|
|
insert into public.strategy_nodes
|
|
select seeded.* from jsonb_populate_recordset(
|
|
null::public.strategy_nodes, {_jsonb_literal(receipt_nodes)}
|
|
) seeded;
|
|
if (select count(*) from public.strategies
|
|
where agent_id = {agent}::uuid and active) <> 1 then
|
|
raise exception 'revise_strategy normalized one-active invariant mismatch';
|
|
end if;
|
|
end
|
|
$reconstruction$;
|
|
commit;
|
|
"""
|
|
|
|
|
|
def build_proposal_readback_sql(proposal_id: str) -> str:
|
|
literal = apply_engine.sql_literal(proposal_id)
|
|
return f"""select jsonb_build_object(
|
|
'proposal', (select to_jsonb(p) from kb_stage.kb_proposals p
|
|
where p.id = {literal}::uuid),
|
|
'approval_snapshot', (select to_jsonb(a) - 'approved_proposal_snapshot'
|
|
from kb_stage.kb_proposal_approvals a
|
|
where a.proposal_id = {literal}::uuid),
|
|
'approved_proposal_snapshot', (select a.approved_proposal_snapshot
|
|
from kb_stage.kb_proposal_approvals a
|
|
where a.proposal_id = {literal}::uuid)
|
|
)::text;
|
|
"""
|
|
|
|
|
|
def build_v3_canonical_timestamp_normalization_sql(entry: LedgerEntry) -> str:
|
|
if entry.contract_version != 3:
|
|
raise ReconstructionError(
|
|
"invalid_v3_normalization",
|
|
"canonical V3 timestamp normalization requires a V3 ledger entry",
|
|
)
|
|
proposal_id = entry.applied_proposal["id"]
|
|
table_contracts = {
|
|
"claims": ("accepted_by_proposal_id", "teleo_v3_claim_immutable"),
|
|
"sources": ("accepted_by_proposal_id", "teleo_v3_source_immutable"),
|
|
"claim_evidence": ("accepted_from_proposal_id", "teleo_v3_evidence_immutable"),
|
|
"claim_evidence_assessments": (
|
|
"accepted_by_proposal_id",
|
|
"teleo_v3_assessment_immutable",
|
|
),
|
|
"claim_edges": ("accepted_by_proposal_id", "teleo_v3_edge_immutable"),
|
|
}
|
|
statements = ["begin;", "set local standard_conforming_strings = on;"]
|
|
for table in CANONICAL_TABLE_ORDER:
|
|
rows = entry.receipt["canonical_rows"].get(table) or []
|
|
if not rows:
|
|
continue
|
|
if table not in table_contracts or any("id" not in row for row in rows):
|
|
raise ReconstructionError(
|
|
"invalid_v3_normalization",
|
|
"V3 canonical timestamp rows lack an exact proposal-owned identity",
|
|
)
|
|
timestamp_fields = sorted(TYPED_TIMESTAMP_FIELDS.intersection(*(set(row) for row in rows)))
|
|
if not timestamp_fields:
|
|
continue
|
|
proposal_column, immutable_trigger = table_contracts[table]
|
|
assignments = ",\n ".join(f"{field} = pinned.{field}" for field in timestamp_fields)
|
|
statements.append(
|
|
f"""lock table public.{table} in access exclusive mode;
|
|
do $preflight_{table}$
|
|
begin
|
|
if not exists (
|
|
select 1
|
|
from pg_catalog.pg_trigger trigger
|
|
where trigger.tgrelid = 'public.{table}'::regclass
|
|
and trigger.tgname = '{immutable_trigger}'
|
|
and trigger.tgfoid = 'kb_stage.teleo_v3_guard_immutable_fields()'::regprocedure
|
|
and not trigger.tgisinternal
|
|
and trigger.tgenabled = 'O'
|
|
) then
|
|
raise exception 'V3 {table} immutable guard is not enabled before deterministic normalization';
|
|
end if;
|
|
end
|
|
$preflight_{table}$;
|
|
alter table public.{table} disable trigger {immutable_trigger};
|
|
do $normalize_{table}$
|
|
declare
|
|
changed_rows integer;
|
|
begin
|
|
update public.{table} target
|
|
set {assignments}
|
|
from jsonb_populate_recordset(
|
|
null::public.{table}, {_jsonb_literal(rows)}
|
|
) pinned
|
|
where target.id = pinned.id
|
|
and target.{proposal_column} = {apply_engine.sql_literal(proposal_id)}::uuid;
|
|
get diagnostics changed_rows = row_count;
|
|
if changed_rows <> {len(rows)} then
|
|
raise exception 'deterministic V3 {table} timestamp normalization failed';
|
|
end if;
|
|
end
|
|
$normalize_{table}$;
|
|
alter table public.{table} enable trigger {immutable_trigger};
|
|
do $postflight_{table}$
|
|
begin
|
|
if not exists (
|
|
select 1
|
|
from pg_catalog.pg_trigger trigger
|
|
where trigger.tgrelid = 'public.{table}'::regclass
|
|
and trigger.tgname = '{immutable_trigger}'
|
|
and trigger.tgfoid = 'kb_stage.teleo_v3_guard_immutable_fields()'::regprocedure
|
|
and not trigger.tgisinternal
|
|
and trigger.tgenabled = 'O'
|
|
) then
|
|
raise exception 'V3 {table} immutable guard was not restored after deterministic normalization';
|
|
end if;
|
|
end
|
|
$postflight_{table}$;"""
|
|
)
|
|
statements.append("commit;")
|
|
return "\n".join(statements) + "\n"
|
|
|
|
|
|
def build_applied_timestamp_normalization_sql(entry: LedgerEntry) -> str:
|
|
applied = entry.applied_proposal
|
|
if entry.contract_version == 3:
|
|
return f"""begin;
|
|
lock table kb_stage.kb_proposals in access exclusive mode;
|
|
do $preflight$
|
|
begin
|
|
if not exists (
|
|
select 1
|
|
from pg_catalog.pg_trigger trigger
|
|
where trigger.tgrelid = 'kb_stage.kb_proposals'::regclass
|
|
and trigger.tgname = 'teleo_v3_guard_linked_proposal_decision'
|
|
and trigger.tgfoid = 'kb_stage.teleo_v3_guard_linked_proposal_decision()'::regprocedure
|
|
and not trigger.tgisinternal
|
|
and trigger.tgenabled = 'O'
|
|
) then
|
|
raise exception 'V3 proposal decision guard is not enabled before deterministic normalization';
|
|
end if;
|
|
end
|
|
$preflight$;
|
|
alter table kb_stage.kb_proposals
|
|
disable trigger teleo_v3_guard_linked_proposal_decision;
|
|
do $normalization$
|
|
declare
|
|
changed_rows integer;
|
|
begin
|
|
update kb_stage.kb_proposals
|
|
set applied_at = {apply_engine.sql_literal(applied["applied_at"])}::timestamptz,
|
|
updated_at = {apply_engine.sql_literal(applied["updated_at"])}::timestamptz
|
|
where id = {apply_engine.sql_literal(applied["id"])}::uuid
|
|
and status = 'applied'
|
|
and applied_by_handle = {apply_engine.sql_literal(applied["applied_by_handle"])}
|
|
and applied_by_agent_id is not distinct from
|
|
{apply_engine.sql_literal(applied["applied_by_agent_id"])}::uuid;
|
|
get diagnostics changed_rows = row_count;
|
|
if changed_rows <> 1 then
|
|
raise exception 'deterministic V3 proposal timestamp normalization failed';
|
|
end if;
|
|
end
|
|
$normalization$;
|
|
alter table kb_stage.kb_proposals
|
|
enable trigger teleo_v3_guard_linked_proposal_decision;
|
|
do $postflight$
|
|
begin
|
|
if not exists (
|
|
select 1
|
|
from pg_catalog.pg_trigger trigger
|
|
where trigger.tgrelid = 'kb_stage.kb_proposals'::regclass
|
|
and trigger.tgname = 'teleo_v3_guard_linked_proposal_decision'
|
|
and trigger.tgfoid = 'kb_stage.teleo_v3_guard_linked_proposal_decision()'::regprocedure
|
|
and not trigger.tgisinternal
|
|
and trigger.tgenabled = 'O'
|
|
) then
|
|
raise exception 'V3 proposal decision guard was not restored after deterministic normalization';
|
|
end if;
|
|
end
|
|
$postflight$;
|
|
commit;
|
|
"""
|
|
return f"""do $reconstruction$
|
|
declare
|
|
changed_rows integer;
|
|
begin
|
|
update kb_stage.kb_proposals
|
|
set applied_at = {apply_engine.sql_literal(applied["applied_at"])}::timestamptz,
|
|
updated_at = {apply_engine.sql_literal(applied["updated_at"])}::timestamptz
|
|
where id = {apply_engine.sql_literal(applied["id"])}::uuid
|
|
and status = 'applied'
|
|
and applied_by_handle = {apply_engine.sql_literal(applied["applied_by_handle"])}
|
|
and applied_by_agent_id is not distinct from
|
|
{apply_engine.sql_literal(applied["applied_by_agent_id"])}::uuid;
|
|
get diagnostics changed_rows = row_count;
|
|
if changed_rows <> 1 then
|
|
raise exception 'deterministic proposal timestamp normalization failed';
|
|
end if;
|
|
end
|
|
$reconstruction$;
|
|
"""
|
|
|
|
|
|
def _run_stdin(
|
|
command: list[str],
|
|
*,
|
|
stdin: str,
|
|
timeout: float,
|
|
) -> subprocess.CompletedProcess[str]:
|
|
return subprocess.run(
|
|
command,
|
|
input=stdin,
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
timeout=timeout,
|
|
)
|
|
|
|
|
|
def _diagnostic_fingerprint(completed: subprocess.CompletedProcess[str]) -> str:
|
|
output = (completed.stderr or "") + "\n" + (completed.stdout or "")
|
|
return _sha256_text(output)
|
|
|
|
|
|
def _require_command_success(
|
|
completed: subprocess.CompletedProcess[str],
|
|
*,
|
|
code: str,
|
|
action: str,
|
|
) -> str:
|
|
if completed.returncode != 0:
|
|
fingerprint = _diagnostic_fingerprint(completed)
|
|
raise ReconstructionError(
|
|
code,
|
|
f"{action} failed; private command output withheld; diagnostic_sha256={fingerprint}",
|
|
)
|
|
return completed.stdout
|
|
|
|
|
|
def _psql(
|
|
args: argparse.Namespace,
|
|
container: str,
|
|
sql: str,
|
|
*,
|
|
role: str,
|
|
timeout: float,
|
|
code: str,
|
|
action: str,
|
|
) -> str:
|
|
command = [
|
|
args.docker_bin,
|
|
"exec",
|
|
"-i",
|
|
container,
|
|
"psql",
|
|
"-X",
|
|
"-U",
|
|
role,
|
|
"-h",
|
|
"127.0.0.1",
|
|
"-d",
|
|
args.database,
|
|
"-Atq",
|
|
"-v",
|
|
"ON_ERROR_STOP=1",
|
|
]
|
|
try:
|
|
completed = _run_stdin(command, stdin=sql, timeout=timeout)
|
|
except (OSError, subprocess.TimeoutExpired) as exc:
|
|
raise ReconstructionError(code, f"{action} did not complete; private details withheld") from exc
|
|
return _require_command_success(completed, code=code, action=action)
|
|
|
|
|
|
def _read_json(
|
|
args: argparse.Namespace,
|
|
container: str,
|
|
sql: str,
|
|
*,
|
|
code: str,
|
|
action: str,
|
|
) -> Any:
|
|
raw = _psql(
|
|
args,
|
|
container,
|
|
sql,
|
|
role="postgres",
|
|
timeout=args.command_timeout,
|
|
code=code,
|
|
action=action,
|
|
).strip()
|
|
try:
|
|
return json.loads(raw)
|
|
except json.JSONDecodeError as exc:
|
|
raise ReconstructionError(code, f"{action} returned an invalid private JSON readback") from exc
|
|
|
|
|
|
def install_v3_contract(args: argparse.Namespace, container: str) -> dict[str, Any]:
|
|
completed = canonical_rebuild._run(
|
|
[
|
|
args.docker_bin,
|
|
"cp",
|
|
str(V3_CONTRACT_PATH),
|
|
f"{container}:{V3_CONTRACT_DESTINATION}",
|
|
],
|
|
timeout=args.command_timeout,
|
|
)
|
|
_require_command_success(
|
|
completed,
|
|
code="v3_contract_copy_failed",
|
|
action="V3 epistemic contract copy",
|
|
)
|
|
|
|
cutover = _read_json(
|
|
args,
|
|
container,
|
|
"""begin;
|
|
alter role kb_apply nologin;
|
|
do $cutover$
|
|
begin
|
|
if exists (
|
|
select 1
|
|
from pg_catalog.pg_stat_activity activity
|
|
where activity.datid = (select oid from pg_catalog.pg_database where datname = current_database())
|
|
and activity.usename = 'kb_apply'
|
|
and activity.pid <> pg_catalog.pg_backend_pid()
|
|
) then
|
|
raise exception 'V3 reconstruction cutover found an active kb_apply session';
|
|
end if;
|
|
end
|
|
$cutover$;
|
|
commit;
|
|
select jsonb_build_object(
|
|
'apply_role_login_disabled', not (select role.rolcanlogin from pg_catalog.pg_roles role where role.rolname = 'kb_apply'),
|
|
'active_apply_sessions', (
|
|
select count(*)
|
|
from pg_catalog.pg_stat_activity activity
|
|
where activity.datid = (select oid from pg_catalog.pg_database where datname = current_database())
|
|
and activity.usename = 'kb_apply'
|
|
and activity.pid <> pg_catalog.pg_backend_pid()
|
|
)
|
|
)::text;
|
|
""",
|
|
code="v3_cutover_preflight_failed",
|
|
action="V3 disposable writer cutover preflight",
|
|
)
|
|
if cutover != {"apply_role_login_disabled": True, "active_apply_sessions": 0}:
|
|
raise ReconstructionError(
|
|
"v3_cutover_preflight_failed",
|
|
"V3 disposable writer cutover did not disable login with zero active apply sessions",
|
|
)
|
|
|
|
for destination, label in (
|
|
(V3_CONTRACT_DESTINATION, "V3 epistemic contract"),
|
|
(GUARD_PREREQUISITES_DESTINATION, "guarded apply prerequisites after V3"),
|
|
):
|
|
completed = canonical_rebuild._run(
|
|
[
|
|
args.docker_bin,
|
|
"exec",
|
|
container,
|
|
"psql",
|
|
"-X",
|
|
"-U",
|
|
"postgres",
|
|
"-d",
|
|
args.database,
|
|
"-Atq",
|
|
"-v",
|
|
"ON_ERROR_STOP=1",
|
|
"-f",
|
|
destination,
|
|
],
|
|
timeout=args.apply_timeout,
|
|
)
|
|
_require_command_success(
|
|
completed,
|
|
code="v3_contract_install_failed",
|
|
action=label,
|
|
)
|
|
|
|
catalog = _read_json(
|
|
args,
|
|
container,
|
|
"""select jsonb_build_object(
|
|
'assessment_table_present', to_regclass('public.claim_evidence_assessments') is not null,
|
|
'apply_role_login_restored', (select role.rolcanlogin from pg_catalog.pg_roles role where role.rolname = 'kb_apply'),
|
|
'cutover_guard_enabled', exists (
|
|
select 1 from pg_catalog.pg_trigger trigger
|
|
where trigger.tgrelid = 'kb_stage.kb_proposals'::regclass
|
|
and trigger.tgname = 'teleo_v3_00_enforce_apply_contract_cutover'
|
|
and trigger.tgfoid = 'kb_stage.teleo_v3_enforce_apply_contract_cutover()'::regprocedure
|
|
and not trigger.tgisinternal
|
|
and trigger.tgenabled = 'O'
|
|
),
|
|
'decision_guard_enabled', exists (
|
|
select 1 from pg_catalog.pg_trigger trigger
|
|
where trigger.tgrelid = 'kb_stage.kb_proposals'::regclass
|
|
and trigger.tgname = 'teleo_v3_guard_linked_proposal_decision'
|
|
and trigger.tgfoid = 'kb_stage.teleo_v3_guard_linked_proposal_decision()'::regprocedure
|
|
and not trigger.tgisinternal
|
|
and trigger.tgenabled = 'O'
|
|
)
|
|
)::text;
|
|
""",
|
|
code="v3_contract_readback_failed",
|
|
action="V3 epistemic contract catalog readback",
|
|
)
|
|
if catalog != {
|
|
"assessment_table_present": True,
|
|
"apply_role_login_restored": True,
|
|
"cutover_guard_enabled": True,
|
|
"decision_guard_enabled": True,
|
|
}:
|
|
raise ReconstructionError("v3_contract_readback_failed", "V3 contract catalog readback was incomplete")
|
|
return {
|
|
"status": "pass",
|
|
"sha256": canonical_rebuild.sha256_file(V3_CONTRACT_PATH),
|
|
"cutover": cutover,
|
|
"guard_prerequisites_rerun": True,
|
|
"catalog": catalog,
|
|
}
|
|
|
|
|
|
def _assert_exact_json(actual: Any, expected: Any, *, code: str, action: str) -> None:
|
|
if _canonical_json(actual) != _canonical_json(expected):
|
|
raise ReconstructionError(code, f"{action} did not match the hash-pinned material")
|
|
|
|
|
|
def _assert_exact_canonical_rows(actual: Any, expected: Any, *, code: str, action: str) -> None:
|
|
def normalize(value: Any, path: str) -> dict[str, list[dict[str, Any]]]:
|
|
if not isinstance(value, dict):
|
|
raise ValueError(f"{path} must be an object")
|
|
normalized: dict[str, list[dict[str, Any]]] = {}
|
|
for table, rows in value.items():
|
|
if (
|
|
not isinstance(table, str)
|
|
or not isinstance(rows, list)
|
|
or any(not isinstance(row, dict) for row in rows)
|
|
):
|
|
raise ValueError(f"{path} must map table names to row arrays")
|
|
normalized[table] = [
|
|
replay_receipt.normalize_typed_row(row, path=f"{path}.{table}[{index}]")
|
|
for index, row in enumerate(rows)
|
|
]
|
|
return normalized
|
|
|
|
try:
|
|
normalized_actual = normalize(actual, "actual canonical rows")
|
|
normalized_expected = normalize(expected, "expected canonical rows")
|
|
except ValueError as exc:
|
|
raise ReconstructionError(code, f"{action} returned invalid typed canonical rows") from exc
|
|
_assert_exact_json(normalized_actual, normalized_expected, code=code, action=action)
|
|
|
|
|
|
def _assert_exact_proposal_readback(actual: Any, expected: Any, *, code: str, action: str) -> None:
|
|
def normalize(value: Any, path: str) -> dict[str, Any]:
|
|
if not isinstance(value, dict):
|
|
raise ValueError(f"{path} must be an object")
|
|
normalized = dict(value)
|
|
for field in ("proposal", "approval_snapshot", "approved_proposal_snapshot"):
|
|
row = normalized.get(field)
|
|
if not isinstance(row, dict):
|
|
raise ValueError(f"{path}.{field} must be an object")
|
|
normalized[field] = replay_receipt.normalize_typed_row(row, path=f"{path}.{field}")
|
|
return normalized
|
|
|
|
try:
|
|
normalized_actual = normalize(actual, "actual proposal readback")
|
|
normalized_expected = normalize(expected, "expected proposal readback")
|
|
except ValueError as exc:
|
|
raise ReconstructionError(code, f"{action} returned invalid typed proposal rows") from exc
|
|
_assert_exact_json(normalized_actual, normalized_expected, code=code, action=action)
|
|
|
|
|
|
def _entry_summary(entry: LedgerEntry) -> dict[str, Any]:
|
|
proposal = entry.applied_proposal
|
|
return {
|
|
"sequence": entry.sequence,
|
|
"proposal_id": proposal["id"],
|
|
"proposal_type": proposal["proposal_type"],
|
|
"contract_version": entry.contract_version,
|
|
"material_sha256": entry.material_sha256,
|
|
"replay_material_sha256": entry.replay_material_sha256,
|
|
"original_apply_sql_sha256": entry.receipt["apply_engine"]["apply_sql_sha256"],
|
|
"current_apply_sql_sha256": entry.current_apply_sql_sha256,
|
|
"expected_row_counts": entry.receipt["expected_row_counts"],
|
|
"seed_exact": False,
|
|
"proposal_seed_exact": False,
|
|
"canonical_seed_exact": False,
|
|
"guarded_apply_executed": False,
|
|
"mutating_prestate_captured": False,
|
|
"mutating_delta_validated": False,
|
|
"mutating_poststate_normalized": False,
|
|
"proposal_exact": False,
|
|
"canonical_rows_exact": False,
|
|
"status": "pending",
|
|
}
|
|
|
|
|
|
def apply_entry(
|
|
args: argparse.Namespace,
|
|
container: str,
|
|
entry: LedgerEntry,
|
|
summary: dict[str, Any],
|
|
) -> None:
|
|
mutating_prestate: dict[str, Any] | None = None
|
|
try:
|
|
_psql(
|
|
args,
|
|
container,
|
|
build_seed_sql(entry),
|
|
role="postgres",
|
|
timeout=args.apply_timeout,
|
|
code="entry_seed_failed",
|
|
action=f"ledger entry {entry.sequence} exact seed",
|
|
)
|
|
proposal_readback = _read_json(
|
|
args,
|
|
container,
|
|
build_proposal_readback_sql(entry.approved_proposal["id"]),
|
|
code="entry_seed_readback_failed",
|
|
action=f"ledger entry {entry.sequence} proposal seed readback",
|
|
)
|
|
_assert_exact_proposal_readback(
|
|
proposal_readback,
|
|
{
|
|
"proposal": entry.approved_proposal,
|
|
"approval_snapshot": entry.approval_snapshot,
|
|
"approved_proposal_snapshot": entry.approved_proposal,
|
|
},
|
|
code="entry_seed_mismatch",
|
|
action=f"ledger entry {entry.sequence} proposal seed",
|
|
)
|
|
summary["proposal_seed_exact"] = True
|
|
if entry.applied_proposal["proposal_type"] == "revise_strategy":
|
|
mutating_prestate = _validate_revise_strategy_prestate(
|
|
_read_json(
|
|
args,
|
|
container,
|
|
build_revise_strategy_prestate_sql(entry),
|
|
code="mutating_prestate_readback_failed",
|
|
action=f"ledger entry {entry.sequence} revise_strategy prestate readback",
|
|
)
|
|
)
|
|
summary["mutating_prestate_captured"] = True
|
|
elif entry.contract_version == 3:
|
|
seeded_rows = _read_json(
|
|
args,
|
|
container,
|
|
replay_receipt.build_postflight_sql(entry.receipt["proposal"]),
|
|
code="entry_seed_readback_failed",
|
|
action=f"ledger entry {entry.sequence} V3 empty canonical prestate readback",
|
|
)
|
|
_assert_exact_canonical_rows(
|
|
seeded_rows,
|
|
{table: [] for table in entry.receipt["canonical_rows"]},
|
|
code="entry_seed_mismatch",
|
|
action=f"ledger entry {entry.sequence} V3 empty canonical prestate",
|
|
)
|
|
summary["canonical_seed_exact"] = True
|
|
else:
|
|
seeded_rows = _read_json(
|
|
args,
|
|
container,
|
|
replay_receipt.build_postflight_sql(entry.receipt["proposal"]),
|
|
code="entry_seed_readback_failed",
|
|
action=f"ledger entry {entry.sequence} canonical seed readback",
|
|
)
|
|
_assert_exact_canonical_rows(
|
|
seeded_rows,
|
|
entry.receipt["canonical_rows"],
|
|
code="entry_seed_mismatch",
|
|
action=f"ledger entry {entry.sequence} canonical seed",
|
|
)
|
|
summary["canonical_seed_exact"] = True
|
|
summary["seed_exact"] = summary["proposal_seed_exact"] and summary["canonical_seed_exact"]
|
|
|
|
_psql(
|
|
args,
|
|
container,
|
|
entry.current_apply_sql,
|
|
role="kb_apply",
|
|
timeout=args.apply_timeout,
|
|
code="guarded_apply_failed",
|
|
action=f"ledger entry {entry.sequence} guarded apply",
|
|
)
|
|
summary["guarded_apply_executed"] = True
|
|
|
|
if mutating_prestate is not None:
|
|
generated_rows = _read_json(
|
|
args,
|
|
container,
|
|
build_revise_strategy_generated_delta_sql(entry, mutating_prestate),
|
|
code="mutating_delta_readback_failed",
|
|
action=f"ledger entry {entry.sequence} revise_strategy generated delta readback",
|
|
)
|
|
normalization_sql = build_revise_strategy_normalization_sql(entry, mutating_prestate, generated_rows)
|
|
summary["mutating_delta_validated"] = True
|
|
_psql(
|
|
args,
|
|
container,
|
|
normalization_sql,
|
|
role="postgres",
|
|
timeout=args.apply_timeout,
|
|
code="mutating_poststate_normalization_failed",
|
|
action=f"ledger entry {entry.sequence} revise_strategy exact poststate normalization",
|
|
)
|
|
summary["mutating_poststate_normalized"] = True
|
|
elif entry.contract_version == 3:
|
|
_psql(
|
|
args,
|
|
container,
|
|
build_v3_canonical_timestamp_normalization_sql(entry),
|
|
role="postgres",
|
|
timeout=args.command_timeout,
|
|
code="v3_canonical_timestamp_normalization_failed",
|
|
action=f"ledger entry {entry.sequence} V3 canonical timestamp normalization",
|
|
)
|
|
summary["mutating_poststate_normalized"] = True
|
|
|
|
_psql(
|
|
args,
|
|
container,
|
|
build_applied_timestamp_normalization_sql(entry),
|
|
role="postgres",
|
|
timeout=args.command_timeout,
|
|
code="proposal_timestamp_normalization_failed",
|
|
action=f"ledger entry {entry.sequence} deterministic ledger timestamp normalization",
|
|
)
|
|
proposal_readback = _read_json(
|
|
args,
|
|
container,
|
|
build_proposal_readback_sql(entry.applied_proposal["id"]),
|
|
code="entry_postflight_failed",
|
|
action=f"ledger entry {entry.sequence} applied proposal readback",
|
|
)
|
|
_assert_exact_proposal_readback(
|
|
proposal_readback,
|
|
{
|
|
"proposal": entry.applied_proposal,
|
|
"approval_snapshot": entry.approval_snapshot,
|
|
"approved_proposal_snapshot": entry.approved_proposal,
|
|
},
|
|
code="entry_postflight_mismatch",
|
|
action=f"ledger entry {entry.sequence} applied proposal",
|
|
)
|
|
summary["proposal_exact"] = True
|
|
|
|
final_rows = _read_json(
|
|
args,
|
|
container,
|
|
replay_receipt.build_postflight_sql(entry.receipt["proposal"]),
|
|
code="entry_postflight_failed",
|
|
action=f"ledger entry {entry.sequence} exact row readback",
|
|
)
|
|
_assert_exact_canonical_rows(
|
|
final_rows,
|
|
entry.receipt["canonical_rows"],
|
|
code="entry_postflight_mismatch",
|
|
action=f"ledger entry {entry.sequence} canonical rows",
|
|
)
|
|
summary["canonical_rows_exact"] = True
|
|
summary["status"] = "pass"
|
|
except ReconstructionError:
|
|
summary["status"] = "fail"
|
|
raise
|
|
|
|
|
|
def _sanitize_parity(result: dict[str, Any]) -> dict[str, Any]:
|
|
source = dict(result["source_manifest"])
|
|
source.pop("path", None)
|
|
return {
|
|
"status": result["status"],
|
|
"verifier": result["verifier"],
|
|
"source_manifest": source,
|
|
"target_manifest": result["target_manifest"],
|
|
"problems": result["problems"],
|
|
"details": result["details"],
|
|
}
|
|
|
|
|
|
def _run_parity(
|
|
args: argparse.Namespace,
|
|
container: str,
|
|
manifest_path: Path,
|
|
manifest: dict[str, Any],
|
|
*,
|
|
code: str,
|
|
action: str,
|
|
) -> dict[str, Any]:
|
|
try:
|
|
result = canonical_rebuild.run_parity_check(
|
|
args.docker_bin,
|
|
container,
|
|
args.database,
|
|
source_manifest_path=manifest_path,
|
|
source_manifest=manifest,
|
|
manifest_sql=PARITY_SQL_PATH,
|
|
command_timeout=args.manifest_timeout,
|
|
max_target_query_ms=args.max_target_query_ms,
|
|
max_target_source_ratio=args.max_target_source_ratio,
|
|
)
|
|
except (OSError, ValueError, RuntimeError, subprocess.TimeoutExpired, json.JSONDecodeError) as exc:
|
|
raise ReconstructionError(code, f"{action} could not complete; private details withheld") from exc
|
|
sanitized = _sanitize_parity(result)
|
|
if sanitized["status"] != "pass":
|
|
raise ReconstructionError(code, f"{action} found a manifest mismatch")
|
|
return sanitized
|
|
|
|
|
|
def _initial_receipt(args: argparse.Namespace, container: str) -> dict[str, Any]:
|
|
return {
|
|
"artifact": "local_genesis_plus_ledger_reconstruction",
|
|
"contract_version": 1,
|
|
"generated_at_utc": datetime.now(UTC).isoformat(),
|
|
"status": "running",
|
|
"claim_ceiling": CLAIM_CEILING,
|
|
"not_proven": list(NOT_PROVEN),
|
|
"database": args.database,
|
|
"inputs": {
|
|
"genesis_dump": {"bytes": None, "sha256": None, "pin_verified": False},
|
|
"genesis_manifest": {"bytes": None, "sha256": None, "pin_verified": False},
|
|
"ledger": {"bytes": None, "sha256": None, "pin_verified": False},
|
|
"admission_manifest": {"bytes": None, "sha256": None, "pin_verified": False},
|
|
"final_manifest": {"bytes": None, "sha256": None, "pin_verified": False},
|
|
"engine_hashes": {},
|
|
},
|
|
"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,
|
|
},
|
|
"guard_prerequisites": {"status": "pending", "sha256": None},
|
|
"v3_contract": {"status": "not_required", "sha256": None},
|
|
"genesis_parity": {"status": "pending"},
|
|
"genesis_key_counts": {},
|
|
"ledger": {"entry_count": None, "entries": []},
|
|
"admission": {
|
|
"status": "pending",
|
|
"canonical_retrieval_eligible": False,
|
|
},
|
|
"final_parity": {"status": "pending"},
|
|
"final_key_counts": {},
|
|
"cleanup": {
|
|
"attempted": False,
|
|
"container": container,
|
|
"container_absent": None,
|
|
"status": "pending",
|
|
},
|
|
"timings_seconds": {},
|
|
"safety": {
|
|
"production_database_touched": False,
|
|
"network_access_available_to_container": False,
|
|
"persistent_postgres_storage_used": False,
|
|
"private_replay_material_emitted": False,
|
|
"vps_gcp_telegram_or_live_db_used": False,
|
|
},
|
|
"error": None,
|
|
}
|
|
|
|
|
|
def run_reconstruction(args: argparse.Namespace) -> dict[str, Any]:
|
|
started_at = time.monotonic()
|
|
container = canonical_rebuild.new_container_name(args.container_prefix)
|
|
receipt = _initial_receipt(args, container)
|
|
phase = "input_preflight"
|
|
primary_error: dict[str, Any] | None = None
|
|
|
|
try:
|
|
bundle = load_bundle(args)
|
|
receipt["inputs"] = {
|
|
"genesis_dump": {
|
|
"bytes": bundle.genesis_dump_bytes,
|
|
"sha256": bundle.genesis_dump_sha256,
|
|
"pin_verified": True,
|
|
},
|
|
"genesis_manifest": {
|
|
"bytes": bundle.genesis_manifest_bytes,
|
|
"sha256": bundle.genesis_manifest_sha256,
|
|
"pin_verified": True,
|
|
},
|
|
"ledger": {
|
|
"bytes": bundle.ledger_bytes,
|
|
"sha256": bundle.ledger_sha256,
|
|
"pin_verified": True,
|
|
},
|
|
"admission_manifest": {
|
|
"bytes": bundle.admission_manifest_bytes,
|
|
"sha256": bundle.admission_manifest_sha256,
|
|
"pin_verified": bundle.admission_manifest_sha256 is not None,
|
|
},
|
|
"final_manifest": {
|
|
"bytes": bundle.final_manifest_bytes,
|
|
"sha256": bundle.final_manifest_sha256,
|
|
"pin_verified": True,
|
|
},
|
|
"engine_hashes": bundle.engine_hashes,
|
|
}
|
|
receipt["ledger"]["entry_count"] = len(bundle.entries)
|
|
receipt["ledger"]["entries"] = [_entry_summary(entry) for entry in bundle.entries]
|
|
receipt["admission"] = bundle.admission
|
|
if bundle.admission["canonical_retrieval_eligible"] is not True:
|
|
receipt["claim_ceiling"] = f"forensic-only reconstruction; {CLAIM_CEILING}"
|
|
if any(entry.contract_version == 3 for entry in bundle.entries):
|
|
receipt["v3_contract"] = {
|
|
"status": "pending",
|
|
"sha256": bundle.engine_hashes["v3_contract_sha256"],
|
|
}
|
|
|
|
phase = "container_name_preflight"
|
|
absent, absence_error = canonical_rebuild.inspect_container_absence(
|
|
args.docker_bin, container, timeout=args.command_timeout
|
|
)
|
|
if absent is not True:
|
|
raise ReconstructionError(
|
|
"container_name_not_proven_unused",
|
|
"generated disposable container name could not be proven unused",
|
|
) from RuntimeError(absence_error or "container exists")
|
|
receipt["container"]["preexisting_absence_proven"] = True
|
|
|
|
phase = "container_start"
|
|
tmpfs_options = f"{canonical_rebuild.DATA_DIR}:rw,nosuid,nodev,noexec,size={args.tmpfs_mb}m"
|
|
completed = canonical_rebuild._run(
|
|
[
|
|
args.docker_bin,
|
|
"run",
|
|
"--detach",
|
|
"--rm",
|
|
"--network",
|
|
"none",
|
|
"--name",
|
|
container,
|
|
"--label",
|
|
f"{canonical_rebuild.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_command_success(
|
|
completed,
|
|
code="container_start_failed",
|
|
action="networkless disposable PostgreSQL start",
|
|
).strip()
|
|
if not container_id:
|
|
raise ReconstructionError("container_start_failed", "container start returned no id")
|
|
receipt["container"]["started"] = True
|
|
receipt["container"]["id_prefix"] = container_id[:12]
|
|
receipt["container"]["isolation"] = canonical_rebuild.inspect_container_isolation(
|
|
args.docker_bin,
|
|
container,
|
|
expected_image=args.image,
|
|
expected_label=container,
|
|
timeout=args.command_timeout,
|
|
)
|
|
|
|
phase = "database_readiness"
|
|
receipt["container"]["named_database_psql_attempts"] = canonical_rebuild.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
|
|
|
|
phase = "application_role_bootstrap"
|
|
receipt["application_roles_bootstrapped"] = canonical_rebuild.execute_application_role_bootstrap(
|
|
args.docker_bin,
|
|
container,
|
|
args.database,
|
|
bundle.genesis_manifest,
|
|
timeout=args.command_timeout,
|
|
)
|
|
|
|
phase = "genesis_restore"
|
|
completed = canonical_rebuild._run(
|
|
[
|
|
args.docker_bin,
|
|
"cp",
|
|
str(args.genesis_dump.resolve()),
|
|
f"{container}:{canonical_rebuild.DUMP_DESTINATION}",
|
|
],
|
|
timeout=args.command_timeout,
|
|
)
|
|
_require_command_success(completed, code="genesis_copy_failed", action="genesis dump copy")
|
|
completed = canonical_rebuild._run(
|
|
[
|
|
args.docker_bin,
|
|
"exec",
|
|
container,
|
|
"pg_restore",
|
|
"-U",
|
|
"postgres",
|
|
"-d",
|
|
args.database,
|
|
"--no-owner",
|
|
"--no-privileges",
|
|
"--exit-on-error",
|
|
canonical_rebuild.DUMP_DESTINATION,
|
|
],
|
|
timeout=args.restore_timeout,
|
|
)
|
|
_require_command_success(completed, code="genesis_restore_failed", action="genesis restore")
|
|
|
|
phase = "guard_prerequisites"
|
|
completed = canonical_rebuild._run(
|
|
[
|
|
args.docker_bin,
|
|
"cp",
|
|
str(GUARD_PREREQUISITES_PATH),
|
|
f"{container}:{GUARD_PREREQUISITES_DESTINATION}",
|
|
],
|
|
timeout=args.command_timeout,
|
|
)
|
|
_require_command_success(
|
|
completed,
|
|
code="guard_prerequisites_copy_failed",
|
|
action="guarded apply prerequisites copy",
|
|
)
|
|
completed = canonical_rebuild._run(
|
|
[
|
|
args.docker_bin,
|
|
"exec",
|
|
container,
|
|
"psql",
|
|
"-X",
|
|
"-U",
|
|
"postgres",
|
|
"-d",
|
|
args.database,
|
|
"-Atq",
|
|
"-v",
|
|
"ON_ERROR_STOP=1",
|
|
"-f",
|
|
GUARD_PREREQUISITES_DESTINATION,
|
|
],
|
|
timeout=args.apply_timeout,
|
|
)
|
|
_require_command_success(
|
|
completed,
|
|
code="guard_prerequisites_failed",
|
|
action="guarded apply prerequisites",
|
|
)
|
|
receipt["guard_prerequisites"] = {
|
|
"status": "pass",
|
|
"sha256": bundle.engine_hashes["guard_prerequisites_sha256"],
|
|
}
|
|
|
|
phase = "genesis_parity"
|
|
receipt["genesis_parity"] = _run_parity(
|
|
args,
|
|
container,
|
|
args.genesis_manifest,
|
|
bundle.genesis_manifest,
|
|
code="genesis_parity_failed",
|
|
action="genesis parity verification",
|
|
)
|
|
receipt["genesis_key_counts"], _ = canonical_rebuild.collect_key_counts(
|
|
args.docker_bin,
|
|
container,
|
|
args.database,
|
|
timeout=args.command_timeout,
|
|
)
|
|
|
|
phase = "ordered_ledger_replay"
|
|
v3_installed = False
|
|
for entry, summary in zip(bundle.entries, receipt["ledger"]["entries"], strict=True):
|
|
if entry.contract_version == 3 and not v3_installed:
|
|
phase = "v3_contract_install"
|
|
receipt["v3_contract"] = install_v3_contract(args, container)
|
|
v3_installed = True
|
|
phase = "ordered_ledger_replay"
|
|
apply_entry(args, container, entry, summary)
|
|
|
|
phase = "final_parity"
|
|
receipt["final_parity"] = _run_parity(
|
|
args,
|
|
container,
|
|
bundle.final_manifest_path,
|
|
bundle.final_manifest,
|
|
code="final_parity_failed",
|
|
action="final parity verification",
|
|
)
|
|
receipt["final_key_counts"], _ = canonical_rebuild.collect_key_counts(
|
|
args.docker_bin,
|
|
container,
|
|
args.database,
|
|
timeout=args.command_timeout,
|
|
)
|
|
receipt["status"] = "pass"
|
|
except ReconstructionError as exc:
|
|
primary_error = {
|
|
"phase": phase,
|
|
"code": exc.code,
|
|
"type": type(exc).__name__,
|
|
"message": exc.safe_message,
|
|
}
|
|
receipt["status"] = "fail"
|
|
receipt["error"] = primary_error
|
|
except (
|
|
OSError,
|
|
ValueError,
|
|
RuntimeError,
|
|
subprocess.TimeoutExpired,
|
|
json.JSONDecodeError,
|
|
KeyboardInterrupt,
|
|
canonical_rebuild.TerminationRequested,
|
|
) as exc:
|
|
primary_error = {
|
|
"phase": phase,
|
|
"code": f"{phase}_failed",
|
|
"type": type(exc).__name__,
|
|
"message": f"{phase} failed; private details withheld",
|
|
}
|
|
receipt["status"] = "fail"
|
|
receipt["error"] = primary_error
|
|
finally:
|
|
cleanup_started = time.monotonic()
|
|
cleanup = canonical_rebuild.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",
|
|
"code": "cleanup_absence_unproven",
|
|
"type": "CleanupProofError",
|
|
"message": "disposable 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",
|
|
"code": "cleanup_interrupted",
|
|
"type": "TerminationRequested",
|
|
"message": "cleanup was interrupted; private details withheld",
|
|
}
|
|
if primary_error is None:
|
|
receipt["error"] = interruption_error
|
|
else:
|
|
receipt["cleanup_interruption"] = interruption_error
|
|
return receipt
|
|
|
|
|
|
def _declared_artifact_paths(ledger_path: Path) -> set[Path]:
|
|
ledger = _load_json_object(ledger_path, "ledger")
|
|
paths = {ledger_path.resolve()}
|
|
final = ledger.get("final_parity")
|
|
if isinstance(final, dict) and isinstance(final.get("manifest"), str):
|
|
paths.add(_resolve_artifact(ledger_path, final["manifest"], "final parity manifest"))
|
|
entries = ledger.get("entries")
|
|
if isinstance(entries, list):
|
|
for index, entry in enumerate(entries):
|
|
if isinstance(entry, dict) and isinstance(entry.get("material"), str):
|
|
paths.add(_resolve_artifact(ledger_path, entry["material"], f"entry {index + 1} material"))
|
|
return paths
|
|
|
|
|
|
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--genesis-dump", required=True, type=Path)
|
|
parser.add_argument("--genesis-manifest", required=True, type=Path)
|
|
parser.add_argument("--ledger", required=True, type=Path)
|
|
parser.add_argument("--ledger-sha256", required=True)
|
|
parser.add_argument("--admission-manifest", type=Path)
|
|
parser.add_argument("--admission-manifest-sha256")
|
|
parser.add_argument("--database", default="teleo")
|
|
parser.add_argument("--image", default=DEFAULT_REBUILD_IMAGE)
|
|
parser.add_argument("--container-prefix", default="teleo-genesis-ledger-rebuild")
|
|
parser.add_argument("--tmpfs-mb", default=1024, type=int)
|
|
parser.add_argument("--startup-timeout", default=60.0, type=float)
|
|
parser.add_argument("--restore-timeout", default=300.0, type=float)
|
|
parser.add_argument("--apply-timeout", default=180.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("--max-target-query-ms", default=2000.0, type=float)
|
|
parser.add_argument("--max-target-source-ratio", default=20.0, type=float)
|
|
parser.add_argument("--docker-bin", default="docker")
|
|
parser.add_argument("--output", required=True, type=Path)
|
|
args = parser.parse_args(argv)
|
|
|
|
if not canonical_rebuild.DATABASE_NAME_RE.fullmatch(args.database):
|
|
parser.error("--database must be a safe PostgreSQL identifier up to 63 characters")
|
|
if not canonical_rebuild.CONTAINER_PREFIX_RE.fullmatch(args.container_prefix):
|
|
parser.error("--container-prefix must be 1-40 Docker-safe characters")
|
|
if not IMAGE_DIGEST_RE.fullmatch(args.image):
|
|
parser.error("--image must be pinned by a lowercase sha256 digest")
|
|
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.apply_timeout, "--apply-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")
|
|
for path, flag in (
|
|
(args.genesis_dump, "--genesis-dump"),
|
|
(args.genesis_manifest, "--genesis-manifest"),
|
|
(args.ledger, "--ledger"),
|
|
):
|
|
if not path.is_file():
|
|
parser.error(f"{flag} must be an existing regular file")
|
|
if not SHA256_RE.fullmatch(args.ledger_sha256):
|
|
parser.error("--ledger-sha256 must be a lowercase SHA-256 digest")
|
|
if (args.admission_manifest is None) != (args.admission_manifest_sha256 is None):
|
|
parser.error("--admission-manifest and --admission-manifest-sha256 must be supplied together")
|
|
if args.admission_manifest is not None:
|
|
if not args.admission_manifest.is_file():
|
|
parser.error("--admission-manifest must be an existing regular file")
|
|
if not SHA256_RE.fullmatch(args.admission_manifest_sha256):
|
|
parser.error("--admission-manifest-sha256 must be a lowercase SHA-256 digest")
|
|
try:
|
|
protected = {
|
|
args.genesis_dump.resolve(),
|
|
args.genesis_manifest.resolve(),
|
|
*_declared_artifact_paths(args.ledger),
|
|
*FIXED_ENGINE_PATHS,
|
|
}
|
|
if args.admission_manifest is not None:
|
|
protected.add(args.admission_manifest.resolve())
|
|
except ReconstructionError as exc:
|
|
parser.error(exc.safe_message)
|
|
if args.output.resolve() in protected:
|
|
parser.error("--output must not overwrite a reconstruction input")
|
|
return args
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
args = parse_args(argv)
|
|
previous_handlers: dict[signal.Signals, Any] = {}
|
|
for watched_signal in (signal.SIGINT, signal.SIGTERM):
|
|
previous_handlers[watched_signal] = signal.getsignal(watched_signal)
|
|
signal.signal(watched_signal, canonical_rebuild._termination_handler)
|
|
try:
|
|
payload = run_reconstruction(args)
|
|
finally:
|
|
for watched_signal, previous in previous_handlers.items():
|
|
signal.signal(watched_signal, previous)
|
|
|
|
try:
|
|
canonical_rebuild.write_receipt(args.output, payload)
|
|
except OSError:
|
|
payload["status"] = "fail"
|
|
payload["receipt_write_error"] = "receipt write failed; private details withheld"
|
|
print(json.dumps(payload, indent=2, sort_keys=True))
|
|
return 0 if payload["status"] == "pass" else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|