From 40344c91d7c141c6f9993eb3d79ea29c6b33db9d Mon Sep 17 00:00:00 2001 From: twentyOne2x Date: Wed, 15 Jul 2026 04:00:53 +0200 Subject: [PATCH] enforce revise strategy approval timestamp provenance --- docs/kb-rebuild-and-recompile.md | 2 + ops/run_local_genesis_ledger_rebuild.py | 72 ++++------ .../test_run_local_genesis_ledger_rebuild.py | 129 +++++++++++++++--- 3 files changed, 135 insertions(+), 68 deletions(-) diff --git a/docs/kb-rebuild-and-recompile.md b/docs/kb-rebuild-and-recompile.md index 00996dc..b21eb56 100644 --- a/docs/kb-rebuild-and-recompile.md +++ b/docs/kb-rebuild-and-recompile.md @@ -240,6 +240,8 @@ The exact v1 claim ceiling is intentionally bounded: 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`. + It must also fall within the immutable, timezone-aware interval + `reviewed_at <= transaction timestamp <= applied_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 diff --git a/ops/run_local_genesis_ledger_rebuild.py b/ops/run_local_genesis_ledger_rebuild.py index a3361e1..a500fc6 100644 --- a/ops/run_local_genesis_ledger_rebuild.py +++ b/ops/run_local_genesis_ledger_rebuild.py @@ -49,12 +49,9 @@ MATERIAL_ARTIFACT = "teleo_strict_proposal_replay_material" LEDGER_CONTRACT_VERSION = 1 MATERIAL_CONTRACT_VERSION = 1 DEFAULT_REBUILD_IMAGE = ( - "docker.io/library/postgres:16-alpine@" - "sha256:57c72fd2a128e416c7fcc499958864df5301e940bca0a56f58fddf30ffc07777" -) -SUPPORTED_PROPOSAL_TYPES = frozenset( - {"add_edge", "attach_evidence", "approve_claim", "revise_strategy"} + "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, approve_claim, and revise_strategy; mutating strategy replay " @@ -346,9 +343,7 @@ def _validate_proposal_rows( ) -def _parse_aware_timestamp( - value: Any, field: str, *, code: str = "invalid_mutating_receipt" -) -> datetime: +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: @@ -360,9 +355,7 @@ def _parse_aware_timestamp( return parsed -def _validated_uuid_list( - value: Any, field: str, *, code: str = "invalid_mutating_state" -) -> list[str]: +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] = [] @@ -382,9 +375,7 @@ def _validate_revise_strategy_receipt_rows( 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" - ) + 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: @@ -395,9 +386,7 @@ def _validate_revise_strategy_receipt_rows( 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" - ) + 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) @@ -418,9 +407,7 @@ def _validate_revise_strategy_receipt_rows( 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] + 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", @@ -448,7 +435,14 @@ def _validate_revise_strategy_receipt_rows( "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" @@ -767,9 +761,7 @@ def _validate_revise_strategy_prestate(value: Any) -> dict[str, Any]: ) 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" - ) + 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" @@ -824,9 +816,7 @@ def _validate_revise_strategy_generated_delta( ) -> 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" - ) + raise ReconstructionError("invalid_mutating_delta", "revise_strategy generated delta must be a row object") try: replay_receipt.build_replay_receipt( entry.receipt["proposal"], @@ -848,18 +838,12 @@ def _validate_revise_strategy_generated_delta( "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" - ) + 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" - ) + 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" - ) + _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 @@ -868,9 +852,7 @@ def _validate_revise_strategy_generated_delta( "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([strategy.get("id")], "generated strategy id", code="invalid_mutating_delta") _validated_uuid_list( [row.get("id") for row in nodes], "generated strategy node ids", @@ -885,9 +867,7 @@ def build_revise_strategy_normalization_sql( 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 - ) + 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"] ) @@ -896,9 +876,9 @@ def build_revise_strategy_normalization_sql( "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"]): + 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" ) @@ -1219,9 +1199,7 @@ def apply_entry( 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 - ) + normalization_sql = build_revise_strategy_normalization_sql(entry, mutating_prestate, generated_rows) summary["mutating_delta_validated"] = True _psql( args, diff --git a/tests/test_run_local_genesis_ledger_rebuild.py b/tests/test_run_local_genesis_ledger_rebuild.py index 721772c..1eb8ca0 100644 --- a/tests/test_run_local_genesis_ledger_rebuild.py +++ b/tests/test_run_local_genesis_ledger_rebuild.py @@ -266,9 +266,7 @@ def replay_material() -> dict: } -def revise_strategy_replay_material( - *, apply_sql_source: str = "exact_executed_sql", version: int = 2 -) -> 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" @@ -322,6 +320,19 @@ def revise_strategy_replay_material( } +def rehash_revise_strategy_receipt(material: dict) -> None: + prior_receipt = material["replay_receipt"] + proposal = applied_envelope(material["applied_proposal"]) + apply_sql = rebuild.apply_engine.build_apply_sql(proposal, material["applied_proposal"]["applied_by_handle"]) + material["replay_receipt"] = rebuild.replay_receipt.build_replay_receipt( + proposal, + prior_receipt["canonical_rows"], + apply_sql_sha256=rebuild.replay_receipt.sha256_text(apply_sql), + apply_sql_source=prior_receipt["apply_engine"]["source"], + exported_at_utc=prior_receipt["exported_at_utc"], + ) + + def write_bundle(tmp_path: Path, *, material: dict | None = None) -> argparse.Namespace: dump = tmp_path / "genesis.dump" dump.write_bytes(b"PGDMPfixture") @@ -451,13 +462,94 @@ def test_revise_strategy_material_rejects_reconstructed_apply_engine(tmp_path: P assert exc_info.value.code == "mutating_apply_engine_mismatch" +def test_revise_strategy_rejects_fully_rehashed_transaction_before_approval( + tmp_path: Path, +) -> None: + material = revise_strategy_replay_material() + original_receipt = material["replay_receipt"] + forged_timestamp = "2026-07-13T01:00:30+00:00" + canonical_rows = copy.deepcopy(original_receipt["canonical_rows"]) + canonical_rows["strategies"][0]["created_at"] = forged_timestamp + for row in canonical_rows["strategy_nodes"]: + row["created_at"] = forged_timestamp + row["updated_at"] = forged_timestamp + material["replay_receipt"]["canonical_rows"] = canonical_rows + rehash_revise_strategy_receipt(material) + args = write_bundle(tmp_path, material=material) + ledger = json.loads(args.ledger.read_text(encoding="utf-8")) + material_path = args.ledger.parent / ledger["entries"][0]["material"] + + assert material["replay_receipt"]["hashes"] != original_receipt["hashes"] + assert ledger["entries"][0]["sha256"] == sha256_file(material_path) + assert ( + ledger["entries"][0]["replay_material_sha256"] == material["replay_receipt"]["hashes"]["replay_material_sha256"] + ) + assert args.ledger_sha256 == sha256_file(args.ledger) + with pytest.raises(rebuild.ReconstructionError) as exc_info: + rebuild.load_bundle(args) + + assert exc_info.value.code == "invalid_mutating_receipt" + assert "before proposal approval" in exc_info.value.safe_message + + +@pytest.mark.parametrize( + ("reviewed_at", "transaction_timestamp", "applied_at"), + ( + ( + "2026-07-14T03:00:00+02:00", + "2026-07-14T01:00:00+00:00", + "2026-07-14T01:01:00+00:00", + ), + ( + "2026-07-14T01:00:00+00:00", + "2026-07-14T01:01:00+00:00", + "2026-07-14T03:01:00+02:00", + ), + ), +) +def test_revise_strategy_accepts_timezone_aware_closed_timestamp_boundaries( + tmp_path: Path, + reviewed_at: str, + transaction_timestamp: str, + applied_at: str, +) -> None: + material = revise_strategy_replay_material() + for proposal_row in (material["approved_proposal"], material["applied_proposal"]): + proposal_row["reviewed_at"] = reviewed_at + material["approval_snapshot"]["reviewed_at"] = reviewed_at + material["applied_proposal"]["applied_at"] = applied_at + rows = material["replay_receipt"]["canonical_rows"] + rows["strategies"][0]["created_at"] = transaction_timestamp + for row in rows["strategy_nodes"]: + row["created_at"] = transaction_timestamp + row["updated_at"] = transaction_timestamp + rehash_revise_strategy_receipt(material) + + entry = rebuild.load_bundle(write_bundle(tmp_path, material=material)).entries[0] + + assert entry.receipt["canonical_rows"]["strategies"][0]["created_at"] == transaction_timestamp + + +def test_revise_strategy_rejects_reviewed_at_without_timezone(tmp_path: Path) -> None: + material = revise_strategy_replay_material() + reviewed_at = "2026-07-14T01:00:00" + for proposal_row in (material["approved_proposal"], material["applied_proposal"]): + proposal_row["reviewed_at"] = reviewed_at + material["approval_snapshot"]["reviewed_at"] = reviewed_at + rehash_revise_strategy_receipt(material) + + with pytest.raises(rebuild.ReconstructionError) as exc_info: + rebuild.load_bundle(write_bundle(tmp_path, material=material)) + + assert exc_info.value.code == "invalid_mutating_receipt" + assert "reviewed_at must include a timezone" in exc_info.value.safe_message + + @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: +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": @@ -480,9 +572,7 @@ def test_revise_strategy_receipt_rejects_impossible_poststate( 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] + 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], @@ -508,9 +598,7 @@ def test_revise_strategy_transition_builders_preserve_historical_prestate(tmp_pa 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] + entry = rebuild.load_bundle(write_bundle(tmp_path, material=revise_strategy_replay_material(version=1))).entries[0] prestate = { "strategy_ids": [], "active_strategy_ids": [], @@ -1085,9 +1173,9 @@ def run_genesis_ledger_command( "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"], + "replay_material_sha256": json.loads(material_path.read_text(encoding="utf-8"))["replay_receipt"][ + "hashes" + ]["replay_material_sha256"], } ], "final_parity": { @@ -1326,9 +1414,8 @@ def test_live_revise_strategy_rebuild_is_exact_repeatable_and_leaves_no_containe 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" - ] + 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"] + )