feat(sre): register independent verifier coverage
This commit is contained in:
Your Name
231
apps/api/src/services/independent_verifier_registry.py
Normal file
231
apps/api/src/services/independent_verifier_registry.py
Normal file
@@ -0,0 +1,231 @@
|
||||
"""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": None,
|
||||
"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,
|
||||
},
|
||||
}
|
||||
@@ -279,12 +279,21 @@ def load_sre_k3s_controlled_automation_work_items(
|
||||
completion = _dict(payload, "completion_contract", path)
|
||||
if completion.get("source_test_cd_green_is_runtime_closure") is not False:
|
||||
raise ValueError(f"{path}: source/CD cannot be runtime closure")
|
||||
from src.services.independent_verifier_registry import (
|
||||
build_independent_verifier_registry,
|
||||
)
|
||||
|
||||
verifier_registry = build_independent_verifier_registry()
|
||||
if verifier_registry.get("source_coverage_complete") is not True:
|
||||
raise ValueError(f"{path}: independent verifier source coverage incomplete")
|
||||
payload["independent_verifier_registry"] = verifier_registry
|
||||
return payload
|
||||
|
||||
|
||||
def build_sre_k3s_program_projection(payload: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Return a bounded cockpit projection without dropping truth labels."""
|
||||
|
||||
verifier_registry = payload["independent_verifier_registry"]
|
||||
return {
|
||||
"schema_version": payload["schema_version"],
|
||||
"program_id": payload["program_id"],
|
||||
@@ -302,6 +311,14 @@ def build_sre_k3s_program_projection(payload: dict[str, Any]) -> dict[str, Any]:
|
||||
"gemini_paid_call_allowed"
|
||||
],
|
||||
"domain_count": len(payload["domain_routes"]),
|
||||
"independent_verifier_registry": {
|
||||
"status": verifier_registry["status"],
|
||||
"source_coverage_complete": verifier_registry[
|
||||
"source_coverage_complete"
|
||||
],
|
||||
"runtime_closed": verifier_registry["runtime_closed"],
|
||||
"rollups": verifier_registry["rollups"],
|
||||
},
|
||||
"agent99_bridge": payload["agent99_host_operations_bridge"],
|
||||
"full_readback_api": (
|
||||
"/api/v1/agents/sre-k3s-controlled-automation-work-items"
|
||||
|
||||
Reference in New Issue
Block a user