diff --git a/docs/kb-rebuild-and-recompile.md b/docs/kb-rebuild-and-recompile.md index acbf408..25be5a6 100644 --- a/docs/kb-rebuild-and-recompile.md +++ b/docs/kb-rebuild-and-recompile.md @@ -264,6 +264,13 @@ worker receipt, and publishes a mode-`0600` file once. An identical rerun is a no-op; drift or output aliasing is refused. Already-applied legacy approvals whose exact approved row was never retained remain explicitly unavailable. +The rollback boundary is this transition capability only. Stop review/apply +workers, revoke and drop +`kb_stage.export_applied_proposal_replay_rows(uuid)`, then deploy the prior +revision and rerun its `kb_apply_prereqs.sql` before restarting workers. Keep +the nullable `approved_proposal_snapshot` column and any captured rows as inert +evidence; dropping them is neither required nor safe for rollback. + ## Genesis Plus Strict Ledger: Working Deterministic Slice `ops/run_local_genesis_ledger_rebuild.py` now executes the first exact diff --git a/ops/run_local_genesis_ledger_rebuild.py b/ops/run_local_genesis_ledger_rebuild.py index 74accfb..3fd39af 100644 --- a/ops/run_local_genesis_ledger_rebuild.py +++ b/ops/run_local_genesis_ledger_rebuild.py @@ -720,6 +720,7 @@ def _validate_entry_material( 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, @@ -728,23 +729,34 @@ def _validate_entry_material( apply_sql_source=apply_metadata.get("source"), exported_at_utc=receipt.get("exported_at_utc"), ) - except (KeyError, TypeError, ValueError) as exc: - raise ReconstructionError( - "invalid_replay_receipt", "replay receipt failed strict payload and exact-row validation" - ) from exc + 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 receipt.get("hashes") != rebuilt["hashes"]: + 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["expected_row_counts"]: + 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["actual_row_counts"]: + if receipt.get("actual_row_counts") != rebuilt.get("actual_row_counts"): raise ReconstructionError("replay_receipt_count_mismatch", "replay receipt actual counts are invalid") - actual_replay_hash = rebuilt["hashes"]["replay_material_sha256"] + 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 = {**receipt, "canonical_rows": rebuilt["canonical_rows"]} + receipt = rebuilt + proposal = receipt["proposal"] approved = material["approved_proposal"] approval = material["approval_snapshot"] diff --git a/scripts/export_kb_transition_replay_material.py b/scripts/export_kb_transition_replay_material.py index 1fd95a8..a9749dd 100644 --- a/scripts/export_kb_transition_replay_material.py +++ b/scripts/export_kb_transition_replay_material.py @@ -9,13 +9,15 @@ function, and atomically publishes deterministic genesis-ledger input. from __future__ import annotations import argparse +import hashlib import json import os +import re import stat import sys import tempfile import uuid -from datetime import datetime +from datetime import datetime, timezone from pathlib import Path from typing import Any @@ -92,6 +94,11 @@ RECEIPT_PROPOSAL_FIELDS = ( "applied_by_agent_id", "applied_at", ) +RECEIPT_PROPOSAL_FIELD_SET = frozenset(RECEIPT_PROPOSAL_FIELDS) +LEGACY_RECEIPT_TIMESTAMP_FIELDS = frozenset({"captured_at", "created_at", "updated_at", "reviewed_at", "applied_at"}) +LEGACY_POSTGRES_UTC_TIMESTAMP = re.compile( + r"^[0-9]{4}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2}(?:\.[0-9]{1,6})?\+00$" +) class ExportError(RuntimeError): @@ -106,6 +113,10 @@ def _same_json(left: Any, right: Any) -> bool: return _canonical_json(left) == _canonical_json(right) +def _sha256_json(value: Any) -> str: + return hashlib.sha256(_canonical_json(value).encode("utf-8")).hexdigest() + + def _reject_duplicate_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]: result: dict[str, Any] = {} for key, value in pairs: @@ -187,6 +198,17 @@ def _canonical_timestamp(value: Any, *, label: str, nullable: bool = False) -> s raise ExportError(f"{label} must be a timezone-aware timestamp") from exc +def _canonical_legacy_postgres_timestamp(value: Any, *, label: str) -> str: + if not isinstance(value, str) or not LEGACY_POSTGRES_UTC_TIMESTAMP.fullmatch(value): + raise ExportError(f"{label} must use the exact legacy PostgreSQL UTC timestamp format") + timestamp_format = "%Y-%m-%d %H:%M:%S.%f%z" if "." in value else "%Y-%m-%d %H:%M:%S%z" + try: + parsed = datetime.strptime(value + ":00", timestamp_format) + except ValueError as exc: + raise ExportError(f"{label} must be a valid legacy PostgreSQL UTC timestamp") from exc + return parsed.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%fZ") + + def _normalize_proposal(value: Any, *, label: str) -> dict[str, Any]: row = dict(_require_exact_fields(value, PROPOSAL_FIELDS, label=label)) for field in PROPOSAL_UUID_FIELDS: @@ -236,6 +258,186 @@ def _normalize_approval(value: Any, *, label: str) -> dict[str, Any]: return row +def _normalize_receipt_proposal(value: Any, *, label: str) -> dict[str, Any]: + row = dict(_require_exact_fields(value, RECEIPT_PROPOSAL_FIELD_SET, label=label)) + for field in RECEIPT_PROPOSAL_FIELD_SET & PROPOSAL_UUID_FIELDS: + row[field] = _canonical_uuid( + row[field], + label=f"{label}.{field}", + nullable=field != "id", + ) + for field in RECEIPT_PROPOSAL_FIELD_SET & PROPOSAL_TIMESTAMP_FIELDS: + row[field] = _canonical_timestamp( + row[field], + label=f"{label}.{field}", + nullable=False, + ) + return row + + +def _normalize_legacy_receipt_proposal(value: Any, *, label: str) -> dict[str, Any]: + row = dict(_require_exact_fields(value, RECEIPT_PROPOSAL_FIELD_SET, label=label)) + for field in RECEIPT_PROPOSAL_FIELD_SET & PROPOSAL_UUID_FIELDS: + row[field] = _canonical_uuid( + row[field], + label=f"{label}.{field}", + nullable=field != "id", + ) + for field in RECEIPT_PROPOSAL_FIELD_SET & PROPOSAL_TIMESTAMP_FIELDS: + row[field] = _canonical_legacy_postgres_timestamp( + row[field], + label=f"{label}.{field}", + ) + return row + + +def _normalize_legacy_receipt_rows(value: Any) -> dict[str, list[dict[str, Any]]]: + if not isinstance(value, dict): + raise ExportError("worker replay receipt canonical rows 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 ExportError("worker replay receipt canonical rows must map table names to row arrays") + normalized_rows: list[dict[str, Any]] = [] + for index, row in enumerate(rows): + normalized_row: dict[str, Any] = {} + for field, field_value in row.items(): + path = f"replay_receipt.canonical_rows.{table}[{index}].{field}" + if field in LEGACY_RECEIPT_TIMESTAMP_FIELDS and field_value is not None: + normalized_row[field] = _canonical_legacy_postgres_timestamp(field_value, label=path) + else: + try: + normalized_row[field] = replay_receipt.normalize_typed_value( + field, + field_value, + path=path, + ) + except ValueError as exc: + raise ExportError("worker replay receipt contains an invalid typed row value") from exc + normalized_rows.append(normalized_row) + normalized[table] = normalized_rows + return normalized + + +def _has_legacy_postgres_timestamp(proposal: dict[str, Any], rows: Any) -> bool: + values = [proposal.get("reviewed_at"), proposal.get("applied_at")] + if isinstance(rows, dict): + values.extend( + row.get(field) + for table_rows in rows.values() + if isinstance(table_rows, list) + for row in table_rows + if isinstance(row, dict) + for field in LEGACY_RECEIPT_TIMESTAMP_FIELDS + ) + return any(isinstance(value, str) and LEGACY_POSTGRES_UTC_TIMESTAMP.fullmatch(value) for value in values) + + +def _validate_legacy_worker_replay_receipt( + receipt: dict[str, Any], + *, + expected_proposal_id: str | None, +) -> dict[str, Any]: + proposal = receipt.get("proposal") + rows = receipt.get("canonical_rows") + apply_metadata = receipt.get("apply_engine") + if not isinstance(proposal, dict) or not isinstance(rows, dict) or not isinstance(apply_metadata, dict): + raise ExportError("worker replay receipt is missing proposal, rows, or apply metadata") + if not _has_legacy_postgres_timestamp(proposal, rows): + raise ExportError("worker replay receipt is not a supported legacy PostgreSQL rendering") + + normalized_proposal = _normalize_legacy_receipt_proposal(proposal, label="replay_receipt.proposal") + if any( + proposal[field] is not None and proposal[field] != normalized_proposal[field] + for field in RECEIPT_PROPOSAL_FIELD_SET & PROPOSAL_UUID_FIELDS + ): + raise ExportError("legacy worker replay receipt proposal UUIDs are not canonical") + if expected_proposal_id is not None and normalized_proposal["id"] != _canonical_uuid( + expected_proposal_id, + label="expected_proposal_id", + ): + raise ExportError("worker replay receipt proposal id does not match") + normalized_rows = _normalize_legacy_receipt_rows(rows) + if any( + row.get(field) is not None and row.get(field) != normalized_rows[table][index].get(field) + for table, table_rows in rows.items() + for index, row in enumerate(table_rows) + for field in row + if field not in LEGACY_RECEIPT_TIMESTAMP_FIELDS + ): + raise ExportError("legacy worker replay receipt row UUIDs are not canonical") + exported_at_utc = _canonical_timestamp( + receipt.get("exported_at_utc"), + label="replay_receipt.exported_at_utc", + ) + try: + normalized_receipt = replay_receipt.build_replay_receipt( + normalized_proposal, + normalized_rows, + apply_sql_sha256=apply_metadata.get("apply_sql_sha256"), + apply_sql_source=apply_metadata.get("source"), + exported_at_utc=exported_at_utc, + ) + except (KeyError, TypeError, ValueError) as exc: + raise ExportError("legacy worker replay receipt failed strict payload and row validation") from exc + + expected_counts = normalized_receipt["expected_row_counts"] + if any(not isinstance(rows.get(table), list) for table in expected_counts): + raise ExportError("legacy worker replay receipt is missing an expected row array") + raw_rows = {table: sorted(rows.get(table, []), key=_canonical_json) for table in expected_counts} + actual_counts = {table: len(table_rows) for table, table_rows in raw_rows.items()} + raw_apply_metadata = { + "apply_sql_sha256": apply_metadata.get("apply_sql_sha256"), + "source": apply_metadata.get("source"), + } + replay_material = { + "receipt_contract_version": replay_receipt.RECEIPT_CONTRACT_VERSION, + "apply_engine": raw_apply_metadata, + "proposal": proposal, + "canonical_rows": raw_rows, + } + raw_apply_payload = proposal.get("payload", {}).get("apply_payload") + expected_receipt = { + **replay_material, + "artifact": "kb_apply_replay_receipt", + "exported_at_utc": receipt.get("exported_at_utc"), + "expected_row_counts": expected_counts, + "actual_row_counts": actual_counts, + "checks": normalized_receipt["checks"], + "hashes": { + "apply_payload_sha256": _sha256_json(raw_apply_payload), + "proposal_payload_sha256": _sha256_json(proposal.get("payload")), + "canonical_rows_sha256": _sha256_json(raw_rows), + "row_sha256": {table: [_sha256_json(row) for row in table_rows] for table, table_rows in raw_rows.items()}, + "table_sha256": {table: _sha256_json(table_rows) for table, table_rows in raw_rows.items()}, + "replay_material_sha256": _sha256_json(replay_material), + }, + "replay_ready": True, + "privacy": "private receipt; may contain claim bodies and source excerpts; do not commit", + } + if not _same_json(receipt, expected_receipt): + raise ExportError("legacy worker replay receipt hashes or contract fields are inconsistent") + return normalized_receipt + + +def validate_worker_replay_receipt( + receipt: dict[str, Any], + *, + expected_proposal_id: str | None = None, +) -> dict[str, Any]: + """Validate a current receipt or one exact legacy PostgreSQL timestamp rendering.""" + try: + return replay_receipt.validate_replay_receipt( + receipt, + expected_proposal_id=expected_proposal_id, + ) + except ValueError: + return _validate_legacy_worker_replay_receipt( + receipt, + expected_proposal_id=expected_proposal_id, + ) + + def _parse_timestamp(value: str) -> datetime: return datetime.fromisoformat(value.replace("Z", "+00:00")) @@ -298,14 +500,18 @@ def validate_material( if not isinstance(receipt, dict): raise ExportError("replay_receipt must be an object") try: - replay_receipt.validate_replay_receipt(receipt, expected_proposal_id=proposal_id) - except ValueError as exc: + normalized_receipt = validate_worker_replay_receipt(receipt, expected_proposal_id=proposal_id) + except (ExportError, ValueError) as exc: raise ExportError("worker replay receipt failed its exact row-level contract") from exc receipt_proposal = receipt.get("proposal") if not isinstance(receipt_proposal, dict): raise ExportError("worker replay receipt has no proposal envelope") + normalized_receipt_proposal = _normalize_receipt_proposal( + normalized_receipt.get("proposal"), + label="replay_receipt.proposal", + ) applied_envelope = {field: applied[field] for field in RECEIPT_PROPOSAL_FIELDS} - if not _same_json(receipt_proposal, applied_envelope): + if not _same_json(normalized_receipt_proposal, applied_envelope): raise ExportError("worker replay receipt does not match the exact applied proposal row") created_at = _parse_timestamp(approved["created_at"]) diff --git a/scripts/kb_apply_prereqs.sql b/scripts/kb_apply_prereqs.sql index 7998133..96b44e5 100644 --- a/scripts/kb_apply_prereqs.sql +++ b/scripts/kb_apply_prereqs.sql @@ -633,6 +633,7 @@ create or replace function kb_stage.approve_strict_proposal( language plpgsql security definer set search_path = pg_catalog, pg_temp +set timezone = 'UTC' as $function$ declare v_row kb_stage.kb_proposals%rowtype; @@ -734,6 +735,7 @@ create or replace function kb_stage.assert_approved_proposal( language plpgsql security definer set search_path = pg_catalog, pg_temp +set timezone = 'UTC' as $function$ begin perform 1 @@ -779,6 +781,7 @@ create or replace function kb_stage.finish_approved_proposal( language plpgsql security definer set search_path = pg_catalog, pg_temp +set timezone = 'UTC' as $function$ declare v_row kb_stage.kb_proposals%rowtype; @@ -862,6 +865,7 @@ language plpgsql stable security definer set search_path = pg_catalog, pg_temp +set timezone = 'UTC' as $function$ declare v_rows jsonb; @@ -1144,8 +1148,11 @@ begin and ( owner_role.rolname <> 'kb_gate_owner' or not procedure.prosecdef - or pg_catalog.array_to_string(procedure.proconfig, ',') is distinct from - 'search_path=pg_catalog, pg_temp' + or pg_catalog.cardinality(procedure.proconfig) <> 2 + or not procedure.proconfig @> array[ + 'search_path=pg_catalog, pg_temp'::text, + 'TimeZone=UTC'::text + ] or ( procedure.proname = 'export_applied_proposal_replay_rows' and ( diff --git a/tests/test_export_kb_transition_replay_material.py b/tests/test_export_kb_transition_replay_material.py index 7ca2b71..30b1595 100644 --- a/tests/test_export_kb_transition_replay_material.py +++ b/tests/test_export_kb_transition_replay_material.py @@ -2,6 +2,7 @@ from __future__ import annotations import argparse import copy +import hashlib import json import os import stat @@ -104,6 +105,41 @@ def _material() -> dict: } +def _rehash_worker_receipt(material: dict) -> dict: + receipt = material["replay_receipt"] + + def sha256_json(value) -> str: + encoded = json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=True).encode("utf-8") + return hashlib.sha256(encoded).hexdigest() + + rows = receipt["canonical_rows"] + replay_material = { + "receipt_contract_version": receipt["receipt_contract_version"], + "apply_engine": receipt["apply_engine"], + "proposal": receipt["proposal"], + "canonical_rows": rows, + } + receipt["hashes"] = { + "apply_payload_sha256": sha256_json(receipt["proposal"]["payload"]["apply_payload"]), + "proposal_payload_sha256": sha256_json(receipt["proposal"]["payload"]), + "canonical_rows_sha256": sha256_json(rows), + "row_sha256": {table: [sha256_json(row) for row in table_rows] for table, table_rows in rows.items()}, + "table_sha256": {table: sha256_json(table_rows) for table, table_rows in rows.items()}, + "replay_material_sha256": sha256_json(replay_material), + } + return material + + +def _legacy_space_timestamp_material() -> dict: + material = copy.deepcopy(_material()) + receipt = material["replay_receipt"] + receipt["proposal"]["reviewed_at"] = "2026-07-18 10:01:00+00" + receipt["proposal"]["applied_at"] = "2026-07-18 10:02:00+00" + receipt["canonical_rows"]["claim_edges"][0]["created_at"] = "2026-07-18 10:01:30+00" + receipt["exported_at_utc"] = "2026-07-18T10:03:00+00:00" + return _rehash_worker_receipt(material) + + def test_material_matches_genesis_replay_row_contract() -> None: material = exporter.validate_material( _material(), @@ -123,6 +159,78 @@ def test_material_matches_genesis_replay_row_contract() -> None: ) +def test_legacy_postgres_worker_receipt_exports_and_reconstructs_without_rehashing( + tmp_path: Path, +) -> None: + material = _legacy_space_timestamp_material() + legacy_hash = material["replay_receipt"]["hashes"]["replay_material_sha256"] + + validated = exporter.validate_material( + material, + expected_proposal_id=PROPOSAL_ID, + expected_sequence=1, + ) + assert validated["replay_receipt"]["proposal"]["reviewed_at"] == "2026-07-18 10:01:00+00" + assert validated["replay_receipt"]["hashes"]["replay_material_sha256"] == legacy_hash + + material_path = tmp_path / "legacy-transition.json" + material_path.write_text(json.dumps(validated), encoding="utf-8") + entry = genesis_rebuild._validate_entry_material( + validated, + expected_sequence=1, + expected_replay_hash=legacy_hash, + material_path=material_path, + material_sha256=hashlib.sha256(material_path.read_bytes()).hexdigest(), + ) + assert entry.replay_material_sha256 == legacy_hash + assert entry.receipt["proposal"]["reviewed_at"] == "2026-07-18T10:01:00.000000Z" + assert entry.receipt["canonical_rows"]["claim_edges"][0]["created_at"] == ("2026-07-18T10:01:30.000000Z") + + tampered = copy.deepcopy(material) + tampered["replay_receipt"]["hashes"]["replay_material_sha256"] = "0" * 64 + with pytest.raises(exporter.ExportError, match="worker replay receipt"): + exporter.validate_material(tampered, expected_proposal_id=PROPOSAL_ID) + + +@pytest.mark.parametrize( + "separator", + ["X", "_", "\U0001f642"], + ids=["x", "underscore", "emoji"], +) +def test_current_material_rejects_noncanonical_timestamp_separators(separator: str) -> None: + material = _material() + invalid = material["approved_proposal"]["reviewed_at"].replace("T", separator, 1) + material["approved_proposal"]["reviewed_at"] = invalid + material["applied_proposal"]["reviewed_at"] = invalid + material["approval_snapshot"]["reviewed_at"] = invalid + + with pytest.raises(exporter.ExportError, match="timezone-aware timestamp"): + exporter.validate_material(material, expected_proposal_id=PROPOSAL_ID) + + +@pytest.mark.parametrize( + "separator", + ["X", "_", "\U0001f642"], + ids=["x", "underscore", "emoji"], +) +@pytest.mark.parametrize("target", ["proposal", "canonical_row"]) +def test_rehashed_legacy_receipt_rejects_hybrid_timestamp_separators( + separator: str, + target: str, +) -> None: + material = _legacy_space_timestamp_material() + receipt = material["replay_receipt"] + if target == "proposal": + receipt["proposal"]["applied_at"] = receipt["proposal"]["applied_at"].replace(" ", separator, 1) + else: + row = receipt["canonical_rows"]["claim_edges"][0] + row["created_at"] = row["created_at"].replace(" ", separator, 1) + _rehash_worker_receipt(material) + + with pytest.raises(exporter.ExportError, match="worker replay receipt"): + exporter.validate_material(material, expected_proposal_id=PROPOSAL_ID) + + @pytest.mark.parametrize( ("mutation", "message"), [ diff --git a/tests/test_kb_apply_prereqs.py b/tests/test_kb_apply_prereqs.py index 63f4ac6..b5adf99 100644 --- a/tests/test_kb_apply_prereqs.py +++ b/tests/test_kb_apply_prereqs.py @@ -1,5 +1,6 @@ from __future__ import annotations +import json import os import shutil import subprocess @@ -18,6 +19,7 @@ DATABASE = "teleo_clone" LEGACY_APPROVED_ID = "aaaaaaaa-0000-4000-8000-000000000001" LEGACY_APPLIED_ID = "aaaaaaaa-0000-4000-8000-000000000002" FRESH_PROPOSAL_ID = "aaaaaaaa-0000-4000-8000-000000000003" +TIMEZONE_PROPOSAL_ID = "aaaaaaaa-0000-4000-8000-000000000004" REVIEWER_ID = "11111111-1111-1111-1111-111111111111" STALE_REVIEWER_ID = "22222222-2222-2222-2222-222222222222" LEGACY_REVIEWER_HANDLE = "m3ta" @@ -556,6 +558,111 @@ order by 1; assert exported[-3:] == ["applied_proposal", "approval_snapshot", "approved_proposal"] +def test_gate_functions_are_timezone_independent_across_connections( + migrated_postgres: str, +) -> None: + payload = '{"apply_payload":{"from_claim":"aaaaaaaa-1000-4000-8000-000000000004"}}' + _psql( + migrated_postgres, + f""" +insert into kb_stage.kb_proposals (id, proposal_type, status, payload, updated_at) +values ('{TIMEZONE_PROPOSAL_ID}'::uuid, 'add_edge', 'pending_review', '{payload}'::jsonb, + '2026-07-18T12:00:00Z'); +set timezone = 'America/Los_Angeles'; +set session authorization kb_review; +select kb_stage.approve_strict_proposal( + '{TIMEZONE_PROPOSAL_ID}'::uuid, 'add_edge', '{payload}'::jsonb, '{CANONICAL_REVIEWER_HANDLE}', + 'Cross-timezone exact approval snapshot.' +); +""", + ) + + asserted_and_finished = _psql( + migrated_postgres, + f""" +set timezone = 'Asia/Tokyo'; +set session authorization kb_apply; +select kb_stage.assert_approved_proposal( + proposal.id, proposal.proposal_type, proposal.payload, proposal.reviewed_by_handle, + proposal.reviewed_by_agent_id, proposal.reviewed_at, proposal.review_note +) +from kb_stage.kb_proposals proposal where proposal.id = '{TIMEZONE_PROPOSAL_ID}'::uuid; +select kb_stage.finish_approved_proposal( + proposal.id, proposal.proposal_type, proposal.payload, proposal.reviewed_by_handle, + proposal.reviewed_by_agent_id, proposal.reviewed_at, proposal.review_note, 'kb-apply' +) +from kb_stage.kb_proposals proposal where proposal.id = '{TIMEZONE_PROPOSAL_ID}'::uuid; +""", + ) + assert '"status": "applied"' in asserted_and_finished.stdout + + exported = json.loads( + _psql( + migrated_postgres, + f""" +set timezone = 'Pacific/Auckland'; +set session authorization kb_apply; +select kb_stage.export_applied_proposal_replay_rows('{TIMEZONE_PROPOSAL_ID}'::uuid)::text; +""", + ).stdout.splitlines()[-1] + ) + approved = exported["approved_proposal"] + applied = exported["applied_proposal"] + approval = exported["approval_snapshot"] + assert approved["reviewed_at"].endswith("+00:00") + assert applied["reviewed_at"] == approved["reviewed_at"] + assert approval["reviewed_at"] == approved["reviewed_at"] + assert applied["status"] == "applied" + + +def test_transition_export_rollback_preserves_evidence_and_reinstalls_cleanly() -> None: + with _postgres_clone() as container: + _apply_migration(container) + before = _psql( + container, + f""" +select approved_proposal_snapshot::text + from kb_stage.kb_proposal_approvals + where proposal_id = '{LEGACY_APPROVED_ID}'::uuid; +""", + ).stdout.strip() + + _psql( + container, + """ +revoke all on function kb_stage.export_applied_proposal_replay_rows(uuid) + from public, kb_gate_owner, kb_apply, kb_review; +drop function kb_stage.export_applied_proposal_replay_rows(uuid); +""", + ) + rolled_back = _psql( + container, + f""" +select pg_catalog.to_regprocedure( + 'kb_stage.export_applied_proposal_replay_rows(uuid)' + ) is null; +select approved_proposal_snapshot::text + from kb_stage.kb_proposal_approvals + where proposal_id = '{LEGACY_APPROVED_ID}'::uuid; +""", + ).stdout.splitlines() + assert rolled_back == ["t", before] + + _apply_migration(container) + restored = _psql( + container, + f""" +select pg_catalog.to_regprocedure( + 'kb_stage.export_applied_proposal_replay_rows(uuid)' + ) is not null; +select approved_proposal_snapshot::text + from kb_stage.kb_proposal_approvals + where proposal_id = '{LEGACY_APPROVED_ID}'::uuid; +""", + ).stdout.splitlines() + assert restored == ["t", before] + + def _catalog_snapshot(container: str) -> str: return _psql( container, @@ -689,7 +796,14 @@ select protected_role.rolname || '|' || protected_schema.nspname || '|' || migrated_postgres, """ select procedure.proname || '(' || pg_catalog.oidvectortypes(procedure.proargtypes) || ')|' || - owner_role.rolname || '|' || procedure.prosecdef || '|' || procedure.proconfig::text + owner_role.rolname || '|' || procedure.prosecdef || '|' || + ( + pg_catalog.cardinality(procedure.proconfig) = 2 + and procedure.proconfig @> array[ + 'search_path=pg_catalog, pg_temp'::text, + 'TimeZone=UTC'::text + ] + ) from pg_catalog.pg_proc procedure join pg_catalog.pg_namespace namespace on namespace.oid = procedure.pronamespace join pg_catalog.pg_roles owner_role on owner_role.oid = procedure.proowner @@ -702,12 +816,12 @@ select procedure.proname || '(' || pg_catalog.oidvectortypes(procedure.proargtyp ) ) assert function_rows == { - 'approve_strict_proposal(uuid, text, jsonb, text, text)|kb_gate_owner|true|{"search_path=pg_catalog, pg_temp"}', + "approve_strict_proposal(uuid, text, jsonb, text, text)|kb_gate_owner|true|true", "assert_approved_proposal(uuid, text, jsonb, text, uuid, timestamp with time zone, text)|" - 'kb_gate_owner|true|{"search_path=pg_catalog, pg_temp"}', + "kb_gate_owner|true|true", "finish_approved_proposal(uuid, text, jsonb, text, uuid, timestamp with time zone, text, text)|" - 'kb_gate_owner|true|{"search_path=pg_catalog, pg_temp"}', - 'export_applied_proposal_replay_rows(uuid)|kb_gate_owner|true|{"search_path=pg_catalog, pg_temp"}', + "kb_gate_owner|true|true", + "export_applied_proposal_replay_rows(uuid)|kb_gate_owner|true|true", } assert ( _psql( diff --git a/tests/test_run_local_genesis_ledger_rebuild.py b/tests/test_run_local_genesis_ledger_rebuild.py index 8e7df65..0030404 100644 --- a/tests/test_run_local_genesis_ledger_rebuild.py +++ b/tests/test_run_local_genesis_ledger_rebuild.py @@ -1816,6 +1816,45 @@ def test_live_genesis_plus_ledger_command_rebuilds_exactly_and_proves_cleanup( rebuild.build_applied_timestamp_normalization_sql(entry), ) + worker_args = argparse.Namespace( + container=source_container, + role="kb_apply", + host="127.0.0.1", + db=database, + proposal_id=entry.applied_proposal["id"], + receipt_out=str(tmp_path / "worker-replay-receipt.json"), + receipt_dir=None, + ) + worker_proposal = rebuild.apply_engine.load_proposal(worker_args, "") + assert "T" in worker_proposal["reviewed_at"] + assert worker_proposal["reviewed_at"].endswith("Z") + _, worker_receipt = rebuild.apply_engine.capture_replay_receipt( + worker_args, + worker_proposal, + "", + apply_sql=entry.current_apply_sql, + ) + transition_rows = rebuild.replay_exporter.fetch_transition_rows( + worker_args, + "", + entry.applied_proposal["id"], + ) + exported_material = rebuild.replay_exporter.validate_material( + { + "artifact": rebuild.replay_exporter.MATERIAL_ARTIFACT, + "contract_version": rebuild.replay_exporter.MATERIAL_CONTRACT_VERSION, + "sequence": 1, + "approved_proposal": transition_rows["approved_proposal"], + "approval_snapshot": transition_rows["approval_snapshot"], + "applied_proposal": transition_rows["applied_proposal"], + "replay_receipt": worker_receipt, + }, + expected_proposal_id=entry.applied_proposal["id"], + expected_sequence=1, + ) + assert exported_material["replay_receipt"]["hashes"] == worker_receipt["hashes"] + assert stat.S_IMODE(Path(worker_args.receipt_out).stat().st_mode) == 0o600 + final_manifest = tmp_path / "final-manifest.jsonl" capture_manifest(source_container, database, final_manifest) completed, receipt, output = run_genesis_ledger_command(