"""Deterministic and property-based tests for proposal apply rollback evidence.""" from __future__ import annotations import copy import hashlib import hmac import json import string import sys from datetime import datetime, timezone from pathlib import Path import pytest from hypothesis import given, settings from hypothesis import strategies as st REPO_ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(REPO_ROOT / "scripts")) import apply_proposal as apply # noqa: E402 import proposal_apply_lifecycle as lifecycle # noqa: E402 PROPOSAL_ID = "99999999-9999-4999-8999-999999999999" CLAIM_A = "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa" CLAIM_B = "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb" SOURCE_ID = "cccccccc-cccc-4ccc-8ccc-cccccccccccc" TOOL_ID = "dddddddd-dddd-4ddd-8ddd-dddddddddddd" AGENT_ID = "11111111-1111-4111-8111-111111111111" APPLIED_AT = "2026-07-15T10:15:00+00:00" 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, "agent_id": AGENT_ID, "claims": [ { "id": CLAIM_A, "type": "structural", "text": "Canonical claim A", "status": "open", "confidence": 0.8, "tags": ["working-leo", "lifecycle"], "created_by": AGENT_ID, }, { "id": CLAIM_B, "type": "concept", "text": "Canonical claim B", "status": "open", "confidence": None, "tags": [], "created_by": AGENT_ID, }, ], "sources": [ { "id": SOURCE_ID, "source_type": "paper", "url": "https://example.test/lifecycle", "storage_path": None, "excerpt": "Bound lifecycle evidence.", "hash": "sha256:lifecycle-source", "created_by": AGENT_ID, } ], "evidence": [ { "claim_id": CLAIM_A, "source_id": SOURCE_ID, "role": "grounds", "weight": 0.9, "created_by": AGENT_ID, } ], "edges": [ { "from_claim": CLAIM_A, "to_claim": CLAIM_B, "edge_type": "supports", "weight": 0.7, "created_by": AGENT_ID, } ], "reasoning_tools": [ { "id": TOOL_ID, "agent_id": AGENT_ID, "name": "Lifecycle verifier", "description": "Checks the apply and compensating rollback boundary.", "category": "verification", } ], } def _proposal_before() -> dict: return { "id": PROPOSAL_ID, "proposal_type": "approve_claim", "status": "approved", "payload": {"apply_payload": _apply_payload()}, "reviewed_by_handle": "independent-reviewer", "reviewed_by_agent_id": None, "reviewed_at": REVIEWED_AT, "review_note": "Approved this exact strict payload.", "applied_by_handle": None, "applied_by_agent_id": None, "applied_at": None, } def _approval_snapshot(proposal: dict) -> dict: return { "proposal_id": proposal["id"], "proposal_type": proposal["proposal_type"], "payload": copy.deepcopy(proposal["payload"]), "reviewed_by_handle": proposal["reviewed_by_handle"], "reviewed_by_agent_id": proposal["reviewed_by_agent_id"], "reviewed_by_db_role": "kb_reviewer", "reviewed_at": proposal["reviewed_at"], "review_note": proposal["review_note"], } def _canonical_rows() -> dict[str, list[dict]]: evidence_id = apply.deterministic_row_uuid(PROPOSAL_ID, "claim_evidence", CLAIM_A, SOURCE_ID, "grounds") edge_id = apply.deterministic_row_uuid(PROPOSAL_ID, "claim_edge", CLAIM_A, CLAIM_B, "supports") return { "claims": [ { "id": CLAIM_A, "type": "structural", "text": "Canonical claim A", "status": "open", "confidence": 0.8, "tags": ["working-leo", "lifecycle"], "created_by": AGENT_ID, "superseded_by": None, "created_at": APPLIED_AT, "updated_at": APPLIED_AT, }, { "id": CLAIM_B, "type": "concept", "text": "Canonical claim B", "status": "open", "confidence": None, "tags": [], "created_by": AGENT_ID, "superseded_by": None, "created_at": APPLIED_AT, "updated_at": APPLIED_AT, }, ], "sources": [ { "id": SOURCE_ID, "source_type": "paper", "url": "https://example.test/lifecycle", "storage_path": None, "excerpt": "Bound lifecycle evidence.", "hash": "sha256:lifecycle-source", "created_by": AGENT_ID, "created_at": APPLIED_AT, } ], "claim_evidence": [ { "id": evidence_id, "claim_id": CLAIM_A, "source_id": SOURCE_ID, "role": "grounds", "weight": 0.9, "created_by": AGENT_ID, "created_at": APPLIED_AT, } ], "claim_edges": [ { "id": edge_id, "from_claim": CLAIM_A, "to_claim": CLAIM_B, "edge_type": "supports", "weight": 0.7, "created_by": AGENT_ID, "created_at": APPLIED_AT, } ], "reasoning_tools": [ { "id": TOOL_ID, "agent_id": AGENT_ID, "name": "Lifecycle verifier", "description": "Checks the apply and compensating rollback boundary.", "category": "verification", "created_at": APPLIED_AT, "updated_at": APPLIED_AT, } ], } def _empty_targets() -> dict[str, list[dict]]: return {family: [] for family in ROW_FAMILIES} def _apply_receipt(proposal: dict, rows: dict[str, list[dict]], preflight: dict) -> dict: applied = copy.deepcopy(proposal) applied.update( { "status": "applied", "applied_by_handle": apply.SERVICE_AGENT_HANDLE, "applied_by_agent_id": AGENT_ID, "applied_at": APPLIED_AT, } ) 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), apply_sql_sha256=lifecycle.sha256_text(apply_sql), apply_sql_source="exact_executed_sql", exported_at_utc="2026-07-15T10:16:00+00:00", ) def _materials() -> tuple[dict, dict, dict, dict, str]: proposal = _proposal_before() approval = _approval_snapshot(proposal) preflight = lifecycle.build_fresh_preflight( proposal, _empty_targets(), captured_at_utc="2026-07-15T10:00:00+00:00", ) 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 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, preflight=preflight, apply_receipt=receipt, rollback_sql=rollback_sql, proposal_after=proposal_after, approval_after=copy.deepcopy(approval), target_rows_after=_empty_targets(), counts_before={family: 7 + index for index, family in enumerate(ROW_FAMILIES)}, counts_after={family: 7 + index for index, family in enumerate(ROW_FAMILIES)}, unrelated_before={"fingerprint": "same", "row_count": 17}, unrelated_after={"fingerprint": "same", "row_count": 17}, rollback_replay_refused=True, generated_at_utc="2026-07-15T10:30:00+00:00", ) def _destination(*, candidate_present: bool = True) -> dict: lifecycle_receipt = _passing_lifecycle_receipt() candidates = [] if candidate_present: candidates.append( { "id": PROPOSAL_ID, "proposal_type": "approve_claim", "status": "approved", "applied_at": None, "proposal_payload_sha256": lifecycle_receipt["proposal_payload_sha256"], "apply_payload_sha256": lifecycle_receipt["apply_payload_sha256"], "approval_snapshot_sha256": lifecycle_receipt["approval_snapshot_sha256"], "reviewed_by_handle": "independent-reviewer", "reviewed_by_agent_id": None, "reviewed_at": REVIEWED_AT, "review_note": "Approved this exact strict payload.", "immutable_approval_exact": True, } ) return { "observed_at_utc": "2026-07-15T10:45:00+00:00", "read_only_transaction": True, "container": "teleo-pg", "database": "teleo", "system_identifier": "7649789040005668902", "server_version_num": "160014", "approval_gate": { "immutable_approval_table_present": True, "approve_function_present": True, "assert_function_present": True, "finish_function_present": True, "review_role_present": True, "apply_role_present": True, }, "approved_strict_candidates": candidates, "downstream_dependency_count": 0, } def _authorization_materials() -> tuple[dict, dict, dict, dict]: proposal, approval, preflight, apply_receipt, _production_rollback_sql = _materials() receipt = _passing_lifecycle_receipt() destination = _destination() destination_identity = { key: destination[key] for key in ("container", "database", "system_identifier", "server_version_num") } production_rollback_artifact = lifecycle.build_production_rollback_artifact( proposal_before=proposal, approval_snapshot=approval, preflight=preflight, apply_receipt=apply_receipt, destination=destination_identity, generated_at_utc="2026-07-15T10:49:00+00:00", ) packet = lifecycle.build_authorization_packet( receipt, destination, repo_git_sha=lifecycle.current_git_sha(REPO_ROOT), apply_engine_path=REPO_ROOT / "scripts" / "apply_proposal.py", production_rollback_artifact=production_rollback_artifact, generated_at_utc="2026-07-15T10:50:00+00:00", ) record = { "authorized": True, "rollback_authorized": True, "binding_sha256": packet["binding_sha256"], "production_rollback_sql_sha256": production_rollback_artifact["rollback_sql_sha256"], "operator_identity": "release-operator@example.test", "received_at_utc": "2026-07-15T11:00:00+00:00", "expires_at_utc": "2026-07-15T13:00:00+00:00", } 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 def test_fresh_preflight_requires_every_owned_target_to_be_absent_and_is_deterministic(): proposal = _proposal_before() first = lifecycle.build_fresh_preflight( proposal, _empty_targets(), captured_at_utc="2026-07-15T10:00:00+00:00", ) second = lifecycle.build_fresh_preflight( proposal, _empty_targets(), captured_at_utc="2026-07-15T10:01:00+00:00", ) assert first["fresh_owned_targets"] is True assert first["preflight_sha256"] == second["preflight_sha256"] assert first["preflight_artifact_sha256"] != second["preflight_artifact_sha256"] assert first["captured_at_utc"] != second["captured_at_utc"] nonfresh = _empty_targets() nonfresh["claims"] = [{"id": CLAIM_A}] with pytest.raises(ValueError, match="absent target rows"): lifecycle.build_fresh_preflight(proposal, nonfresh) incomplete = _empty_targets() incomplete.pop("claim_edges") with pytest.raises(ValueError, match="every approve_claim target table"): 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", ( "artifact", "schema", "captured_at_utc", "proposal_id", "proposal_payload_sha256", "apply_payload_sha256", "target_family_missing", "target_row_present", "fresh_flag_type", "state_hash", "artifact_hash", "extra_field", ), ) def test_rollback_refuses_every_preflight_contract_mutation(mutation: str): proposal, approval, preflight, receipt, _rollback_sql = _materials() changed = copy.deepcopy(preflight) if mutation == "artifact": changed["artifact"] = "proposal_apply_fresh_preflight_tampered" elif mutation == "schema": changed["schema"] = "livingip.proposalApplyFreshPreflight.v0" elif mutation == "captured_at_utc": changed["captured_at_utc"] = "2026-07-15T10:00:01+00:00" elif mutation == "proposal_id": changed["proposal_id"] = CLAIM_A elif mutation == "proposal_payload_sha256": changed["proposal_payload_sha256"] = "0" * 64 elif mutation == "apply_payload_sha256": changed["apply_payload_sha256"] = "0" * 64 elif mutation == "target_family_missing": changed["target_rows_before"].pop("claim_edges") elif mutation == "target_row_present": changed["target_rows_before"]["claims"] = [{"id": CLAIM_A}] elif mutation == "fresh_flag_type": changed["fresh_owned_targets"] = 1 elif mutation == "state_hash": changed["preflight_sha256"] = "0" * 64 elif mutation == "artifact_hash": changed["preflight_artifact_sha256"] = "0" * 64 else: changed["unexpected"] = True with pytest.raises(ValueError): lifecycle.build_compensating_rollback_sql(proposal, approval, changed, receipt) def test_compensating_rollback_is_atomic_deterministic_and_refuses_state_drift(): proposal, approval, preflight, receipt, rollback_sql = _materials() assert rollback_sql == lifecycle.build_compensating_rollback_sql(proposal, approval, preflight, receipt) assert rollback_sql.startswith("begin;") assert rollback_sql.rstrip().endswith("commit;") assert "proposal rollback drift:" in rollback_sql assert "to_jsonb(t)" in rollback_sql assert rollback_sql.index("proposal rollback drift:") < rollback_sql.index("delete from public.claim_evidence") delete_positions = [ 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 = '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) drifted_receipt["proposal"]["payload"]["apply_payload"]["claims"][0]["text"] = "drift" with pytest.raises(ValueError, match="payload does not match preflight"): lifecycle.build_compensating_rollback_sql(proposal, approval, preflight, drifted_receipt) @given( suffix=st.text( alphabet=string.ascii_letters + string.digits + " -_", min_size=1, max_size=32, ) ) @settings(max_examples=30, deadline=None) def test_rollback_refuses_any_canonical_row_tampering_without_rebuilt_receipt(suffix: str): proposal, approval, preflight, receipt, _rollback_sql = _materials() changed = copy.deepcopy(receipt) changed["canonical_rows"]["claims"][0]["text"] = f"mutated:{suffix}" with pytest.raises(ValueError, match="replay-contract validation"): lifecycle.build_compensating_rollback_sql(proposal, approval, preflight, changed) @pytest.mark.parametrize( "mutation", ("contract_bool", "expected_count_bool", "actual_count_float", "check_integer"), ) def test_replay_receipt_json_types_are_exact_not_python_equal(mutation: str): proposal, approval, preflight, receipt, _rollback_sql = _materials() changed = copy.deepcopy(receipt) if mutation == "contract_bool": changed["receipt_contract_version"] = True elif mutation == "expected_count_bool": changed["expected_row_counts"]["sources"] = True elif mutation == "actual_count_float": changed["actual_row_counts"]["sources"] = 1.0 else: changed["checks"]["proposal_applied"] = 1 with pytest.raises(ValueError, match="replay-contract validation"): lifecycle.build_compensating_rollback_sql(proposal, approval, preflight, changed) 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, preflight) changed_sql = lifecycle.build_compensating_rollback_sql( proposal, approval, preflight, changed_receipt, ) assert lifecycle.sha256_text(changed_sql) != lifecycle.sha256_text(rollback_sql) assert "to_jsonb(t)" in changed_sql assert "proposal rollback drift: claims[0] row does not match apply receipt" in changed_sql @pytest.mark.parametrize("family", ("claim_evidence", "claim_edges")) def test_fresh_apply_receipt_requires_deterministic_generated_row_ids(family: str): 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, preflight) with pytest.raises(ValueError, match="is not deterministic"): lifecycle.build_compensating_rollback_sql( proposal, approval, preflight, changed_receipt, ) def test_production_rollback_artifact_reconstructs_exact_sql_and_rejects_noop_substitution(): proposal, approval, preflight, apply_receipt, rollback_sql = _materials() destination = { "container": "teleo-pg", "database": "teleo", "system_identifier": "7649789040005668902", "server_version_num": "160014", } artifact = lifecycle.build_production_rollback_artifact( proposal_before=proposal, approval_snapshot=approval, preflight=preflight, apply_receipt=apply_receipt, destination=destination, generated_at_utc="2026-07-15T10:49:00+00:00", ) assert ( lifecycle.validate_production_rollback_artifact( artifact, expected_destination=destination, ) == artifact ) assert artifact["rollback_sql"] == rollback_sql noop = copy.deepcopy(artifact) noop["rollback_sql"] = ( "begin; -- proposal rollback drift: -- immutable approval snapshot drifted " "-- set status = 'canceled'\ncommit;\n" ) noop["rollback_sql_sha256"] = lifecycle.sha256_text(noop["rollback_sql"]) unsigned = dict(noop) unsigned.pop("artifact_sha256") noop["artifact_sha256"] = lifecycle.sha256_json(unsigned) with pytest.raises(ValueError, match="reconstructed exact SQL"): lifecycle.validate_production_rollback_artifact(noop, expected_destination=destination) def test_lifecycle_receipt_proves_inverse_and_has_self_consistent_hashes(): receipt = _passing_lifecycle_receipt() unsigned = dict(receipt) receipt_hash = unsigned.pop("lifecycle_receipt_sha256") assert receipt["pass"] is True assert receipt["current_tier"] == "T2_runtime" assert all(receipt["checks"].values()) assert all(not rows for rows in receipt["rollback"]["target_rows_after"].values()) assert receipt["rollback"]["counts_after_rollback"] == receipt["rollback"]["counts_before_apply"] assert receipt["production_apply_executed"] is False assert receipt["production_apply_authorization_present"] is False assert receipt_hash == lifecycle.sha256_json(unsigned) def test_lifecycle_receipt_downgrades_when_unrelated_state_changes(): proposal, approval, preflight, receipt, rollback_sql = _materials() result = lifecycle.build_lifecycle_receipt( proposal_before=proposal, approval_before=approval, preflight=preflight, apply_receipt=receipt, rollback_sql=rollback_sql, proposal_after=copy.deepcopy(proposal), approval_after=copy.deepcopy(approval), target_rows_after=_empty_targets(), counts_before={family: 1 for family in ROW_FAMILIES}, counts_after={family: 1 for family in ROW_FAMILIES}, unrelated_before={"fingerprint": "before"}, unrelated_after={"fingerprint": "after"}, rollback_replay_refused=True, ) assert result["pass"] is False assert result["current_tier"] == "T1_model" assert result["checks"]["unrelated_rows_unchanged"] is False def test_retained_current_lifecycle_evidence_is_self_consistent(): report_dir = REPO_ROOT / "docs" / "reports" / "leo-working-state-20260709" canary = json.loads((report_dir / "proposal-apply-lifecycle-canary-current.json").read_text(encoding="utf-8")) receipt = json.loads((report_dir / "proposal-apply-lifecycle-receipt-current.json").read_text(encoding="utf-8")) rollback = (report_dir / "proposal-apply-clone-rollback-current.sql").read_text(encoding="utf-8") unsigned = dict(receipt) receipt_sha256 = unsigned.pop("lifecycle_receipt_sha256") assert canary["status"] == "pass" assert canary["current_tier"] == "T2_runtime" assert canary["applied_rollback_receipt"] == receipt assert canary["outer_isolation"]["cleanup_readback"]["label_scoped_orphan_count"] == 0 assert all(delta == 0 for delta in canary["outer_isolation"]["canonical_table_deltas"].values()) assert receipt["pass"] is True assert receipt["production_apply_executed"] is False assert receipt_sha256 == lifecycle.sha256_json(unsigned) assert hashlib.sha256(rollback.encode("utf-8")).hexdigest() == receipt["rollback"]["rollback_sql_sha256"] def test_retained_production_authorization_packet_is_exact_and_unexecuted(): packet_path = ( REPO_ROOT / "docs" / "reports" / "leo-working-state-20260709" / "proposal-apply-production-authorization-packet-current.json" ) packet = json.loads(packet_path.read_text(encoding="utf-8")) assert packet["binding_sha256"] == lifecycle.sha256_json(packet["binding"]) assert packet["binding"]["engine"]["apply_engine_path"] == "scripts/apply_proposal.py" assert packet["binding"]["approval_gate_dependency"]["path"] == "scripts/kb_apply_prereqs.sql" assert packet["production_preconditions"]["approval_gate_ready"] is False assert packet["production_preconditions"]["strict_proposal_present_and_approved"] is False assert packet["production_preconditions"]["exact_production_rollback_packet_ready"] is False assert packet["safe_to_enter_apply_window"] is False assert packet["production_apply_authorization_present"] is False assert packet["production_apply_executed"] is False assert all(value is False for value in packet["scope_exclusions"].values()) assert "/private/tmp" not in json.dumps(packet, sort_keys=True) def test_authorization_packet_is_exact_but_never_self_authorizes_production(): packet, record, destination, production_rollback_artifact = _authorization_materials() 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 packet["required_tier"] == "T6_authorized_production" assert packet["current_tier"] == "T3_live_readonly" assert packet["safe_to_enter_apply_window"] is False assert packet["production_apply_authorization_present"] is False assert packet["production_apply_executed"] is False assert verification["safe_to_enter_apply_window"] is True assert verification["production_apply_executed"] is False 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) destination["approval_gate"] = {key: False for key in destination["approval_gate"]} packet = lifecycle.build_authorization_packet( receipt, destination, repo_git_sha=lifecycle.current_git_sha(REPO_ROOT), apply_engine_path=REPO_ROOT / "scripts" / "apply_proposal.py", ) markdown_path = tmp_path / "authorization.md" lifecycle.write_authorization_markdown(markdown_path, packet) assert packet["production_preconditions"]["approval_gate_ready"] is False assert packet["production_preconditions"]["matching_candidate_count"] == 0 assert packet["production_preconditions"]["exact_production_rollback_packet_ready"] is False assert packet["binding"]["production_rollback"]["rollback_sql_sha256"] is None assert lifecycle.sha256_json(packet["binding"]) == packet["binding_sha256"] assert packet["safe_to_enter_apply_window"] is False assert packet["production_apply_authorization_present"] is False assert packet["production_apply_executed"] is False assert packet["next_exact_action"].startswith("deploy and independently verify") assert "This text is not actionable" in markdown_path.read_text(encoding="utf-8") @given(value=st.sampled_from([False, None, 0, 1, "true", "yes", [], {}])) @settings(max_examples=8, deadline=None) def test_authorized_field_requires_the_exact_boolean_true(value): packet, record, destination, production_rollback_artifact = _authorization_materials() record["authorized"] = value record["authorization_text"] = lifecycle.expected_authorization_text(packet, record) 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"]["authorized_is_exact_true"] is False assert verification["safe_to_enter_apply_window"] is False AUTHORIZATION_MUTATIONS = ( "packet_schema", "packet_readonly", "approval_gate_packet", "production_rollback_packet", "strict_candidate", "rollback_authorized", "production_rollback_sha256", "binding_sha256", "operator_identity", "authorization_window", "authorization_text", "destination_container", "destination_database", "destination_system_identifier", "destination_server_version", "approval_gate_action", "strict_candidate_action", "apply_engine_sha256", "approval_gate_sha256", "binding_proposal", "binding_gate_false", "binding_preconditions", "packet_safe_flag", "scope_exclusion", "authorization_template", "record_extra_field", "action_readback_stale", "production_rollback_binding", ) @given(mutation=st.sampled_from(AUTHORIZATION_MUTATIONS)) @settings(max_examples=len(AUTHORIZATION_MUTATIONS), deadline=None) def test_every_authorization_binding_field_fails_closed_when_mutated(mutation: str): packet, record, destination, production_rollback_artifact = _authorization_materials() expected_failed_check = { "packet_schema": "packet_schema_exact", "packet_readonly": "packet_destination_readback_was_read_only", "approval_gate_packet": "approval_gate_ready_at_packet", "production_rollback_packet": "exact_production_rollback_packet_ready", "strict_candidate": "strict_proposal_present_and_approved", "rollback_authorized": "rollback_authorized_is_exact_true", "production_rollback_sha256": "production_rollback_sql_sha256_exact", "binding_sha256": "binding_sha256_exact", "operator_identity": "operator_identity_present", "authorization_window": "authorization_window_valid", "authorization_text": "authorization_text_exact", "destination_container": "destination_identity_exact", "destination_database": "destination_identity_exact", "destination_system_identifier": "destination_identity_exact", "destination_server_version": "destination_identity_exact", "approval_gate_action": "approval_gate_identity_exact_at_action", "strict_candidate_action": "strict_proposal_exact_at_action", "apply_engine_sha256": "apply_engine_sha256_exact", "approval_gate_sha256": "approval_gate_prerequisite_sha256_exact", "binding_proposal": "packet_binding_sha256_self_consistent", "binding_gate_false": "approval_gate_ready_at_packet", "binding_preconditions": "packet_preconditions_bound_exactly", "packet_safe_flag": "packet_never_self_authorizes_or_expands_scope", "scope_exclusion": "packet_never_self_authorizes_or_expands_scope", "authorization_template": "authorization_text_template_exact", "record_extra_field": "authorization_record_contract_exact", "action_readback_stale": "destination_readback_fresh_and_read_only_at_action", "production_rollback_binding": "production_rollback_sql_sha256_exact", }[mutation] if mutation == "packet_schema": packet["schema"] = "livingip.proposalApplyAuthorizationPacket.v0" elif mutation == "packet_readonly": packet["production_preconditions"]["destination_readback_was_read_only"] = False elif mutation == "approval_gate_packet": packet["production_preconditions"]["approval_gate_ready"] = False elif mutation == "production_rollback_packet": packet["production_preconditions"]["exact_production_rollback_packet_ready"] = False elif mutation == "strict_candidate": packet["production_preconditions"]["strict_proposal_present_and_approved"] = False elif mutation == "rollback_authorized": record["rollback_authorized"] = False elif mutation == "production_rollback_sha256": record["production_rollback_sql_sha256"] = "f" * 64 elif mutation == "binding_sha256": record["binding_sha256"] = "0" * 64 elif mutation == "operator_identity": record["operator_identity"] = " " elif mutation == "authorization_window": record["expires_at_utc"] = "2026-07-15T11:30:00+00:00" elif mutation == "authorization_text": record["authorization_text"] += " altered" elif mutation.startswith("destination_"): key = { "destination_container": "container", "destination_database": "database", "destination_system_identifier": "system_identifier", "destination_server_version": "server_version_num", }[mutation] destination[key] = f"wrong-{key}" elif mutation == "approval_gate_action": destination["approval_gate"]["review_role_present"] = False elif mutation == "strict_candidate_action": destination["approved_strict_candidates"] = [] elif mutation == "apply_engine_sha256": packet["binding"]["engine"]["apply_engine_sha256"] = "0" * 64 elif mutation == "approval_gate_sha256": packet["binding"]["approval_gate_dependency"]["sha256"] = "0" * 64 elif mutation == "binding_proposal": packet["binding"]["proposal"]["id"] = CLAIM_A elif mutation == "binding_gate_false": packet["binding"]["approval_gate"]["review_role_present"] = False elif mutation == "binding_preconditions": packet["binding"]["production_preconditions"]["approval_gate_ready"] = False elif mutation == "packet_safe_flag": packet["safe_to_enter_apply_window"] = True elif mutation == "scope_exclusion": packet["scope_exclusions"]["telegram_send_authorized"] = True elif mutation == "authorization_template": packet["authorization_text_template"] += " Also authorize all other mutations." elif mutation == "record_extra_field": record["unexpected"] = True elif mutation == "action_readback_stale": destination["observed_at_utc"] = "2026-07-15T10:59:59+00:00" elif mutation == "production_rollback_binding": packet["binding"]["production_rollback"]["rollback_sql_sha256"] = "f" * 64 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"][expected_failed_check] is False assert verification["safe_to_enter_apply_window"] is False assert verification["production_apply_authorization_present"] is False def test_false_bound_gate_cannot_pass_even_after_packet_and_record_are_rehashed(): packet, record, destination, production_rollback_artifact = _authorization_materials() packet["binding"]["approval_gate"]["review_role_present"] = False destination["approval_gate"]["review_role_present"] = False packet["binding_sha256"] = lifecycle.sha256_json(packet["binding"]) packet["authorization_record_required"]["binding_sha256"] = packet["binding_sha256"] packet["authorization_text_template"] = lifecycle._authorization_text_template( packet["binding"], packet["binding_sha256"], ) record["binding_sha256"] = packet["binding_sha256"] record["authorization_text"] = lifecycle.expected_authorization_text(packet, record) 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"]["packet_binding_sha256_self_consistent"] is True assert verification["checks"]["approval_gate_ready_at_packet"] is False assert verification["safe_to_enter_apply_window"] is False def test_authorization_packet_uses_repo_relative_dependency_paths_only(): packet, _record, _destination, _production_rollback_artifact = _authorization_materials() encoded = json.dumps(packet, sort_keys=True) assert packet["binding"]["engine"]["apply_engine_path"] == "scripts/apply_proposal.py" assert packet["binding"]["approval_gate_dependency"]["path"] == "scripts/kb_apply_prereqs.sql" assert "/private/tmp" not in encoded assert "/tmp" not in encoded