diff --git a/scripts/apply_proposal.py b/scripts/apply_proposal.py index 88c3c65..f357591 100644 --- a/scripts/apply_proposal.py +++ b/scripts/apply_proposal.py @@ -480,6 +480,9 @@ def build_approve_claim_sql( proposal_id: str, applied_by: str | None, approval: dict[str, Any], + *, + fresh_owned: bool = False, + fresh_preflight_sha256: str | None = None, ) -> str: """Build one conflict-safe transaction for a reviewed canonical graph bundle.""" _validate_approval_meta(approval, "approve_claim", apply_payload) @@ -672,9 +675,30 @@ def build_approve_claim_sql( ) _dedupe([row["id"] for row in reasoning_tools], "approve_claim.reasoning_tools ids") + if fresh_owned: + if ( + not isinstance(fresh_preflight_sha256, str) + or len(fresh_preflight_sha256) != 64 + or any(character not in "0123456789abcdef" for character in fresh_preflight_sha256) + ): + raise ValueError("fresh-owned approve_claim requires an exact preflight SHA-256") + elif fresh_preflight_sha256 is not None: + raise ValueError("fresh preflight SHA-256 is only valid for fresh-owned approve_claim") + statements: list[str] = [] checks: list[str] = [] + if fresh_owned: + statements.extend( + [ + f"-- fresh-owned preflight SHA-256: {fresh_preflight_sha256}", + ( + "select pg_advisory_xact_lock(hashtextextended(" + f"'proposal-fresh-apply:' || {sql_literal(proposal_id)}, 0));" + ), + ] + ) + if sources: collision_checks = [] for row in sources: @@ -689,12 +713,12 @@ def build_approve_claim_sql( for row in claims: tags = _text_array(row["tags"]) + conflict_clause = ";" if fresh_owned else "\non conflict (id) do nothing;" statements.append( f"""insert into public.claims (id, type, text, status, confidence, tags, created_by, superseded_by) select {sql_literal(row["id"])}::uuid, {sql_literal(row["type"])}, {sql_literal(row["text"])}, {sql_literal(row["status"])}, {sql_literal(row["confidence"])}, {tags}, - {sql_literal(row["created_by"])}::uuid, {sql_literal(row["superseded_by"])}::uuid -on conflict (id) do nothing;""" + {sql_literal(row["created_by"])}::uuid, {sql_literal(row["superseded_by"])}::uuid{conflict_clause}""" ) checks.append( f""" if not exists (select 1 from public.claims @@ -711,12 +735,12 @@ on conflict (id) do nothing;""" ) for row in sources: + conflict_clause = ";" if fresh_owned else "\non conflict do nothing;" statements.append( f"""insert into public.sources (id, source_type, url, storage_path, excerpt, hash, created_by) select {sql_literal(row["id"])}::uuid, {sql_literal(row["source_type"])}, {sql_literal(row["url"])}, {sql_literal(row["storage_path"])}, {sql_literal(row["excerpt"])}, {sql_literal(row["hash"])}, - {sql_literal(row["created_by"])}::uuid -on conflict do nothing;""" + {sql_literal(row["created_by"])}::uuid{conflict_clause}""" ) checks.append( f""" if exists (select 1 from public.sources @@ -737,11 +761,11 @@ on conflict do nothing;""" ) for row in reasoning_tools: + conflict_clause = ";" if fresh_owned else "\non conflict (id) do nothing;" statements.append( f"""insert into public.reasoning_tools (id, agent_id, name, description, category) select {sql_literal(row["id"])}::uuid, {sql_literal(row["agent_id"])}::uuid, - {sql_literal(row["name"])}, {sql_literal(row["description"])}, {sql_literal(row["category"])} -on conflict (id) do nothing;""" + {sql_literal(row["name"])}, {sql_literal(row["description"])}, {sql_literal(row["category"])}{conflict_clause}""" ) checks.append( f""" if not exists (select 1 from public.reasoning_tools @@ -755,12 +779,12 @@ on conflict (id) do nothing;""" ) for row in evidence: + conflict_clause = ";" if fresh_owned else "\non conflict (claim_id, source_id, role) do nothing;" statements.append( f"""insert into public.claim_evidence (id, claim_id, source_id, role, weight, created_by) select {sql_literal(row["id"])}::uuid, {sql_literal(row["claim_id"])}::uuid, {sql_literal(row["source_id"])}::uuid, {sql_literal(row["role"])}::evidence_role, {sql_literal(row["weight"])}, - {sql_literal(row["created_by"])}::uuid -on conflict (claim_id, source_id, role) do nothing;""" + {sql_literal(row["created_by"])}::uuid{conflict_clause}""" ) checks.append( f""" -- Accept a legacy receipt id only when exactly one semantic row matches. @@ -790,17 +814,21 @@ on conflict (claim_id, source_id, role) do nothing;""" for row in edges: edge_lock_key = f"claim_edge:{row['from_claim']}:{row['to_claim']}:{row['edge_type']}" + edge_insert_guard = ( + ";" + if fresh_owned + else f"""\n where not exists ( + select 1 from public.claim_edges + where from_claim = {sql_literal(row["from_claim"])}::uuid + and to_claim = {sql_literal(row["to_claim"])}::uuid + and edge_type = {sql_literal(row["edge_type"])}::edge_type);""" + ) statements.append( f"""select pg_advisory_xact_lock(hashtextextended({sql_literal(edge_lock_key)}, 0)); insert into public.claim_edges (id, from_claim, to_claim, edge_type, weight, created_by) select {sql_literal(row["id"])}::uuid, {sql_literal(row["from_claim"])}::uuid, {sql_literal(row["to_claim"])}::uuid, {sql_literal(row["edge_type"])}::edge_type, {sql_literal(row["weight"])}, - {sql_literal(row["created_by"])}::uuid - where not exists ( - select 1 from public.claim_edges - where from_claim = {sql_literal(row["from_claim"])}::uuid - and to_claim = {sql_literal(row["to_claim"])}::uuid - and edge_type = {sql_literal(row["edge_type"])}::edge_type);""" + {sql_literal(row["created_by"])}::uuid{edge_insert_guard}""" ) checks.append( f""" -- Accept a legacy receipt id only when exactly one semantic row matches. @@ -830,17 +858,29 @@ select {sql_literal(row["id"])}::uuid, {sql_literal(row["from_claim"])}::uuid, { canonical = "\n\n".join(statements) ledger = _ledger_and_verify(proposal_id, applied_by or SERVICE_AGENT_HANDLE, "\n".join(checks), approval) - return _wrap_txn(canonical, ledger, _approval_guard_sql(proposal_id, approval)) + return _wrap_txn( + canonical, + ledger, + _approval_guard_sql(proposal_id, approval), + serializable=fresh_owned, + ) -def _wrap_txn(canonical_sql: str, ledger_sql: str, approval_guard_sql: str) -> str: +def _wrap_txn( + canonical_sql: str, + ledger_sql: str, + approval_guard_sql: str, + *, + serializable: bool = False, +) -> str: return ( "begin;\n" - "set local standard_conforming_strings = on;\n" - f"{approval_guard_sql}\n" - f"{canonical_sql}\n" - f"{ledger_sql}\n" - "commit;\n" + + ("set transaction isolation level serializable;\n" if serializable else "") + + "set local standard_conforming_strings = on;\n" + + f"{approval_guard_sql}\n" + + f"{canonical_sql}\n" + + f"{ledger_sql}\n" + + "commit;\n" ) @@ -852,7 +892,13 @@ BUILDERS = { } -def build_apply_sql(proposal: dict[str, Any], applied_by: str | None) -> str: +def build_apply_sql( + proposal: dict[str, Any], + applied_by: str | None, + *, + fresh_owned: bool = False, + fresh_preflight_sha256: str | None = None, +) -> str: ptype = proposal["proposal_type"] if ptype not in BUILDERS: raise ValueError(f"apply_proposal cannot apply proposal_type {ptype!r}") @@ -863,6 +909,17 @@ def build_apply_sql(proposal: dict[str, Any], applied_by: str | None) -> str: "proposal payload has no 'apply_payload' strict contract. " "Run the normalizer to produce apply_payload before applying." ) + if fresh_owned and ptype != "approve_claim": + raise ValueError("fresh-owned apply is supported only for approve_claim") + if ptype == "approve_claim": + return build_approve_claim_sql( + apply_payload, + proposal["id"], + applied_by or SERVICE_AGENT_HANDLE, + proposal, + fresh_owned=fresh_owned, + fresh_preflight_sha256=fresh_preflight_sha256, + ) return BUILDERS[ptype]( apply_payload, proposal["id"], @@ -990,7 +1047,7 @@ def capture_replay_receipt( raise RuntimeError("postflight query did not return a JSON object") apply_sql_source = "exact_executed_sql" if apply_sql is None: - apply_sql = build_apply_sql(proposal, proposal.get("applied_by_handle")) + apply_sql = build_apply_sql_for_args(proposal, args, proposal.get("applied_by_handle")) apply_sql_source = "reconstructed_current_engine" receipt = replay_receipt.build_replay_receipt( proposal, @@ -1022,6 +1079,14 @@ def parse_args(argv: list[str]) -> argparse.Namespace: p.add_argument("--db", default=DEFAULT_DB) p.add_argument("--host", default=DEFAULT_HOST) p.add_argument("--role", default=DEFAULT_ROLE) + p.add_argument( + "--fresh-preflight", + type=Path, + help=( + "exact proposal_apply_fresh_preflight JSON; makes approve_claim use strict fresh-owned inserts " + "bound to the timestamped preflight" + ), + ) p.add_argument( "--receipt-only", action="store_true", @@ -1043,6 +1108,28 @@ def parse_args(argv: list[str]) -> argparse.Namespace: return args +def build_apply_sql_for_args( + proposal: dict[str, Any], + args: argparse.Namespace, + applied_by: str | None, +) -> str: + fresh_preflight_path = getattr(args, "fresh_preflight", None) + if fresh_preflight_path is None: + return build_apply_sql(proposal, applied_by) + if not fresh_preflight_path.is_file(): + raise SystemExit(f"fresh preflight file not found: {fresh_preflight_path}") + try: + preflight = json.loads(fresh_preflight_path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise SystemExit(f"fresh preflight is unreadable: {fresh_preflight_path}") from exc + import proposal_apply_lifecycle as lifecycle + + try: + return lifecycle.build_fresh_owned_apply_sql(proposal, preflight, applied_by) + except ValueError as exc: + raise SystemExit(f"fresh preflight refused: {exc}") from exc + + def main(argv: list[str] | None = None) -> int: args = parse_args(sys.argv[1:] if argv is None else argv) @@ -1069,13 +1156,13 @@ def main(argv: list[str] | None = None) -> int: password = load_password(args.secrets_file) proposal = load_proposal(args, password) assert_applyable(proposal) - print(build_apply_sql(proposal, args.applied_by)) + print(build_apply_sql_for_args(proposal, args, args.applied_by)) return 0 password = load_password(args.secrets_file) proposal = load_proposal(args, password) assert_applyable(proposal) - sql = build_apply_sql(proposal, args.applied_by) + sql = build_apply_sql_for_args(proposal, args, args.applied_by) run_psql(args, sql, password) applied_proposal = load_proposal(args, password) try: diff --git a/scripts/proposal_apply_lifecycle.py b/scripts/proposal_apply_lifecycle.py index 9ddb600..03ab6d0 100644 --- a/scripts/proposal_apply_lifecycle.py +++ b/scripts/proposal_apply_lifecycle.py @@ -11,7 +11,9 @@ from __future__ import annotations import argparse import hashlib +import hmac import json +import stat import subprocess import sys from datetime import datetime, timezone @@ -24,11 +26,16 @@ sys.path.insert(0, str(HERE)) import apply_proposal as ap # noqa: E402 import kb_apply_replay_receipt as replay_receipt # noqa: E402 -ROLLBACK_SCHEMA = "livingip.proposalApplyRollback.v1" -LIFECYCLE_SCHEMA = "livingip.proposalApplyLifecycleReceipt.v1" -AUTHORIZATION_SCHEMA = "livingip.proposalApplyAuthorizationPacket.v1" +ROLLBACK_SCHEMA = "livingip.proposalApplyRollback.v2" +LIFECYCLE_SCHEMA = "livingip.proposalApplyLifecycleReceipt.v2" +AUTHORIZATION_SCHEMA = "livingip.proposalApplyAuthorizationPacket.v2" FRESH_PREFLIGHT_SCHEMA = "livingip.proposalApplyFreshPreflight.v1" -PRODUCTION_ROLLBACK_ARTIFACT_SCHEMA = "livingip.proposalApplyProductionRollbackArtifact.v1" +PRODUCTION_ROLLBACK_ARTIFACT_SCHEMA = "livingip.proposalApplyProductionRollbackArtifact.v2" +ATTESTATION_SCHEME = "hmac-sha256-v1" +AUTHORIZATION_ATTESTATION_KEY_SOURCE = "/etc/teleo/secrets/production-apply-authorization-attestation.key" +ACTION_READBACK_ATTESTATION_KEY_SOURCE = "/etc/teleo/secrets/production-apply-action-readback-attestation.key" +DEFAULT_AUTHORIZATION_ATTESTATION_KEY_PATH = Path(AUTHORIZATION_ATTESTATION_KEY_SOURCE) +DEFAULT_ACTION_READBACK_ATTESTATION_KEY_PATH = Path(ACTION_READBACK_ATTESTATION_KEY_SOURCE) RUNTIME_DEPENDENCIES = ( "scripts/proposal_apply_lifecycle.py", @@ -210,6 +217,49 @@ def _require_fresh_preflight(proposal_before: dict[str, Any], preflight: dict[st raise ValueError("fresh preflight contract is internally inconsistent") +def build_fresh_owned_apply_sql( + proposal_before: dict[str, Any], + preflight: dict[str, Any], + applied_by: str | None = None, +) -> str: + """Build strict insert-only SQL bound to the exact timestamped preflight.""" + _require_fresh_preflight(proposal_before, preflight) + return ap.build_apply_sql( + proposal_before, + applied_by or ap.SERVICE_AGENT_HANDLE, + fresh_owned=True, + fresh_preflight_sha256=preflight["preflight_artifact_sha256"], + ) + + +def _load_private_attestation_key(path: Path) -> bytes | None: + """Load an operator-owned key without accepting key material from the record.""" + try: + metadata = path.lstat() + except OSError: + return None + if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISREG(metadata.st_mode): + return None + if stat.S_IMODE(metadata.st_mode) & 0o077: + return None + try: + key = path.read_bytes().strip() + except OSError: + return None + return key if len(key) >= 32 else None + + +def _key_id(key: bytes | None) -> str | None: + return hashlib.sha256(key).hexdigest() if key is not None else None + + +def _verify_hmac(key: bytes | None, payload: dict[str, Any], supplied: Any) -> bool: + if key is None or not is_sha256(supplied): + return False + expected = hmac.new(key, canonical_json(payload).encode("utf-8"), hashlib.sha256).hexdigest() + return hmac.compare_digest(expected, supplied) + + def _require_apply_receipt( proposal_before: dict[str, Any], preflight: dict[str, Any], @@ -292,6 +342,12 @@ def _delete_statement(table: str, rows: list[dict[str, Any]], label: str) -> str end if;""" +def _uuid_array(values: list[str]) -> str: + if not values: + return "array[]::uuid[]" + return "array[" + ", ".join(f"{ap.sql_literal(value)}::uuid" for value in values) + "]::uuid[]" + + def _require_approval_snapshot(proposal_before: dict[str, Any], approval_snapshot: dict[str, Any]) -> None: expected_keys = { "proposal_id", @@ -345,6 +401,10 @@ def build_compensating_rollback_sql( applied_by_agent_id = ap.sql_literal(applied.get("applied_by_agent_id")) applied_at = ap.sql_literal(applied.get("applied_at")) approval_json = ap.sql_literal(canonical_json(approval_snapshot)) + claim_ids = _uuid_array([str(row["id"]) for row in rows["claims"]]) + source_ids = _uuid_array([str(row["id"]) for row in rows["sources"]]) + evidence_ids = _uuid_array([str(row["id"]) for row in rows["claim_evidence"]]) + edge_ids = _uuid_array([str(row["id"]) for row in rows["claim_edges"]]) row_assertions = [] for family in DELETE_ORDER: @@ -362,6 +422,7 @@ do $rollback$ declare affected integer; approval_now jsonb; + downstream_dependency_count integer; begin perform 1 from kb_stage.kb_proposals where id = {proposal_id}::uuid @@ -400,10 +461,30 @@ begin if approval_now is distinct from {approval_json}::jsonb then raise exception 'proposal rollback: immutable approval snapshot drifted'; end if; + select count(*) into downstream_dependency_count + from ( + select ce.id + from public.claim_evidence ce + where (ce.claim_id = any({claim_ids}) or ce.source_id = any({source_ids})) + and ce.id <> all({evidence_ids}) + union all + select edge.id + from public.claim_edges edge + where (edge.from_claim = any({claim_ids}) or edge.to_claim = any({claim_ids})) + and edge.id <> all({edge_ids}) + union all + select claim.id + from public.claims claim + where claim.superseded_by = any({claim_ids}) + and claim.id <> all({claim_ids}) + ) external_dependencies; + if downstream_dependency_count <> 0 then + raise exception 'proposal rollback: % downstream dependenc(ies) exist; rollback quarantined', downstream_dependency_count; + end if; {chr(10).join(row_assertions)} {chr(10).join(deletions)} update kb_stage.kb_proposals - set status = 'approved', + set status = 'canceled', applied_by_handle = null, applied_by_agent_id = null, applied_at = null, @@ -557,6 +638,12 @@ def build_lifecycle_receipt( generated_at_utc: str | None = None, ) -> dict[str, Any]: rows = _require_apply_receipt(proposal_before, preflight, apply_receipt) + expected_fresh_apply_sql = build_fresh_owned_apply_sql( + proposal_before, + preflight, + apply_receipt["proposal"].get("applied_by_handle"), + ) + expected_fresh_apply_sha256 = sha256_text(expected_fresh_apply_sql) expected_rollback_sql = build_compensating_rollback_sql( proposal_before, approval_before, @@ -569,20 +656,23 @@ def build_lifecycle_receipt( expected_after = dict(proposal_before) expected_after.update( { - "status": "approved", + "status": "canceled", "applied_by_handle": None, "applied_by_agent_id": None, "applied_at": None, } ) - ledger_restored = all(proposal_after.get(key) == value for key, value in expected_after.items()) + ledger_quarantined = all(proposal_after.get(key) == value for key, value in expected_after.items()) checks = { "fresh_owned_targets": preflight.get("fresh_owned_targets") is True, "apply_receipt_replay_ready": apply_receipt.get("replay_ready") is True, + "fresh_owned_preflight_consumed_by_apply": ( + apply_receipt["apply_engine"].get("apply_sql_sha256") == expected_fresh_apply_sha256 + ), "apply_rows_have_exact_ids": all(rows[name] == apply_receipt["canonical_rows"][name] for name in ROW_TABLES), "rollback_target_rows_absent": target_rows_absent, "rollback_counts_restored": counts_after == counts_before, - "rollback_ledger_restored_to_approved": ledger_restored, + "rollback_ledger_quarantined_from_worker": ledger_quarantined, "immutable_approval_unchanged": approval_after == approval_before, "unrelated_rows_unchanged": unrelated_after == unrelated_before, "rollback_replay_refused_without_mutation": rollback_replay_refused is True, @@ -618,6 +708,7 @@ def build_lifecycle_receipt( "fresh_preflight_artifact_sha256": preflight["preflight_artifact_sha256"], "apply_receipt_sha256": sha256_json(apply_receipt), "apply_sql_sha256": apply_receipt["apply_engine"]["apply_sql_sha256"], + "fresh_owned_apply_sql_sha256": expected_fresh_apply_sha256, "applied_row_ids": _row_ids(rows), "applied_row_sha256": {name: [sha256_json(row) for row in table_rows] for name, table_rows in rows.items()}, "rollback": { @@ -672,6 +763,7 @@ def _require_lifecycle_receipt(lifecycle_receipt: dict[str, Any]) -> None: "fresh_preflight_artifact_sha256", "apply_receipt_sha256", "apply_sql_sha256", + "fresh_owned_apply_sql_sha256", "applied_row_ids", "applied_row_sha256", "rollback", @@ -718,9 +810,12 @@ def _require_lifecycle_receipt(lifecycle_receipt: dict[str, Any]) -> None: "fresh_preflight_artifact_sha256", "apply_receipt_sha256", "apply_sql_sha256", + "fresh_owned_apply_sql_sha256", ) if any(not is_sha256(lifecycle_receipt.get(field)) for field in hash_fields): raise ValueError("lifecycle receipt is missing an exact SHA-256 binding") + if lifecycle_receipt.get("apply_sql_sha256") != lifecycle_receipt.get("fresh_owned_apply_sql_sha256"): + raise ValueError("lifecycle receipt apply SQL did not consume the fresh-owned preflight") row_ids = lifecycle_receipt.get("applied_row_ids") row_hashes = lifecycle_receipt.get("applied_row_sha256") if ( @@ -747,6 +842,7 @@ def _require_lifecycle_receipt(lifecycle_receipt: dict[str, Any]) -> None: or set(target_rows_after) != set(ROW_TABLES) or any(not isinstance(rows, list) or rows for rows in target_rows_after.values()) or rollback.get("counts_before_apply") != rollback.get("counts_after_rollback") + or (rollback.get("proposal_after") or {}).get("status") != "canceled" ): raise ValueError("lifecycle receipt does not prove exact rollback cleanup") @@ -917,6 +1013,12 @@ def build_authorization_packet( else None ) observed_at_utc = destination_readback.get("observed_at_utc") + downstream_dependency_count = destination_readback.get("downstream_dependency_count") + downstream_dependency_readback_valid = ( + isinstance(downstream_dependency_count, int) + and not isinstance(downstream_dependency_count, bool) + and downstream_dependency_count >= 0 + ) destination_readback_was_read_only = ( destination_readback.get("read_only_transaction") is True and _parse_utc(observed_at_utc) is not None ) @@ -926,6 +1028,7 @@ def build_authorization_packet( "approval_gate_ready": approval_gate_ready, "strict_proposal_present_and_approved": len(matching) == 1, "matching_candidate_count": len(matching), + "zero_downstream_dependencies": (downstream_dependency_readback_valid and downstream_dependency_count == 0), "exact_production_rollback_packet_ready": production_rollback_sha256 is not None, "production_apply_authorization_present": False, "production_apply_executed": False, @@ -949,6 +1052,14 @@ def build_authorization_packet( "destination_readback": { "observed_at_utc": observed_at_utc, "read_only_transaction": destination_readback.get("read_only_transaction"), + "downstream_dependency_count": downstream_dependency_count, + }, + "attestation_policy": { + "scheme": ATTESTATION_SCHEME, + "authorization_key_source": AUTHORIZATION_ATTESTATION_KEY_SOURCE, + "action_readback_key_source": ACTION_READBACK_ATTESTATION_KEY_SOURCE, + "separate_operator_owned_keys_required": True, + "locally_synthesized_records_refused": True, }, "approval_gate": bound_gate, "candidate_readback": { @@ -1010,6 +1121,22 @@ def build_authorization_packet( "received_at_utc": "", "expires_at_utc": "", "authorization_text": "", + "authorization_provenance": { + "scheme": ATTESTATION_SCHEME, + "source": "codex_platform_user_message", + "source_event_id": "", + "source_event_sha256": "", + "key_id": "", + "attestation_hmac_sha256": "", + }, + "action_readback_provenance": { + "scheme": ATTESTATION_SCHEME, + "collector": "teleo_production_readonly_collector", + "receipt_id": "", + "observed_destination_sha256": "", + "key_id": "", + "attestation_hmac_sha256": "", + }, }, "authorization_text_template": _authorization_text_template(binding, binding_sha256), "safe_to_enter_apply_window": False, @@ -1044,6 +1171,8 @@ def build_authorization_packet( if not matching else "generate and retain the exact production rollback SQL" if production_rollback_sha256 is None + else "collect an independently attested zero-dependency action-time readback" + if not production_preconditions["zero_downstream_dependencies"] else "obtain the exact action-time authorization record" ), "claim_ceiling": ( @@ -1080,6 +1209,59 @@ def _parse_utc(value: Any) -> datetime | None: return parsed.astimezone(timezone.utc) +def action_readback_projection(observed_destination: dict[str, Any]) -> dict[str, Any]: + """Canonical action-time DB readback covered by the independent attestation.""" + return { + "observed_at_utc": observed_destination.get("observed_at_utc"), + "read_only_transaction": observed_destination.get("read_only_transaction"), + "container": observed_destination.get("container"), + "database": observed_destination.get("database"), + "system_identifier": observed_destination.get("system_identifier"), + "server_version_num": observed_destination.get("server_version_num"), + "approval_gate": observed_destination.get("approval_gate"), + "approved_strict_candidates": observed_destination.get("approved_strict_candidates"), + "downstream_dependency_count": observed_destination.get("downstream_dependency_count"), + } + + +def authorization_attestation_payload(packet: dict[str, Any], record: dict[str, Any]) -> dict[str, Any]: + provenance = record.get("authorization_provenance") + provenance = provenance if isinstance(provenance, dict) else {} + return { + "scheme": ATTESTATION_SCHEME, + "binding_sha256": packet.get("binding_sha256"), + "authorized": record.get("authorized"), + "rollback_authorized": record.get("rollback_authorized"), + "production_rollback_sql_sha256": record.get("production_rollback_sql_sha256"), + "operator_identity": record.get("operator_identity"), + "received_at_utc": record.get("received_at_utc"), + "expires_at_utc": record.get("expires_at_utc"), + "authorization_text": record.get("authorization_text"), + "source": provenance.get("source"), + "source_event_id": provenance.get("source_event_id"), + "source_event_sha256": provenance.get("source_event_sha256"), + "key_id": provenance.get("key_id"), + } + + +def action_readback_attestation_payload( + packet: dict[str, Any], + observed_destination: dict[str, Any], + record: dict[str, Any], +) -> dict[str, Any]: + provenance = record.get("action_readback_provenance") + provenance = provenance if isinstance(provenance, dict) else {} + return { + "scheme": ATTESTATION_SCHEME, + "binding_sha256": packet.get("binding_sha256"), + "collector": provenance.get("collector"), + "receipt_id": provenance.get("receipt_id"), + "observed_destination_sha256": provenance.get("observed_destination_sha256"), + "key_id": provenance.get("key_id"), + "readback": action_readback_projection(observed_destination), + } + + def verify_authorization_record( packet: dict[str, Any], record: dict[str, Any], @@ -1232,6 +1414,8 @@ def verify_authorization_record( "received_at_utc", "expires_at_utc", "authorization_text", + "authorization_provenance", + "action_readback_provenance", } expected_record_template = { "authorized": True, @@ -1246,6 +1430,22 @@ def verify_authorization_record( "received_at_utc": "", "expires_at_utc": "", "authorization_text": "", + "authorization_provenance": { + "scheme": ATTESTATION_SCHEME, + "source": "codex_platform_user_message", + "source_event_id": "", + "source_event_sha256": "", + "key_id": "", + "attestation_hmac_sha256": "", + }, + "action_readback_provenance": { + "scheme": ATTESTATION_SCHEME, + "collector": "teleo_production_readonly_collector", + "receipt_id": "", + "observed_destination_sha256": "", + "key_id": "", + "attestation_hmac_sha256": "", + }, } try: expected_record_text = expected_authorization_text(packet, record) @@ -1281,6 +1481,75 @@ def verify_authorization_record( and received <= action_observed_at <= now and observed_destination.get("read_only_transaction") is True ) + bound_attestation_policy = ( + binding.get("attestation_policy") if isinstance(binding.get("attestation_policy"), dict) else {} + ) + expected_attestation_policy = { + "scheme": ATTESTATION_SCHEME, + "authorization_key_source": AUTHORIZATION_ATTESTATION_KEY_SOURCE, + "action_readback_key_source": ACTION_READBACK_ATTESTATION_KEY_SOURCE, + "separate_operator_owned_keys_required": True, + "locally_synthesized_records_refused": True, + } + authorization_provenance = ( + record.get("authorization_provenance") if isinstance(record.get("authorization_provenance"), dict) else {} + ) + action_readback_provenance = ( + record.get("action_readback_provenance") if isinstance(record.get("action_readback_provenance"), dict) else {} + ) + authorization_key = _load_private_attestation_key(DEFAULT_AUTHORIZATION_ATTESTATION_KEY_PATH) + action_readback_key = _load_private_attestation_key(DEFAULT_ACTION_READBACK_ATTESTATION_KEY_PATH) + authorization_provenance_contract = { + "scheme", + "source", + "source_event_id", + "source_event_sha256", + "key_id", + "attestation_hmac_sha256", + } + action_readback_provenance_contract = { + "scheme", + "collector", + "receipt_id", + "observed_destination_sha256", + "key_id", + "attestation_hmac_sha256", + } + authorization_provenance_valid = ( + set(authorization_provenance) == authorization_provenance_contract + and authorization_provenance.get("scheme") == ATTESTATION_SCHEME + and authorization_provenance.get("source") == "codex_platform_user_message" + and isinstance(authorization_provenance.get("source_event_id"), str) + and bool(authorization_provenance.get("source_event_id", "").strip()) + and authorization_provenance.get("source_event_sha256") + == sha256_text(str(record.get("authorization_text") or "")) + and authorization_provenance.get("key_id") == _key_id(authorization_key) + and _verify_hmac( + authorization_key, + authorization_attestation_payload(packet, record), + authorization_provenance.get("attestation_hmac_sha256"), + ) + ) + action_projection = action_readback_projection(observed_destination) + action_readback_provenance_valid = ( + set(action_readback_provenance) == action_readback_provenance_contract + and action_readback_provenance.get("scheme") == ATTESTATION_SCHEME + and action_readback_provenance.get("collector") == "teleo_production_readonly_collector" + and isinstance(action_readback_provenance.get("receipt_id"), str) + and bool(action_readback_provenance.get("receipt_id", "").strip()) + and action_readback_provenance.get("observed_destination_sha256") == sha256_json(action_projection) + and action_readback_provenance.get("key_id") == _key_id(action_readback_key) + and authorization_key is not None + and action_readback_key is not None + and not hmac.compare_digest(authorization_key, action_readback_key) + and _verify_hmac( + action_readback_key, + action_readback_attestation_payload(packet, observed_destination, record), + action_readback_provenance.get("attestation_hmac_sha256"), + ) + ) + bound_dependency_count = bound_destination_readback.get("downstream_dependency_count") + action_dependency_count = observed_destination.get("downstream_dependency_count") checks = { "packet_schema_exact": ( packet.get("artifact") == "proposal_apply_production_authorization_packet" @@ -1293,6 +1562,9 @@ def verify_authorization_record( "packet_preconditions_bound_exactly": ( canonical_json(packet_preconditions) == canonical_json(bound_preconditions) ), + "independent_attestation_policy_bound": ( + canonical_json(bound_attestation_policy) == canonical_json(expected_attestation_policy) + ), "proposal_binding_self_consistent": proposal_binding_self_consistent, "packet_destination_readback_was_read_only": ( packet_preconditions.get("destination_readback_was_read_only") is True @@ -1313,6 +1585,11 @@ def verify_authorization_record( and packet_preconditions.get("matching_candidate_count") == 1 and bound_candidate_exact ), + "zero_downstream_dependencies_at_packet": ( + packet_preconditions.get("zero_downstream_dependencies") is True + and bound_dependency_count == 0 + and not isinstance(bound_dependency_count, bool) + ), "packet_never_self_authorizes_or_expands_scope": ( packet.get("safe_to_enter_apply_window") is False and packet.get("production_apply_authorization_present") is False @@ -1342,10 +1619,15 @@ def verify_authorization_record( ), "authorization_text_exact": bool(expected_record_text) and record.get("authorization_text") == expected_record_text, + "authorization_provenance_independently_attested": authorization_provenance_valid, + "action_readback_provenance_independently_attested": action_readback_provenance_valid, "destination_identity_exact": destination_matches, "destination_readback_fresh_and_read_only_at_action": action_readback_fresh, "approval_gate_identity_exact_at_action": gate_matches, "strict_proposal_exact_at_action": len(matching_candidates) == 1, + "zero_downstream_dependencies_at_action": ( + action_dependency_count == 0 and not isinstance(action_dependency_count, bool) + ), "apply_engine_sha256_exact": ( apply_engine_path.is_file() and apply_engine_relative == bound_engine.get("apply_engine_path") @@ -1359,7 +1641,7 @@ def verify_authorization_record( ), } return { - "schema": "livingip.proposalApplyAuthorizationVerification.v1", + "schema": "livingip.proposalApplyAuthorizationVerification.v2", "checks": checks, "safe_to_enter_apply_window": all(checks.values()), "production_apply_authorization_present": all(checks.values()), diff --git a/scripts/run_approve_claim_clone_canary.py b/scripts/run_approve_claim_clone_canary.py index 87c4de2..2ea28e6 100644 --- a/scripts/run_approve_claim_clone_canary.py +++ b/scripts/run_approve_claim_clone_canary.py @@ -1095,6 +1095,7 @@ def _apply( *, dry_run: bool = False, receipt_out: Path | None = None, + fresh_preflight: Path | None = None, ) -> subprocess.CompletedProcess[str]: command = [ sys.executable, @@ -1115,6 +1116,8 @@ def _apply( ] if receipt_out is not None: command.extend(["--receipt-out", str(receipt_out)]) + if fresh_preflight is not None: + command.extend(["--fresh-preflight", str(fresh_preflight)]) if dry_run: command.append("--dry-run") return _run(command, check=False) @@ -1766,11 +1769,69 @@ def run_canary(args: argparse.Namespace) -> dict[str, Any]: result["clone_counts_before_apply"] = _base_counts(args.container, clone_db) with tempfile.TemporaryDirectory(prefix="leo-proposal-apply-receipt-") as receipt_dir: private_receipt_path = Path(receipt_dir) / "apply-receipt.json" + fresh_preflight_path = Path(receipt_dir) / "fresh-preflight.json" + fresh_preflight_path.write_text( + json.dumps(fresh_preflight, indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + fresh_preflight_path.chmod(0o600) + intervening_claim = expected_rows["claims"][0] + intervening_tags = ( + "array[" + ", ".join(ap.sql_literal(tag) for tag in intervening_claim["tags"]) + "]::text[]" + ) + _psql( + args.container, + clone_db, + f"""insert into public.claims + (id, type, text, status, confidence, tags, created_by, superseded_by) +values + ({ap.sql_literal(intervening_claim["id"])}::uuid, + {ap.sql_literal(intervening_claim["type"])}, + {ap.sql_literal(intervening_claim["text"])}, + {ap.sql_literal(intervening_claim["status"])}, + {ap.sql_literal(intervening_claim["confidence"])}, + {intervening_tags}, + {ap.sql_literal(intervening_claim["created_by"])}::uuid, + {ap.sql_literal(intervening_claim["superseded_by"])}::uuid);""", + tuples=False, + ) + intervening_apply = _apply( + args, + fixture["proposal_id"], + clone_db, + fresh_preflight=fresh_preflight_path, + ) + intervening_row_after = _psql( + args.container, + clone_db, + f"select count(*) from public.claims where id = {ap.sql_literal(intervening_claim['id'])}::uuid;", + ) + proposal_after_intervening_apply = _proposal_readback( + args.container, + clone_db, + fixture["proposal_id"], + ) + result["intervening_exact_row_refusal"] = { + "returncode": intervening_apply.returncode, + "strict_insert_refused": intervening_apply.returncode != 0, + "intervening_row_preserved": intervening_row_after.strip() == "1", + "proposal_remained_approved": ( + proposal_after_intervening_apply is not None + and proposal_after_intervening_apply.get("status") == "approved" + ), + } + _psql( + args.container, + clone_db, + f"delete from public.claims where id = {ap.sql_literal(intervening_claim['id'])}::uuid;", + tuples=False, + ) first_apply = _apply( args, fixture["proposal_id"], clone_db, receipt_out=private_receipt_path, + fresh_preflight=fresh_preflight_path, ) result["first_apply"] = _apply_command_readback(first_apply) if first_apply.returncode != 0 or not private_receipt_path.is_file(): @@ -1815,6 +1876,62 @@ def run_canary(args: argparse.Namespace) -> dict[str, Any]: "rollback_sql_sha256": lifecycle.sha256_text(rollback_sql), "execution_authority": "disposable_clone_admin_only", } + downstream_claim_id = str( + uuid.uuid5(uuid.NAMESPACE_URL, f"working-leo:rollback-dependency:{fixture['proposal_id']}") + ) + downstream_target_claim_id = str(payload["claims"][0]["id"]) + _psql( + args.container, + clone_db, + f"""insert into public.claims + (id, type, text, status, confidence, tags, created_by, superseded_by) +values + ({ap.sql_literal(downstream_claim_id)}::uuid, 'structural', + 'External dependency that must survive refused proposal rollback.', 'open', null, + array['working-leo','rollback-dependency']::text[], null, + {ap.sql_literal(downstream_target_claim_id)}::uuid);""", + tuples=False, + ) + dependency_rollback = _run( + [ + "docker", + "exec", + "-i", + args.container, + "psql", + "-U", + "postgres", + "-d", + clone_db, + "-v", + "ON_ERROR_STOP=1", + ], + input_text=rollback_sql, + check=False, + ) + dependency_after_refusal = _psql( + args.container, + clone_db, + f"select count(*) from public.claims where id = {ap.sql_literal(downstream_claim_id)}::uuid;", + ) + targets_after_dependency_refusal = _exact_bundle_readback(args.container, clone_db, payload) + proposal_after_dependency_refusal = _proposal_readback(args.container, clone_db, fixture["proposal_id"]) + result["downstream_dependency_rollback_refusal"] = { + "returncode": dependency_rollback.returncode, + "dependency_refusal": "downstream dependenc" in dependency_rollback.stderr.lower(), + "dependency_row_preserved": dependency_after_refusal.strip() == "1", + "proposal_remained_applied": ( + proposal_after_dependency_refusal is not None + and proposal_after_dependency_refusal.get("status") == "applied" + ), + "owned_rows_preserved": targets_after_dependency_refusal == result["row_readback"], + } + _psql( + args.container, + clone_db, + f"delete from public.claims where id = {ap.sql_literal(downstream_claim_id)}::uuid;", + tuples=False, + ) _psql(args.container, clone_db, rollback_sql, tuples=False) rollback_proposal = _proposal_readback(args.container, clone_db, fixture["proposal_id"]) rollback_approval = _approval_snapshot_readback(args.container, clone_db, fixture["proposal_id"]) @@ -1859,6 +1976,12 @@ def run_canary(args: argparse.Namespace) -> dict[str, Any]: "state_unchanged": rollback_replay_state_unchanged, "stderr": rollback_replay.stderr.strip(), } + post_rollback_apply = _apply(args, fixture["proposal_id"], clone_db) + result["post_rollback_worker_apply"] = { + "returncode": post_rollback_apply.returncode, + "terminal_status_refusal": "only 'approved' proposals apply" in post_rollback_apply.stderr, + "proposal_status": (rollback_proposal or {}).get("status"), + } lifecycle_receipt = lifecycle.build_lifecycle_receipt( proposal_before=approved_proposal, approval_before=approved_snapshot, @@ -2047,6 +2170,14 @@ drop database if exists {clone_db}; result.get("fresh_owned_preflight", {}).get("fresh_owned_targets") is True and result.get("target_rows_before_apply") == _empty_bundle_readback(payload) ), + "intervening_exact_row_refused_without_adoption": all( + result.get("intervening_exact_row_refusal", {}).get(key) is True + for key in ( + "strict_insert_refused", + "intervening_row_preserved", + "proposal_remained_approved", + ) + ), "compensating_rollback_receipt_passes": ( result.get("applied_rollback_receipt", {}).get("pass") is True and result.get("applied_rollback_receipt", {}).get("current_tier") == "T2_runtime" @@ -2055,6 +2186,20 @@ drop database if exists {clone_db}; result.get("rollback_replay", {}).get("refused") is True and result.get("rollback_replay", {}).get("state_unchanged") is True ), + "rollback_refuses_downstream_dependencies_without_deletion": all( + result.get("downstream_dependency_rollback_refusal", {}).get(key) is True + for key in ( + "dependency_refusal", + "dependency_row_preserved", + "proposal_remained_applied", + "owned_rows_preserved", + ) + ), + "post_rollback_worker_reapply_refused": ( + result.get("post_rollback_worker_apply", {}).get("returncode", 0) != 0 + and result.get("post_rollback_worker_apply", {}).get("terminal_status_refusal") is True + and result.get("post_rollback_worker_apply", {}).get("proposal_status") == "canceled" + ), "payload_projection_exact": result.get("row_readback") == expected_rows, "clone_apply_table_deltas_exact": (result.get("clone_apply_table_deltas") == expected_table_deltas), "apply_transition_exact": ( diff --git a/tests/test_apply_proposal.py b/tests/test_apply_proposal.py index 8f79873..96bf7e1 100644 --- a/tests/test_apply_proposal.py +++ b/tests/test_apply_proposal.py @@ -378,6 +378,34 @@ def test_apply_sql_locks_and_binds_approval_before_canonical_write(): assert "Reviewed exact strict payload." in sql +def test_fresh_owned_apply_uses_strict_inserts_and_binds_preflight_hash(): + payload = _approve_claim_bundle() + proposal = { + "id": "99999999-9999-4999-8999-999999999999", + "proposal_type": "approve_claim", + "status": "approved", + "payload": {"apply_payload": payload}, + "reviewed_by_handle": "m3ta", + "reviewed_by_agent_id": None, + "reviewed_at": "2026-07-10T00:00:00+00:00", + "review_note": "Reviewed exact strict payload.", + } + preflight_sha256 = "a" * 64 + + sql = ap.build_apply_sql( + proposal, + "kb-apply", + fresh_owned=True, + fresh_preflight_sha256=preflight_sha256, + ) + + assert preflight_sha256 in sql + assert "set transaction isolation level serializable" in sql.lower() + assert "on conflict" not in sql.lower() + assert "where not exists" not in sql.lower() + assert sql.index("set transaction isolation level serializable") < sql.index("kb_stage.assert_approved_proposal") + + def test_build_apply_sql_requires_review_provenance(): proposal = { "id": "99999999-9999-4999-8999-999999999999", diff --git a/tests/test_proposal_apply_lifecycle.py b/tests/test_proposal_apply_lifecycle.py index 916f32f..d05fd00 100644 --- a/tests/test_proposal_apply_lifecycle.py +++ b/tests/test_proposal_apply_lifecycle.py @@ -4,6 +4,7 @@ from __future__ import annotations import copy import hashlib +import hmac import json import string import sys @@ -31,6 +32,18 @@ REVIEWED_AT = "2026-07-15T09:30:00+00:00" ROW_FAMILIES = tuple(lifecycle.ROW_TABLES) +@pytest.fixture(autouse=True) +def _trusted_attestation_keys(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + authorization_key = tmp_path / "authorization-attestation.key" + action_readback_key = tmp_path / "action-readback-attestation.key" + authorization_key.write_bytes(b"authorization-attestation-test-key-0001") + action_readback_key.write_bytes(b"action-readback-attestation-test-key-0002") + authorization_key.chmod(0o600) + action_readback_key.chmod(0o600) + monkeypatch.setattr(lifecycle, "DEFAULT_AUTHORIZATION_ATTESTATION_KEY_PATH", authorization_key) + monkeypatch.setattr(lifecycle, "DEFAULT_ACTION_READBACK_ATTESTATION_KEY_PATH", action_readback_key) + + def _apply_payload() -> dict: return { "contract_version": 2, @@ -207,7 +220,7 @@ def _empty_targets() -> dict[str, list[dict]]: return {family: [] for family in ROW_FAMILIES} -def _apply_receipt(proposal: dict, rows: dict[str, list[dict]]) -> dict: +def _apply_receipt(proposal: dict, rows: dict[str, list[dict]], preflight: dict) -> dict: applied = copy.deepcopy(proposal) applied.update( { @@ -217,7 +230,7 @@ def _apply_receipt(proposal: dict, rows: dict[str, list[dict]]) -> dict: "applied_at": APPLIED_AT, } ) - apply_sql = apply.build_apply_sql(proposal, apply.SERVICE_AGENT_HANDLE) + apply_sql = lifecycle.build_fresh_owned_apply_sql(proposal, preflight, apply.SERVICE_AGENT_HANDLE) return apply.replay_receipt.build_replay_receipt( applied, copy.deepcopy(rows), @@ -235,7 +248,7 @@ def _materials() -> tuple[dict, dict, dict, dict, str]: _empty_targets(), captured_at_utc="2026-07-15T10:00:00+00:00", ) - receipt = _apply_receipt(proposal, _canonical_rows()) + receipt = _apply_receipt(proposal, _canonical_rows(), preflight) rollback_sql = lifecycle.build_compensating_rollback_sql(proposal, approval, preflight, receipt) return proposal, approval, preflight, receipt, rollback_sql @@ -243,6 +256,14 @@ def _materials() -> tuple[dict, dict, dict, dict, str]: def _passing_lifecycle_receipt() -> dict: proposal, approval, preflight, receipt, rollback_sql = _materials() proposal_after = copy.deepcopy(proposal) + proposal_after.update( + { + "status": "canceled", + "applied_by_handle": None, + "applied_by_agent_id": None, + "applied_at": None, + } + ) return lifecycle.build_lifecycle_receipt( proposal_before=proposal, approval_before=approval, @@ -297,6 +318,7 @@ def _destination(*, candidate_present: bool = True) -> dict: "apply_role_present": True, }, "approved_strict_candidates": candidates, + "downstream_dependency_count": 0, } @@ -332,8 +354,38 @@ def _authorization_materials() -> tuple[dict, dict, dict, dict]: "received_at_utc": "2026-07-15T11:00:00+00:00", "expires_at_utc": "2026-07-15T13:00:00+00:00", } - record["authorization_text"] = lifecycle.expected_authorization_text(packet, record) destination["observed_at_utc"] = "2026-07-15T11:30:00+00:00" + record["authorization_text"] = lifecycle.expected_authorization_text(packet, record) + authorization_key = lifecycle.DEFAULT_AUTHORIZATION_ATTESTATION_KEY_PATH.read_bytes().strip() + action_readback_key = lifecycle.DEFAULT_ACTION_READBACK_ATTESTATION_KEY_PATH.read_bytes().strip() + record["authorization_provenance"] = { + "scheme": lifecycle.ATTESTATION_SCHEME, + "source": "codex_platform_user_message", + "source_event_id": "codex-user-event-019f-test", + "source_event_sha256": lifecycle.sha256_text(record["authorization_text"]), + "key_id": hashlib.sha256(authorization_key).hexdigest(), + "attestation_hmac_sha256": "", + } + record["authorization_provenance"]["attestation_hmac_sha256"] = hmac.new( + authorization_key, + lifecycle.canonical_json(lifecycle.authorization_attestation_payload(packet, record)).encode("utf-8"), + hashlib.sha256, + ).hexdigest() + record["action_readback_provenance"] = { + "scheme": lifecycle.ATTESTATION_SCHEME, + "collector": "teleo_production_readonly_collector", + "receipt_id": "production-readback-019f-test", + "observed_destination_sha256": lifecycle.sha256_json(lifecycle.action_readback_projection(destination)), + "key_id": hashlib.sha256(action_readback_key).hexdigest(), + "attestation_hmac_sha256": "", + } + record["action_readback_provenance"]["attestation_hmac_sha256"] = hmac.new( + action_readback_key, + lifecycle.canonical_json(lifecycle.action_readback_attestation_payload(packet, destination, record)).encode( + "utf-8" + ), + hashlib.sha256, + ).hexdigest() return packet, record, destination, production_rollback_artifact @@ -366,6 +418,25 @@ def test_fresh_preflight_requires_every_owned_target_to_be_absent_and_is_determi lifecycle.build_fresh_preflight(proposal, incomplete) +def test_fresh_owned_apply_consumes_timestamped_preflight_and_never_adopts_conflicts(): + proposal, _approval, preflight, _receipt, _rollback_sql = _materials() + + strict_sql = lifecycle.build_fresh_owned_apply_sql(proposal, preflight, apply.SERVICE_AGENT_HANDLE) + ordinary_sql = apply.build_apply_sql(proposal, apply.SERVICE_AGENT_HANDLE) + + assert preflight["preflight_artifact_sha256"] in strict_sql + assert "set transaction isolation level serializable" in strict_sql.lower() + assert "proposal-fresh-apply:" in strict_sql + assert "on conflict" not in strict_sql.lower() + assert "where not exists" not in strict_sql.lower() + assert "on conflict" in ordinary_sql.lower() + + stale = copy.deepcopy(preflight) + stale["captured_at_utc"] = "1970-01-01T00:00:00+00:00" + with pytest.raises(ValueError, match="artifact hash"): + lifecycle.build_fresh_owned_apply_sql(proposal, stale, apply.SERVICE_AGENT_HANDLE) + + @pytest.mark.parametrize( "mutation", ( @@ -428,7 +499,9 @@ def test_compensating_rollback_is_atomic_deterministic_and_refuses_state_drift() rollback_sql.index(f"delete from {lifecycle.ROW_TABLES[family]}") for family in lifecycle.DELETE_ORDER ] assert delete_positions == sorted(delete_positions) - assert "set status = 'approved'" in rollback_sql + assert "set status = 'canceled'" in rollback_sql + assert "downstream_dependency_count" in rollback_sql + assert "downstream dependenc(ies) exist" in rollback_sql assert "immutable approval snapshot drifted" in rollback_sql drifted_receipt = copy.deepcopy(receipt) @@ -478,7 +551,7 @@ def test_rollback_sql_hash_binds_every_receipt_observed_field(): proposal, approval, preflight, receipt, rollback_sql = _materials() changed_rows = copy.deepcopy(receipt["canonical_rows"]) changed_rows["claims"][0]["created_at"] = "2026-07-15T10:15:00.123456+00:00" - changed_receipt = _apply_receipt(proposal, changed_rows) + changed_receipt = _apply_receipt(proposal, changed_rows, preflight) changed_sql = lifecycle.build_compensating_rollback_sql( proposal, @@ -497,7 +570,7 @@ def test_fresh_apply_receipt_requires_deterministic_generated_row_ids(family: st proposal, approval, preflight, receipt, _rollback_sql = _materials() changed_rows = copy.deepcopy(receipt["canonical_rows"]) changed_rows[family][0]["id"] = "eeeeeeee-eeee-4eee-8eee-eeeeeeeeeeee" - changed_receipt = _apply_receipt(proposal, changed_rows) + changed_receipt = _apply_receipt(proposal, changed_rows, preflight) with pytest.raises(ValueError, match="is not deterministic"): lifecycle.build_compensating_rollback_sql( @@ -537,7 +610,7 @@ def test_production_rollback_artifact_reconstructs_exact_sql_and_rejects_noop_su noop = copy.deepcopy(artifact) noop["rollback_sql"] = ( "begin; -- proposal rollback drift: -- immutable approval snapshot drifted " - "-- set status = 'approved'\ncommit;\n" + "-- set status = 'canceled'\ncommit;\n" ) noop["rollback_sql_sha256"] = lifecycle.sha256_text(noop["rollback_sql"]) unsigned = dict(noop) @@ -649,6 +722,47 @@ def test_authorization_packet_is_exact_but_never_self_authorizes_production(): assert all(verification["checks"].values()) +def test_locally_synthesized_authorization_record_and_readback_cannot_self_authorize(): + packet, record, destination, production_rollback_artifact = _authorization_materials() + forged = copy.deepcopy(record) + forged["operator_identity"] = "forged-by-local-caller" + forged["authorization_text"] = lifecycle.expected_authorization_text(packet, forged) + forged["authorization_provenance"]["source_event_sha256"] = lifecycle.sha256_text(forged["authorization_text"]) + forged["authorization_provenance"]["attestation_hmac_sha256"] = "0" * 64 + forged["action_readback_provenance"]["attestation_hmac_sha256"] = "0" * 64 + + verification = lifecycle.verify_authorization_record( + packet, + forged, + observed_destination=copy.deepcopy(destination), + apply_engine_path=REPO_ROOT / "scripts" / "apply_proposal.py", + production_rollback_artifact=production_rollback_artifact, + now=datetime(2026, 7, 15, 12, 0, tzinfo=timezone.utc), + ) + + assert verification["checks"]["authorization_provenance_independently_attested"] is False + assert verification["checks"]["action_readback_provenance_independently_attested"] is False + assert verification["safe_to_enter_apply_window"] is False + + +def test_action_time_downstream_dependency_is_bound_and_fails_closed(): + packet, record, destination, production_rollback_artifact = _authorization_materials() + destination["downstream_dependency_count"] = 1 + + verification = lifecycle.verify_authorization_record( + packet, + record, + observed_destination=destination, + apply_engine_path=REPO_ROOT / "scripts" / "apply_proposal.py", + production_rollback_artifact=production_rollback_artifact, + now=datetime(2026, 7, 15, 12, 0, tzinfo=timezone.utc), + ) + + assert verification["checks"]["zero_downstream_dependencies_at_action"] is False + assert verification["checks"]["action_readback_provenance_independently_attested"] is False + assert verification["safe_to_enter_apply_window"] is False + + def test_authorization_packet_names_missing_live_gate_before_requesting_auth(tmp_path): receipt = _passing_lifecycle_receipt() destination = _destination(candidate_present=False)