"""Machine-readable source coverage for every controlled SRE verifier lane. This registry is source truth only. It proves that a typed executor is bound to a separate verifier contract; it never turns source coverage into a production receipt or runtime closure. """ from __future__ import annotations from collections.abc import Iterable, Mapping from typing import Any SCHEMA_VERSION = "independent_verifier_registry_v1" _DOMAIN_CONTRACTS: tuple[dict[str, Any], ...] = ( { "domain": "kubernetes_workload", "mutating": True, "executor": "kubernetes_controlled_executor", "verifier": "kubernetes_rollout_verifier", "runtime_write_allowed": True, "receipt_required": True, "source_ref": "apps/api/src/services/kubernetes_rollout_verifier.py", }, { "domain": "host_systemd", "mutating": True, "executor": "host_ansible_executor", "verifier": "asset_specific_read_only_host_postconditions", "runtime_write_allowed": True, "receipt_required": True, "source_ref": "apps/api/src/services/awooop_ansible_post_verifier.py", }, { "domain": "docker_container", "mutating": True, "executor": "host_ansible_executor", "verifier": "asset_specific_read_only_host_postconditions", "runtime_write_allowed": True, "receipt_required": True, "source_ref": "apps/api/src/services/awooop_ansible_post_verifier.py", }, { "domain": "windows_vmware", "mutating": True, "executor": "Agent99", "verifier": "agent99_independent_runtime_verifier", "runtime_write_allowed": True, "receipt_required": True, "source_ref": "apps/api/src/services/agent99_controlled_dispatch_ledger.py", }, { "domain": "database", "mutating": True, "executor": "db_bounded_executor", "verifier": "db_independent_verifier", "runtime_write_allowed": True, "receipt_required": True, "source_ref": "apps/api/src/services/db_bounded_executor.py", }, { "domain": "control_plane_recovery", "mutating": True, "executor": "Agent99", "verifier": "agent99_independent_runtime_verifier", "runtime_write_allowed": True, "receipt_required": True, "source_ref": "apps/api/src/services/agent99_controlled_dispatch_ledger.py", }, { "domain": "backup_restore", "mutating": False, "executor": "Agent99", "break_glass_executor": "backup_restore_break_glass", "verifier": "backup_restore_readback_verifier", "runtime_write_allowed": False, "receipt_required": True, "source_ref": "apps/api/src/services/backup_restore_signal_automation.py", }, { "domain": "unknown", "mutating": False, "executor": None, "verifier": "asset_identity_drift_verifier", "runtime_write_allowed": False, "receipt_required": True, "source_ref": "apps/api/src/services/controlled_alert_target_router.py", }, ) _EXTERNAL_READBACKS: tuple[dict[str, Any], ...] = ( { "readback_id": "host111_provider_origins", "verifier": "asset_specific_read_only_host_postconditions", "required_origins": ["host_111", "host_120", "host_121"], "runtime_status": "receipt_pending", }, { "readback_id": "gcp_provider_exact_fresh_series", "verifier": "provider_health_generation_and_transport_receipt_verifier", "required_targets": ["ollama_gcp_a", "ollama_gcp_b"], "runtime_status": "receipt_pending", }, { "readback_id": "gitea_native_metrics", "verifier": "gitea_native_metrics_freshness_verifier", "runtime_status": "receipt_pending", }, { "readback_id": "host99_alertmanager_pull_relay", "verifier": "agent99_alertmanager_pull_relay_verifier", "runtime_status": "receipt_pending", }, ) def build_independent_verifier_registry( *, ansible_catalog: Iterable[Mapping[str, Any]] | None = None, ansible_registered_catalog_ids: Iterable[str] | None = None, ) -> dict[str, Any]: """Return deterministic source coverage without performing a runtime read.""" if ansible_catalog is None: from src.services.awooop_ansible_audit_service import list_ansible_catalog ansible_catalog = list_ansible_catalog() if ansible_registered_catalog_ids is None: from src.services.awooop_ansible_post_verifier import registered_catalog_ids ansible_registered_catalog_ids = registered_catalog_ids() catalog_rows = [dict(row) for row in ansible_catalog] catalog_ids = { str(row.get("catalog_id") or "").strip() for row in catalog_rows if str(row.get("catalog_id") or "").strip() } registered_ids = { str(catalog_id).strip() for catalog_id in ansible_registered_catalog_ids if str(catalog_id).strip() } missing_postconditions = sorted(catalog_ids - registered_ids) orphan_postconditions = sorted(registered_ids - catalog_ids) unsafe_auto_apply = sorted( str(row.get("catalog_id") or "") for row in catalog_rows if row.get("auto_apply_enabled") is True and row.get("supports_check_mode") is not True ) invalid_break_glass = sorted( str(row.get("catalog_id") or "") for row in catalog_rows if row.get("supports_check_mode") is not True and not ( row.get("auto_apply_enabled") is False and row.get("break_glass_required") is True ) ) domain_issues: list[str] = [] for contract in _DOMAIN_CONTRACTS: domain = str(contract["domain"]) executor = contract.get("executor") verifier = str(contract.get("verifier") or "") if not verifier: domain_issues.append(f"{domain}:verifier_missing") if contract.get("receipt_required") is not True: domain_issues.append(f"{domain}:receipt_not_required") if contract.get("mutating") is True: if not executor: domain_issues.append(f"{domain}:executor_missing") if str(executor) == verifier: domain_issues.append(f"{domain}:executor_self_verifies") elif contract.get("runtime_write_allowed") is not False: domain_issues.append(f"{domain}:read_only_domain_write_allowed") source_blockers = [ *(f"ansible_postcondition_missing:{item}" for item in missing_postconditions), *(f"ansible_postcondition_orphan:{item}" for item in orphan_postconditions), *(f"ansible_auto_apply_without_check_mode:{item}" for item in unsafe_auto_apply), *(f"ansible_non_break_glass_without_check_mode:{item}" for item in invalid_break_glass), *domain_issues, ] source_coverage_complete = not source_blockers domains = [ {**contract, "cross_domain_fallback_allowed": False} for contract in _DOMAIN_CONTRACTS ] return { "schema_version": SCHEMA_VERSION, "status": ( "source_coverage_complete_runtime_receipts_pending" if source_coverage_complete else "source_coverage_incomplete_fail_closed" ), "source_coverage_complete": source_coverage_complete, "runtime_closed": False, "domain_contracts": domains, "ansible_catalog_coverage": { "catalog_count": len(catalog_ids), "check_mode_catalog_count": sum( row.get("supports_check_mode") is True for row in catalog_rows ), "auto_apply_catalog_count": sum( row.get("auto_apply_enabled") is True for row in catalog_rows ), "registered_postcondition_count": len(registered_ids), "missing_postcondition_catalog_ids": missing_postconditions, "orphan_postcondition_catalog_ids": orphan_postconditions, }, "external_readbacks": [dict(row) for row in _EXTERNAL_READBACKS], "source_blockers": source_blockers, "runtime_blockers": [ "production_same_run_verifier_receipts_pending", "source_coverage_is_not_runtime_closure", ], "rollups": { "domain_count": len(domains), "mutating_domain_count": sum( contract["mutating"] is True for contract in domains ), "read_only_domain_count": sum( contract["mutating"] is False for contract in domains ), "source_blocker_count": len(source_blockers), "runtime_closed_domain_count": 0, }, }