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

replay exact revise strategy database transitions

Browse source
This commit is contained in:
twentyOne2x 2026-07-15 03:19:49 +02:00
parent 15c30f1fbe
commit a6947a3f8b
5 changed files with 1153 additions and 130 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -148,7 +148,7 @@ not retain every column of the proposal ledger, so exact reconstruction also
needs the full approved proposal row, immutable approval snapshot, and final
applied proposal row.
## Genesis Plus Strict Ledger: Working Insert-Only Slice
## Genesis Plus Strict Ledger: Working Deterministic Slice
`ops/run_local_genesis_ledger_rebuild.py` now executes the first exact
genesis-plus-ledger slice in one command:
@ -206,27 +206,65 @@ Each referenced material object has exact top-level keys
must contain every current `kb_stage.kb_proposals` column; partial envelopes
are rejected.
The command verifies every hash before starting Docker, restores a tmpfs-backed
Postgres with `--network none`, reapplies the current guarded prerequisites,
and proves genesis parity. For each entry it seeds the receipt's exact row IDs
and timestamps in the disposable clone, executes the existing `kb_apply`
payload-bound guard and ledger transition, checks exact proposal and canonical
row readbacks, then verifies the complete final parity manifest. Its public
The command verifies every hash before starting Docker, requires a
SHA-256-pinned Postgres image (and defaults to a pinned PostgreSQL 16 Alpine
multi-platform digest), restores it on tmpfs with `--network none`, reapplies
the current guarded prerequisites, and proves genesis parity. Insert-only
entries seed the receipt's exact canonical row IDs and timestamps before the
existing `kb_apply` payload-bound guard executes. For `revise_strategy`, the
runner seeds only the proposal and approval, captures the target agent's
prestate, executes the real guarded transition, validates the generated delta,
and then replaces only those generated rows with the receipt-pinned IDs and
timestamps. Every path checks exact proposal and canonical row readbacks before
verifying the complete final parity manifest. Its public
mode-`0600` receipt contains hashes, IDs, types, counts, parity summaries, and
cleanup proof, but no payloads, rows, SQL, source paths, or command errors.
The legacy `seed_exact` summary is the insert-only aggregate of
`proposal_seed_exact` and `canonical_seed_exact`; it is intentionally false for
successful mutating entries, which instead report
`mutating_delta_validated` and `mutating_poststate_normalized`.
Fresh guard bootstrap rows use a fixed baseline timestamp rather than wall
clock time, so repeated clean restores have identical row hashes.
The exact v1 claim ceiling is intentionally narrow:
The exact v1 claim ceiling is intentionally bounded:
- `add_edge`, `attach_evidence`, and `approve_claim` strict receipts execute;
- `add_edge`, `attach_evidence`, `approve_claim`, and `revise_strategy` strict
receipts execute;
- sequence gaps, hash drift, engine drift, duplicate proposals, and legacy or
freeform payloads fail before container start;
- `revise_strategy` fails closed because its current receipt omits the prior
strategies and nodes that the apply engine deactivates or retires;
- `revise_strategy` is accepted only when the receipt pins the exact SQL that
matches the current guarded apply engine. Immediately before each entry, the
runner captures the target agent's strategy/node IDs, active strategy,
non-retired nodes, and maximum version. It then validates the generated
post-minus-pre delta, requires `version = previous maximum + 1`, and replaces
only the generated strategy/node rows with the exact receipt rows;
- the original transaction timestamp is derived from the fresh strategy
`created_at` and must equal every fresh node's `created_at` and `updated_at`.
Only node IDs observed as non-retired before apply receive that timestamp;
already-retired, unrelated-agent, and shared NULL-agent rows stay untouched;
- generated nodes must have no anchors before normalization, preventing a
delete-and-reinsert step from silently cascading future trigger-created rows;
- full proposal before/after rows are mandatory because the current receipt
envelope does not retain proposal origin fields or exact `updated_at`;
- this proves only an isolated local reconstruction. It does not touch or prove
VPS, GCP, Telegram, a live database, or blank-schema source recompilation.
The transition contract for `revise_strategy` is final-state deterministic, not
a claim that the v1 receipt independently contains a historical before-image:
```text
exact genesis/pre-entry state
+ exact current/original guarded SQL
+ receipt-pinned fresh strategy and nodes
-> prior active strategy inactive
-> exactly the prior non-retired nodes retired at the original transaction time
-> one receipt-exact active strategy and receipt-exact node set
```
The genesis and final manifests remain mandatory oracles. Any incorrect
prestate, unrelated-row mutation, missing/extra generated row, semantic drift,
version drift, hash drift, or final rowset difference fails the reconstruction.
The source compiler now turns one raw artifact, its strict UTF-8 extraction,
and a reviewed extraction manifest into a deterministic, hash-bound
`pending_review` proposal bundle:
@ -272,10 +310,9 @@ neither command applies canonical rows.
The existing full-data clone canary separately proves that a reviewed packet
can create source, claim, evidence, and edge rows and affect later reasoning.
The remaining reconstruction work is to retain complete deltas for mutating
contracts such as `revise_strategy`, backfill or explicitly reject legacy
freeform applies, and extend beyond genesis recovery to a blank-schema source
compiler. The insert-only ledger runner does not prove that every historical
The remaining reconstruction work is to backfill or explicitly reject legacy
freeform applies and extend beyond genesis recovery to a blank-schema source
compiler. The strict ledger runner does not prove that every historical
canonical row can be rebuilt semantically from retained sources.
## Definition Of Working

543
ops/run_local_genesis_ledger_rebuild.py
View file

@ -1,12 +1,14 @@
#!/usr/bin/env python3
"""Deterministically rebuild a disposable Postgres from genesis plus a strict ledger.
The v1 replay boundary is intentionally narrow. It supports strict, insert-only
``add_edge``, ``attach_evidence``, and ``approve_claim`` receipts. Each ledger
entry must also retain the exact approved and applied proposal rows plus the
immutable approval snapshot. Legacy/freeform entries and ``revise_strategy``
fail before Docker starts because their retained material is not a complete
database delta.
The v1 replay boundary supports strict ``add_edge``, ``attach_evidence``,
``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
@ -46,15 +48,21 @@ LEDGER_ARTIFACT = "teleo_genesis_plus_ledger"
MATERIAL_ARTIFACT = "teleo_strict_proposal_replay_material"
LEDGER_CONTRACT_VERSION = 1
MATERIAL_CONTRACT_VERSION = 1
SUPPORTED_PROPOSAL_TYPES = frozenset({"add_edge", "attach_evidence", "approve_claim"})
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 insert-only proposal replay; supported types are "
"add_edge, attach_evidence, and approve_claim; full approved/applied proposal rows "
"and immutable approval snapshots are required"
"exact genesis plus ordered strict proposal replay; supported types are add_edge, "
"attach_evidence, approve_claim, and revise_strategy; 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",
"revise_strategy replay because prior-row mutations are absent from v1 receipts",
"source-corpus recompilation from a blank schema",
"VPS, GCP, Telegram, or any live database mutation",
)
@ -77,6 +85,7 @@ FIXED_ENGINE_PATHS = frozenset(
)
SHA256_RE = re.compile(r"[0-9a-f]{64}\Z")
IMAGE_DIGEST_RE = re.compile(r"\S+@sha256:[0-9a-f]{64}\Z")
PROPOSAL_TRANSITION_FIELDS = frozenset(
{"status", "applied_by_handle", "applied_by_agent_id", "applied_at", "updated_at"}
)
@ -121,6 +130,35 @@ CANONICAL_TABLE_ORDER = (
"claim_evidence",
"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):
@ -308,6 +346,119 @@ def _validate_proposal_rows(
)
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",
)
applied_time = _parse_aware_timestamp(proposal.get("applied_at"), "receipt proposal applied_at")
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],
*,
@ -340,11 +491,6 @@ def _validate_entry_material(
if not isinstance(proposal, dict):
raise ReconstructionError("invalid_material", "replay receipt has no proposal object")
proposal_type = proposal.get("proposal_type")
if proposal_type == "revise_strategy":
raise ReconstructionError(
"unsupported_mutating_receipt",
"revise_strategy receipts omit prior strategy and node mutations and cannot replay exactly",
)
if proposal_type not in SUPPORTED_PROPOSAL_TYPES:
raise ReconstructionError(
"unsupported_proposal_type", "ledger entry proposal type is not supported by v1 reconstruction"
@ -358,6 +504,8 @@ def _validate_entry_material(
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)
try:
rebuilt = replay_receipt.build_replay_receipt(
proposal,
@ -382,6 +530,7 @@ def _validate_entry_material(
raise ReconstructionError(
"replay_material_hash_mismatch", "ledger replay-material pin does not match the private receipt"
)
receipt = {**receipt, "canonical_rows": rebuilt["canonical_rows"]}
approved = material["approved_proposal"]
approval = material["approval_snapshot"]
@ -394,6 +543,15 @@ def _validate_entry_material(
"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,
@ -405,7 +563,7 @@ def _validate_entry_material(
applied_proposal=applied,
receipt=receipt,
current_apply_sql=current_apply_sql,
current_apply_sql_sha256=_sha256_text(current_apply_sql),
current_apply_sql_sha256=current_apply_sql_sha256,
)
@ -550,6 +708,280 @@ def build_seed_sql(entry: LedgerEntry) -> str:
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(
@ -694,7 +1126,12 @@ def _entry_summary(entry: LedgerEntry) -> dict[str, Any]:
"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",
@ -707,6 +1144,7 @@ def apply_entry(
entry: LedgerEntry,
summary: dict[str, Any],
) -> None:
mutating_prestate: dict[str, Any] | None = None
try:
_psql(
args,
@ -733,20 +1171,34 @@ def apply_entry(
code="entry_seed_mismatch",
action=f"ledger entry {entry.sequence} proposal seed",
)
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_json(
seeded_rows,
entry.receipt["canonical_rows"],
code="entry_seed_mismatch",
action=f"ledger entry {entry.sequence} canonical seed",
)
summary["seed_exact"] = True
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
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_json(
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,
@ -759,6 +1211,29 @@ def apply_entry(
)
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
_psql(
args,
container,
@ -1199,7 +1674,7 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
parser.add_argument("--ledger", required=True, type=Path)
parser.add_argument("--ledger-sha256", required=True)
parser.add_argument("--database", default="teleo")
parser.add_argument("--image", default=canonical_rebuild.DEFAULT_IMAGE)
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)
@ -1218,8 +1693,8 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
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 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 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:

5
scripts/kb_apply_prereqs.sql
View file

@ -315,8 +315,9 @@ alter table kb_stage.kb_review_principals owner to kb_gate_owner;
alter table kb_stage.kb_proposal_approvals owner to kb_gate_owner;
insert into kb_stage.kb_review_principals
(db_role, reviewed_by_handle, reviewed_by_agent_id, active)
select 'kb_review'::name, a.handle, a.id, true
(db_role, reviewed_by_handle, reviewed_by_agent_id, active, created_at)
select 'kb_review'::name, a.handle, a.id, true,
'1970-01-01 00:00:00+00'::timestamptz
from public.agents a
where a.handle = 'm3ta'
on conflict (db_role) do nothing;

9
tests/test_kb_apply_prereqs.py
View file

@ -203,6 +203,15 @@ def _lines(completed: subprocess.CompletedProcess[str]) -> set[str]:
return {line for line in completed.stdout.splitlines() if line}
def test_reviewer_principal_seed_timestamp_is_deterministic(migrated_postgres: str) -> None:
created_at = _psql(
migrated_postgres,
"select created_at::text from kb_stage.kb_review_principals where db_role = 'kb_review';",
).stdout.strip()
assert created_at == "1970-01-01 00:00:00+00"
def _catalog_snapshot(container: str) -> str:
return _psql(
container,

659
tests/test_run_local_genesis_ledger_rebuild.py
View file

@ -23,6 +23,18 @@ REVIEWER_ID = "11111111-1111-4111-8111-111111111111"
SERVICE_ID = "44444444-4444-4444-4444-444444444444"
PROPOSAL_ID = "99999999-9999-4999-8999-999999999999"
EDGE_ID = "eeeeeeee-eeee-4eee-8eee-eeeeeeeeeeee"
OTHER_AGENT_ID = "22222222-2222-4222-8222-222222222222"
REVISE_PROPOSAL_ID = "88888888-8888-4888-8888-888888888888"
PRIOR_STRATEGY_ID = "33333333-3333-4333-8333-333333333333"
PRIOR_ACTIVE_NODE_ID = "55555555-5555-4555-8555-555555555555"
PRIOR_RETIRED_NODE_ID = "66666666-6666-4666-8666-666666666666"
OTHER_STRATEGY_ID = "77777777-7777-4777-8777-777777777777"
OTHER_ACTIVE_NODE_ID = "12121212-1212-4212-8212-121212121212"
NULL_AGENT_NODE_ID = "13131313-1313-4313-8313-131313131313"
PRIOR_NODE_ANCHOR_ID = "19191919-1919-4919-8919-191919191919"
RECEIPT_STRATEGY_ID = "14141414-1414-4414-8414-141414141414"
RECEIPT_NODE_A_ID = "15151515-1515-4515-8515-151515151515"
RECEIPT_NODE_B_ID = "16161616-1616-4616-8616-161616161616"
PRIVATE_MARKER = "PRIVATE-REPLAY-MATERIAL-MUST-NOT-LEAK"
@ -155,6 +167,58 @@ def proposal_rows() -> tuple[dict, dict, dict]:
return approved, applied, approval
def revise_strategy_proposal_rows() -> tuple[dict, dict, dict]:
approved, _, approval = proposal_rows()
payload = {
"apply_payload": {
"agent_id": REVIEWER_ID,
"strategy": {
"diagnosis": "Deterministic reconstruction needs exact mutation deltas.",
"guiding_policy": "Replay every guarded transition and fail closed on drift.",
"proximate_objectives": ["exact parity", "zero orphan containers"],
},
"strategy_nodes": [
{
"node_type": "policy",
"title": "Replay exactly",
"body": "Bind the guarded transition to its canonical receipt.",
"rank": 1,
},
{
"node_type": "policy",
"title": "Replay exactly",
"body": "Bind the guarded transition to its canonical receipt.",
"rank": 1,
},
],
},
"private_title": PRIVATE_MARKER,
}
approved = {
**approved,
"id": REVISE_PROPOSAL_ID,
"proposal_type": "revise_strategy",
"payload": payload,
"created_at": "2026-07-14T00:58:00+00:00",
"updated_at": "2026-07-14T01:00:00+00:00",
}
applied = {
**approved,
"status": "applied",
"applied_by_handle": "kb-apply",
"applied_by_agent_id": SERVICE_ID,
"applied_at": "2026-07-14T01:01:00+00:00",
"updated_at": "2026-07-14T01:01:00.000001+00:00",
}
approval = {
**approval,
"proposal_id": REVISE_PROPOSAL_ID,
"proposal_type": "revise_strategy",
"payload": payload,
}
return approved, applied, approval
def applied_envelope(applied: dict) -> dict:
fields = (
"id",
@ -202,6 +266,62 @@ def replay_material() -> dict:
}
def revise_strategy_replay_material(
*, apply_sql_source: str = "exact_executed_sql", version: int = 2
) -> dict:
approved, applied, approval = revise_strategy_proposal_rows()
proposal = applied_envelope(applied)
transaction_timestamp = "2026-07-14T01:00:30+00:00"
canonical_rows = {
"strategies": [
{
"id": RECEIPT_STRATEGY_ID,
"agent_id": REVIEWER_ID,
**proposal["payload"]["apply_payload"]["strategy"],
"version": version,
"active": True,
"created_at": transaction_timestamp,
}
],
"strategy_nodes": [
{
"id": node_id,
"agent_id": REVIEWER_ID,
**node,
"status": "active",
"horizon": None,
"measure": None,
"source_ref": None,
"metadata": {},
"created_at": transaction_timestamp,
"updated_at": transaction_timestamp,
}
for node_id, node in zip(
(RECEIPT_NODE_A_ID, RECEIPT_NODE_B_ID),
proposal["payload"]["apply_payload"]["strategy_nodes"],
strict=True,
)
],
}
apply_sql = rebuild.apply_engine.build_apply_sql(proposal, applied["applied_by_handle"])
receipt = rebuild.replay_receipt.build_replay_receipt(
proposal,
canonical_rows,
apply_sql_sha256=rebuild.replay_receipt.sha256_text(apply_sql),
apply_sql_source=apply_sql_source,
exported_at_utc="2026-07-14T01:02:00+00:00",
)
return {
"artifact": rebuild.MATERIAL_ARTIFACT,
"contract_version": rebuild.MATERIAL_CONTRACT_VERSION,
"sequence": 1,
"approved_proposal": approved,
"approval_snapshot": approval,
"applied_proposal": applied,
"replay_receipt": receipt,
}
def write_bundle(tmp_path: Path, *, material: dict | None = None) -> argparse.Namespace:
dump = tmp_path / "genesis.dump"
dump.write_bytes(b"PGDMPfixture")
@ -238,7 +358,7 @@ def write_bundle(tmp_path: Path, *, material: dict | None = None) -> argparse.Na
ledger=ledger_path,
ledger_sha256=sha256_file(ledger_path),
database="teleo",
image=rebuild.canonical_rebuild.DEFAULT_IMAGE,
image=rebuild.DEFAULT_REBUILD_IMAGE,
container_prefix="teleo-genesis-ledger-test",
tmpfs_mb=256,
startup_timeout=30.0,
@ -302,18 +422,115 @@ def test_dump_hash_mismatch_fails_before_container_start(tmp_path: Path, monkeyp
assert not any(len(command) > 1 and command[1] == "run" for command in commands)
def test_revise_strategy_receipt_fails_closed_before_receipt_rows_are_used(tmp_path: Path) -> None:
material = replay_material()
material["replay_receipt"]["proposal"]["proposal_type"] = "revise_strategy"
material["applied_proposal"]["proposal_type"] = "revise_strategy"
material["approved_proposal"]["proposal_type"] = "revise_strategy"
material["approval_snapshot"]["proposal_type"] = "revise_strategy"
def test_revise_strategy_material_validates_exact_engine_and_canonicalizes_row_order(
tmp_path: Path,
) -> None:
material = revise_strategy_replay_material()
material["replay_receipt"]["canonical_rows"]["strategy_nodes"].reverse()
args = write_bundle(tmp_path, material=material)
with pytest.raises(rebuild.ReconstructionError, match="prior strategy") as exc_info:
entry = rebuild.load_bundle(args).entries[0]
assert entry.applied_proposal["proposal_type"] == "revise_strategy"
assert entry.receipt["apply_engine"]["source"] == "exact_executed_sql"
assert entry.receipt["apply_engine"]["apply_sql_sha256"] == entry.current_apply_sql_sha256
assert entry.receipt["canonical_rows"]["strategy_nodes"] == sorted(
entry.receipt["canonical_rows"]["strategy_nodes"], key=rebuild._canonical_json
)
def test_revise_strategy_material_rejects_reconstructed_apply_engine(tmp_path: Path) -> None:
args = write_bundle(
tmp_path,
material=revise_strategy_replay_material(apply_sql_source="reconstructed_current_engine"),
)
with pytest.raises(rebuild.ReconstructionError) as exc_info:
rebuild.load_bundle(args)
assert exc_info.value.code == "unsupported_mutating_receipt"
assert exc_info.value.code == "mutating_apply_engine_mismatch"
@pytest.mark.parametrize(
"invalid_case",
("extra_strategy_field", "duplicate_node_id", "node_timestamp_drift", "transaction_after_apply"),
)
def test_revise_strategy_receipt_rejects_impossible_poststate(
tmp_path: Path, invalid_case: str
) -> None:
material = revise_strategy_replay_material()
rows = material["replay_receipt"]["canonical_rows"]
if invalid_case == "extra_strategy_field":
rows["strategies"][0]["unexpected"] = "ignored by jsonb_populate_record"
elif invalid_case == "duplicate_node_id":
rows["strategy_nodes"][1]["id"] = rows["strategy_nodes"][0]["id"]
elif invalid_case == "node_timestamp_drift":
rows["strategy_nodes"][0]["updated_at"] = "2026-07-14T01:00:31+00:00"
else:
rows["strategies"][0]["created_at"] = "2026-07-14T01:01:01+00:00"
for row in rows["strategy_nodes"]:
row["created_at"] = "2026-07-14T01:01:01+00:00"
row["updated_at"] = "2026-07-14T01:01:01+00:00"
args = write_bundle(tmp_path, material=material)
with pytest.raises(rebuild.ReconstructionError) as exc_info:
rebuild.load_bundle(args)
assert exc_info.value.code == "invalid_mutating_receipt"
def test_revise_strategy_transition_builders_preserve_historical_prestate(tmp_path: Path) -> None:
entry = rebuild.load_bundle(
write_bundle(tmp_path, material=revise_strategy_replay_material())
).entries[0]
prestate = {
"strategy_ids": [PRIOR_STRATEGY_ID],
"active_strategy_ids": [PRIOR_STRATEGY_ID],
"max_strategy_version": 1,
"strategy_node_ids": [PRIOR_ACTIVE_NODE_ID, PRIOR_RETIRED_NODE_ID],
"non_retired_strategy_node_ids": [PRIOR_ACTIVE_NODE_ID],
}
generated_rows = copy.deepcopy(entry.receipt["canonical_rows"])
generated_rows["strategies"][0]["id"] = "17171717-1717-4717-8717-171717171717"
generated_rows["strategies"][0]["created_at"] = "2026-07-15T01:00:30+00:00"
for index, row in enumerate(generated_rows["strategy_nodes"], start=1):
row["id"] = f"18181818-1818-4818-8818-18181818181{index}"
row["created_at"] = "2026-07-15T01:00:30+00:00"
row["updated_at"] = "2026-07-15T01:00:30+00:00"
sql = rebuild.build_revise_strategy_normalization_sql(entry, prestate, generated_rows)
assert PRIOR_ACTIVE_NODE_ID in sql
assert PRIOR_RETIRED_NODE_ID not in sql
assert PRIOR_STRATEGY_ID in sql
assert RECEIPT_STRATEGY_ID in sql
assert "updated_at = E'2026-07-14T01:00:30+00:00'::timestamptz" in sql
def test_revise_strategy_transition_builders_allow_empty_prestate(tmp_path: Path) -> None:
entry = rebuild.load_bundle(
write_bundle(tmp_path, material=revise_strategy_replay_material(version=1))
).entries[0]
prestate = {
"strategy_ids": [],
"active_strategy_ids": [],
"max_strategy_version": 0,
"strategy_node_ids": [],
"non_retired_strategy_node_ids": [],
}
generated_rows = copy.deepcopy(entry.receipt["canonical_rows"])
generated_rows["strategies"][0]["id"] = "17171717-1717-4717-8717-171717171717"
generated_rows["strategies"][0]["created_at"] = "2026-07-15T01:00:30+00:00"
for index, row in enumerate(generated_rows["strategy_nodes"], start=1):
row["id"] = f"18181818-1818-4818-8818-18181818181{index}"
row["created_at"] = "2026-07-15T01:00:30+00:00"
row["updated_at"] = "2026-07-15T01:00:30+00:00"
sql = rebuild.build_revise_strategy_normalization_sql(entry, prestate, generated_rows)
assert "prior active strategy transition mismatch" not in sql
assert "prior node normalization mismatch" not in sql
assert RECEIPT_STRATEGY_ID in sql
def test_legacy_freeform_receipt_fails_closed(tmp_path: Path) -> None:
@ -331,6 +548,17 @@ def test_legacy_freeform_receipt_fails_closed(tmp_path: Path) -> None:
assert PRIVATE_MARKER not in exc_info.value.safe_message
def test_unknown_proposal_operation_fails_closed_before_replay(tmp_path: Path) -> None:
material = replay_material()
material["replay_receipt"]["proposal"]["proposal_type"] = "delete_claim"
args = write_bundle(tmp_path, material=material)
with pytest.raises(rebuild.ReconstructionError) as exc_info:
rebuild.load_bundle(args)
assert exc_info.value.code == "unsupported_proposal_type"
def test_proposal_metadata_outside_apply_transition_is_rejected(tmp_path: Path) -> None:
material = replay_material()
material["applied_proposal"]["rationale"] = "silently changed after review"
@ -354,6 +582,28 @@ def test_output_cannot_overwrite_a_fixed_guard_or_engine_file(tmp_path: Path) ->
assert exc_info.value.code == "unsafe_output_path"
def test_cli_defaults_to_pinned_image_and_rejects_moving_tag(tmp_path: Path) -> None:
bundle = write_bundle(tmp_path)
argv = [
"--genesis-dump",
str(bundle.genesis_dump),
"--genesis-manifest",
str(bundle.genesis_manifest),
"--ledger",
str(bundle.ledger),
"--ledger-sha256",
bundle.ledger_sha256,
"--output",
str(tmp_path / "receipt.json"),
]
assert rebuild.parse_args(argv).image == rebuild.DEFAULT_REBUILD_IMAGE
with pytest.raises(SystemExit) as exc_info:
rebuild.parse_args([*argv, "--image", "postgres:16-alpine"])
assert exc_info.value.code == 2
def test_unknown_private_field_name_is_not_echoed_in_public_error(tmp_path: Path) -> None:
material = replay_material()
material[PRIVATE_MARKER] = "private value"
@ -484,17 +734,18 @@ create table public.reasoning_tools (
created_at timestamptz not null default now()
);
create table public.strategies (
id uuid primary key,
id uuid primary key default gen_random_uuid(),
agent_id uuid not null references public.agents(id),
diagnosis text,
guiding_policy text,
proximate_objectives jsonb,
version integer not null default 1,
active boolean not null default true,
created_at timestamptz not null default now()
created_at timestamptz not null default now(),
unique (agent_id, version)
);
create table public.strategy_nodes (
id uuid primary key,
id uuid primary key default gen_random_uuid(),
agent_id uuid references public.agents(id),
node_type text,
title text,
@ -508,6 +759,20 @@ create table public.strategy_nodes (
created_at timestamptz not null default now(),
updated_at timestamptz not null default now()
);
create table public.strategy_node_anchors (
id uuid primary key default gen_random_uuid(),
from_node_id uuid not null references public.strategy_nodes(id) on delete cascade,
anchor_role text not null,
to_node_id uuid references public.strategy_nodes(id) on delete cascade,
to_shared_root_id uuid,
belief_id uuid,
claim_id uuid,
source_id uuid,
weight numeric,
note text,
created_at timestamptz not null default now(),
check (num_nonnulls(to_node_id, to_shared_root_id, belief_id, claim_id, source_id) = 1)
);
create table kb_stage.kb_proposals (
id uuid primary key,
proposal_type text not null,
@ -530,10 +795,34 @@ create table kb_stage.kb_proposals (
);
insert into public.agents (id, handle, kind) values
('{REVIEWER_ID}', 'm3ta', 'human');
('{REVIEWER_ID}', 'm3ta', 'human'),
('{OTHER_AGENT_ID}', 'other-agent', 'system');
insert into public.claims (id, type, text, status, confidence, tags, created_by) values
('{CLAIM_A}', 'structural', 'Fixture claim A', 'open', 0.8, array['fixture'], '{REVIEWER_ID}'),
('{CLAIM_B}', 'structural', 'Fixture claim B', 'open', 0.8, array['fixture'], '{REVIEWER_ID}');
insert into public.strategies
(id, agent_id, diagnosis, guiding_policy, proximate_objectives, version, active, created_at)
values
('{PRIOR_STRATEGY_ID}', '{REVIEWER_ID}', 'Prior diagnosis', 'Prior policy', '["prior"]', 1, true,
'2026-07-01T00:00:00+00:00'),
('{OTHER_STRATEGY_ID}', '{OTHER_AGENT_ID}', 'Other diagnosis', 'Other policy', '["other"]', 1, true,
'2026-07-01T00:00:00+00:00');
insert into public.strategy_nodes
(id, agent_id, node_type, title, body, rank, status, created_at, updated_at)
values
('{PRIOR_ACTIVE_NODE_ID}', '{REVIEWER_ID}', 'policy', 'Prior active', 'Retire exactly once', 1,
'active', '2026-07-01T00:00:00+00:00', '2026-07-01T00:00:00+00:00'),
('{PRIOR_RETIRED_NODE_ID}', '{REVIEWER_ID}', 'policy', 'Prior retired', 'Leave timestamp unchanged', 2,
'retired', '2026-06-01T00:00:00+00:00', '2026-06-02T00:00:00+00:00'),
('{OTHER_ACTIVE_NODE_ID}', '{OTHER_AGENT_ID}', 'policy', 'Other active', 'Leave other agent unchanged', 1,
'active', '2026-07-01T00:00:00+00:00', '2026-07-01T00:00:00+00:00'),
('{NULL_AGENT_NODE_ID}', null, 'policy', 'Shared active', 'Leave shared node unchanged', 1,
'active', '2026-07-01T00:00:00+00:00', '2026-07-01T00:00:00+00:00');
insert into public.strategy_node_anchors
(id, from_node_id, anchor_role, to_node_id, weight, note, created_at)
values
('{PRIOR_NODE_ANCHOR_ID}', '{PRIOR_ACTIVE_NODE_ID}', 'serves', '{PRIOR_RETIRED_NODE_ID}', 1.0,
'Existing anchor must survive strategy revision replay', '2026-07-01T00:00:00+00:00');
"""
@ -631,13 +920,17 @@ def source_postgres() -> Iterator[tuple[str, str]]:
"run",
"--rm",
"--detach",
"--network",
"none",
"--name",
container,
"--label",
f"{rebuild.canonical_rebuild.CANARY_LABEL}={container}",
"--env",
f"POSTGRES_DB={database}",
"--env",
"POSTGRES_HOST_AUTH_METHOD=trust",
rebuild.canonical_rebuild.DEFAULT_IMAGE,
rebuild.DEFAULT_REBUILD_IMAGE,
],
text=True,
capture_output=True,
@ -692,14 +985,24 @@ def source_postgres() -> Iterator[tuple[str, str]]:
capture_output=True,
check=False,
)
inspected = subprocess.run(
["docker", "container", "inspect", container],
text=True,
capture_output=True,
check=False,
)
if inspected.returncode == 0:
pytest.fail(f"fixture PostgreSQL container was not removed: {container}")
def test_live_genesis_plus_ledger_command_rebuilds_exactly_and_proves_cleanup(
def capture_source_snapshot(
source_container: str,
database: str,
tmp_path: Path,
source_postgres: tuple[str, str],
) -> None:
source_container, database = source_postgres
genesis_dump = tmp_path / "genesis.dump"
*,
stem: str,
) -> tuple[Path, Path]:
dump = tmp_path / f"{stem}.dump"
dump_result = subprocess.run(
[
"docker",
@ -711,7 +1014,7 @@ def test_live_genesis_plus_ledger_command_rebuilds_exactly_and_proves_cleanup(
"-d",
database,
"--format=custom",
"--file=/tmp/genesis.dump",
f"--file=/tmp/{stem}.dump",
],
text=True,
capture_output=True,
@ -719,15 +1022,130 @@ def test_live_genesis_plus_ledger_command_rebuilds_exactly_and_proves_cleanup(
)
assert dump_result.returncode == 0, dump_result.stderr
copy_result = subprocess.run(
["docker", "cp", f"{source_container}:/tmp/genesis.dump", str(genesis_dump)],
["docker", "cp", f"{source_container}:/tmp/{stem}.dump", str(dump)],
text=True,
capture_output=True,
check=False,
)
assert copy_result.returncode == 0, copy_result.stderr
manifest = tmp_path / f"{stem}-manifest.jsonl"
capture_manifest(source_container, database, manifest)
return dump, manifest
genesis_manifest = tmp_path / "genesis-manifest.jsonl"
capture_manifest(source_container, database, genesis_manifest)
def seed_proposal_and_approval(
source_container: str,
database: str,
approved: dict,
approval: dict,
) -> dict:
sql = f"""begin;
set local standard_conforming_strings = on;
insert into kb_stage.kb_proposals
select seeded.* from jsonb_populate_record(
null::kb_stage.kb_proposals, {rebuild._jsonb_literal(approved)}
) seeded;
insert into kb_stage.kb_proposal_approvals
select seeded.* from jsonb_populate_record(
null::kb_stage.kb_proposal_approvals, {rebuild._jsonb_literal(approval)}
) seeded;
commit;
"""
docker_psql(source_container, database, sql)
raw = docker_psql(
source_container,
database,
rebuild.build_proposal_readback_sql(approved["id"]),
).stdout.strip()
return json.loads(raw)
def run_genesis_ledger_command(
*,
tmp_path: Path,
database: str,
genesis_dump: Path,
genesis_manifest: Path,
material_path: Path,
final_manifest: Path,
artifact_stem: str,
container_prefix: str,
run_number: int,
) -> tuple[subprocess.CompletedProcess[str], dict, Path]:
ledger = {
"artifact": rebuild.LEDGER_ARTIFACT,
"contract_version": rebuild.LEDGER_CONTRACT_VERSION,
"engine": rebuild._engine_hashes(),
"genesis": {
"dump_sha256": sha256_file(genesis_dump),
"parity_manifest_sha256": sha256_file(genesis_manifest),
},
"entries": [
{
"sequence": 1,
"material": material_path.name,
"sha256": sha256_file(material_path),
"replay_material_sha256": json.loads(material_path.read_text(encoding="utf-8"))[
"replay_receipt"
]["hashes"]["replay_material_sha256"],
}
],
"final_parity": {
"manifest": final_manifest.name,
"sha256": sha256_file(final_manifest),
},
}
ledger_path = write_json(tmp_path / f"{artifact_stem}-ledger-run-{run_number}.json", ledger)
output = tmp_path / f"{artifact_stem}-reconstruction-receipt-run-{run_number}.json"
command = [
sys.executable,
str(Path(rebuild.__file__).resolve()),
"--genesis-dump",
str(genesis_dump),
"--genesis-manifest",
str(genesis_manifest),
"--ledger",
str(ledger_path),
"--ledger-sha256",
sha256_file(ledger_path),
"--database",
database,
"--container-prefix",
container_prefix,
"--tmpfs-mb",
"256",
"--max-target-query-ms",
"5000",
"--max-target-source-ratio",
"100",
"--output",
str(output),
]
completed = subprocess.run(
command,
cwd=rebuild.REPO_ROOT,
text=True,
capture_output=True,
check=False,
timeout=180,
)
assert completed.returncode == 0, completed.stderr + completed.stdout
assert output.is_file(), "rebuild command succeeded without writing its receipt"
receipt = json.loads(output.read_text(encoding="utf-8"))
return completed, receipt, output
def test_live_genesis_plus_ledger_command_rebuilds_exactly_and_proves_cleanup(
tmp_path: Path,
source_postgres: tuple[str, str],
) -> None:
source_container, database = source_postgres
genesis_dump, genesis_manifest = capture_source_snapshot(
source_container,
database,
tmp_path,
stem="insert-only-genesis",
)
material = replay_material()
material_path = write_json(tmp_path / "entry-0001.json", material)
@ -753,64 +1171,18 @@ def test_live_genesis_plus_ledger_command_rebuilds_exactly_and_proves_cleanup(
final_manifest = tmp_path / "final-manifest.jsonl"
capture_manifest(source_container, database, final_manifest)
ledger = {
"artifact": rebuild.LEDGER_ARTIFACT,
"contract_version": rebuild.LEDGER_CONTRACT_VERSION,
"engine": rebuild._engine_hashes(),
"genesis": {
"dump_sha256": sha256_file(genesis_dump),
"parity_manifest_sha256": sha256_file(genesis_manifest),
},
"entries": [
{
"sequence": 1,
"material": material_path.name,
"sha256": sha256_file(material_path),
"replay_material_sha256": entry.replay_material_sha256,
}
],
"final_parity": {
"manifest": final_manifest.name,
"sha256": sha256_file(final_manifest),
},
}
ledger_path = write_json(tmp_path / "ledger.json", ledger)
output = tmp_path / "reconstruction-receipt.json"
command = [
sys.executable,
str(Path(rebuild.__file__).resolve()),
"--genesis-dump",
str(genesis_dump),
"--genesis-manifest",
str(genesis_manifest),
"--ledger",
str(ledger_path),
"--ledger-sha256",
sha256_file(ledger_path),
"--database",
database,
"--container-prefix",
"teleo-genesis-ledger-live-test",
"--tmpfs-mb",
"256",
"--max-target-query-ms",
"5000",
"--max-target-source-ratio",
"100",
"--output",
str(output),
]
completed = subprocess.run(
command,
cwd=rebuild.REPO_ROOT,
text=True,
capture_output=True,
check=False,
timeout=180,
completed, receipt, output = run_genesis_ledger_command(
tmp_path=tmp_path,
database=database,
genesis_dump=genesis_dump,
genesis_manifest=genesis_manifest,
material_path=material_path,
final_manifest=final_manifest,
artifact_stem="insert-only",
container_prefix="teleo-genesis-ledger-live-test",
run_number=1,
)
assert completed.returncode == 0, completed.stderr + completed.stdout
receipt = json.loads(output.read_text(encoding="utf-8"))
assert receipt["status"] == "pass"
assert receipt["genesis_parity"]["status"] == "pass"
assert receipt["ledger"]["entries"][0]["status"] == "pass"
@ -831,3 +1203,132 @@ def test_live_genesis_plus_ledger_command_rebuilds_exactly_and_proves_cleanup(
check=False,
)
assert inspect.returncode != 0
def test_live_revise_strategy_rebuild_is_exact_repeatable_and_leaves_no_container(
tmp_path: Path,
source_postgres: tuple[str, str],
) -> None:
source_container, database = source_postgres
genesis_dump, genesis_manifest = capture_source_snapshot(
source_container,
database,
tmp_path,
stem="revise-strategy-genesis",
)
approved, _, approval = revise_strategy_proposal_rows()
pre_apply = seed_proposal_and_approval(
source_container,
database,
approved,
approval,
)
apply_sql = rebuild.apply_engine.build_apply_sql(pre_apply["proposal"], "kb-apply")
docker_psql(source_container, database, apply_sql, role="kb_apply")
post_apply = json.loads(
docker_psql(
source_container,
database,
rebuild.build_proposal_readback_sql(REVISE_PROPOSAL_ID),
).stdout.strip()
)
applied_proposal = post_apply["proposal"]
proposal_envelope = applied_envelope(applied_proposal)
source_receipt_path, receipt = rebuild.apply_engine.capture_replay_receipt(
argparse.Namespace(
container=source_container,
role="kb_apply",
host="127.0.0.1",
db=database,
receipt_out=str(tmp_path / "source-revise-strategy-private-receipt.json"),
receipt_dir=None,
),
proposal_envelope,
"",
apply_sql=apply_sql,
)
assert source_receipt_path.is_file()
assert stat.S_IMODE(source_receipt_path.stat().st_mode) == 0o600
material = {
"artifact": rebuild.MATERIAL_ARTIFACT,
"contract_version": rebuild.MATERIAL_CONTRACT_VERSION,
"sequence": 1,
"approved_proposal": pre_apply["proposal"],
"approval_snapshot": pre_apply["approval_snapshot"],
"applied_proposal": applied_proposal,
"replay_receipt": receipt,
}
material_path = write_json(tmp_path / "revise-strategy-entry-0001.json", material)
final_manifest = tmp_path / "revise-strategy-final-manifest.jsonl"
capture_manifest(source_container, database, final_manifest)
runs = [
run_genesis_ledger_command(
tmp_path=tmp_path,
database=database,
genesis_dump=genesis_dump,
genesis_manifest=genesis_manifest,
material_path=material_path,
final_manifest=final_manifest,
artifact_stem="revise-strategy",
container_prefix="teleo-genesis-ledger-revise",
run_number=run_number,
)
for run_number in (1, 2)
]
for completed, reconstruction, output in runs:
assert reconstruction["status"] == "pass"
assert reconstruction["genesis_parity"]["status"] == "pass"
assert reconstruction["final_parity"]["status"] == "pass"
entry_summary = reconstruction["ledger"]["entries"][0]
assert entry_summary["proposal_type"] == "revise_strategy"
assert entry_summary["proposal_seed_exact"] is True
assert entry_summary["canonical_seed_exact"] is False
assert entry_summary["seed_exact"] is False
assert entry_summary["guarded_apply_executed"] is True
assert entry_summary["mutating_prestate_captured"] is True
assert entry_summary["mutating_delta_validated"] is True
assert entry_summary["mutating_poststate_normalized"] is True
assert entry_summary["proposal_exact"] is True
assert entry_summary["canonical_rows_exact"] is True
assert reconstruction["cleanup"]["container_absent"] is True
assert reconstruction["safety"]["network_access_available_to_container"] is False
assert reconstruction["safety"]["private_replay_material_emitted"] is False
assert PRIVATE_MARKER not in completed.stdout
assert PRIVATE_MARKER not in output.read_text(encoding="utf-8")
assert stat.S_IMODE(output.stat().st_mode) == 0o600
container_name = reconstruction["container"]["name"]
inspect = subprocess.run(
["docker", "container", "inspect", container_name],
text=True,
capture_output=True,
check=False,
)
assert inspect.returncode != 0
labeled = subprocess.run(
[
"docker",
"ps",
"-aq",
"--filter",
f"label={rebuild.canonical_rebuild.CANARY_LABEL}={container_name}",
],
text=True,
capture_output=True,
check=False,
)
assert labeled.returncode == 0
assert labeled.stdout.strip() == ""
first = runs[0][1]
second = runs[1][1]
assert first["inputs"] == second["inputs"]
assert first["ledger"]["entries"][0]["material_sha256"] == second["ledger"]["entries"][0][
"material_sha256"
]
assert first["ledger"]["entries"][0]["replay_material_sha256"] == second["ledger"]["entries"][0][
"replay_material_sha256"
]