Files
awoooi/apps/api/src/services/controlled_alert_target_router.py
2026-07-19 03:21:22 +08:00

1185 lines
43 KiB
Python

"""Domain-aware routing for non-Kubernetes controlled alert targets.
The Alertmanager namespace label is a source attribute, not proof that every
alert is a Kubernetes workload. Cold-start/reboot recovery is a host control
plane lane. It must never fall through to a guessed ``kubectl`` action or the
generic Ansible keyword catalog merely because the source payload mentions a
host, backup, or container.
This module is deliberately pure. It only builds machine-readable routing
and receipt contracts; it does not dispatch Agent99, run a command, or write
runtime state.
"""
from __future__ import annotations
import hashlib
import re
from collections.abc import Mapping
from typing import Any
from src.services.ai_provider_policy import PRODUCTION_PROVIDER_ORDER
from src.services.service_registry import (
ServiceRegistryClient,
StatefulLevel,
get_service_registry,
)
_RECOVERY_MARKERS = (
"cold-start",
"cold_start",
"cold start",
"full-stack-cold-start",
"full_stack_cold_start",
"full stack cold start",
"reboot-auto-recovery",
"reboot_auto_recovery",
"reboot auto recovery",
)
_BACKUP_RESTORE_MARKERS = (
"backup",
"restore",
"escrow",
"offsite",
"retention",
)
_WINDOWS_VMWARE_MARKERS = (
"agent99",
"windows99",
"windows 99",
"vmware",
"vmx",
)
_AGENT99_CONTROL_PLANE_HEALTH_ALERTS = {"agent99serviceunhealthy"}
_TELEGRAM_CALLBACK_FORWARDER_ALERTS = {
"telegramcallbackingressunverified",
"telegramcallbackforwarderdrift",
}
_DISK_ALERTS = {
"hostoutofdiskspace",
"hostdiskusagehigh",
"hostdiskusagecritical",
"hostdiskwillfillin24hours",
}
_NODE_EXPORTER_ALERTS = {
"nodeexporterdown",
"nodeexporterscrapedown",
"nodeexporterunhealthy",
}
_ALERTMANAGER_DELIVERY_ALERTS = {"alertchainbrokenalertmanager"}
_PROVIDER_FRESHNESS_ALERTS = {
"providerfreshnesssignal",
"sourceprovideringestionstale",
}
_OLLAMA_LOCAL_RUNTIME_ALERTS = {
"ollamalocalunavailable",
"ollama111unavailable",
}
_KUBERNETES_KINDS = {
"cronjob",
"daemonset",
"deployment",
"job",
"pod",
"statefulset",
}
_KUBERNETES_RUNTIME_UID = re.compile(
r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$"
)
_KUBERNETES_RUNTIME_IDENTITY_SOURCE = "kubernetes_api_uid_readback"
_HOSTS = {
"99": ("host_99", "192.168.0.99"),
"110": ("host_110", "192.168.0.110"),
"111": ("host_111", "192.168.0.111"),
"112": ("host_112", "192.168.0.112"),
"120": ("host_120", "192.168.0.120"),
"121": ("host_121", "192.168.0.121"),
"188": ("host_188", "192.168.0.188"),
}
def _text_token(value: str | None) -> str:
return " ".join(str(value or "").strip().lower().split())
def _asset_token(value: str) -> str:
normalized = re.sub(r"[^a-z0-9_.-]+", "-", value.lower()).strip("-.")
return normalized[:96] or "unbound"
def _compact_token(value: str | None) -> str:
return re.sub(r"[^a-z0-9]+", "", _text_token(value))
def _kubernetes_drift_work_item_id(*values: str | None) -> str:
key = "|".join(_text_token(value) for value in values) or "missing-target"
digest = hashlib.sha256(key.encode("utf-8")).hexdigest()[:12].upper()
return f"AIA-ASSET-DRIFT-{digest}"
def _kubernetes_identity_evidence(
*,
identity: Mapping[str, Any],
labels: Mapping[str, Any],
namespace: str,
target_resource: str,
runtime_identity_evidence: Mapping[str, Any] | None,
) -> dict[str, Any]:
"""Verify a Kubernetes identity without trusting a workload-kind label alone."""
kind = _text_token(
str(
labels.get("workload_kind")
or labels.get("kubernetes_kind")
or labels.get("kind")
or ""
)
)
name = _text_token(
str(
labels.get("workload_name")
or labels.get(kind)
or labels.get("deployment")
or labels.get("statefulset")
or labels.get("daemonset")
or labels.get("cronjob")
or labels.get("job")
or labels.get("pod")
or ""
)
)
label_namespace = _text_token(str(labels.get("namespace") or ""))
requested_namespace = _text_token(namespace)
target_name = _text_token(target_resource)
registry_name = _text_token(str(identity.get("service_name") or ""))
registry_exact = bool(
identity.get("resolution_status") == "resolved"
and identity.get("asset_domain") == "kubernetes_workload"
and requested_namespace
and label_namespace == requested_namespace
and kind in _KUBERNETES_KINDS
and name
and name == target_name
and (not registry_name or registry_name == target_name)
)
runtime_receipt = (
runtime_identity_evidence
if isinstance(runtime_identity_evidence, Mapping)
else {}
)
runtime_uid = _text_token(str(runtime_receipt.get("runtime_uid") or ""))
runtime_namespace = _text_token(
str(runtime_receipt.get("namespace") or "")
)
runtime_kind = _text_token(str(runtime_receipt.get("kind") or ""))
runtime_name = _text_token(str(runtime_receipt.get("name") or ""))
runtime_source = _text_token(
str(runtime_receipt.get("source") or "")
)
runtime_status = _text_token(
str(runtime_receipt.get("status") or "")
)
runtime_uid_verified = bool(
runtime_receipt.get("schema_version")
== "kubernetes_runtime_identity_receipt_v1"
and bool(
_KUBERNETES_RUNTIME_UID.fullmatch(runtime_uid)
)
and runtime_source == _KUBERNETES_RUNTIME_IDENTITY_SOURCE
and runtime_status == "verified"
and runtime_receipt.get("durable_readback_ack") is True
and runtime_receipt.get("verified_by")
== "kubernetes_runtime_identity_verifier"
and all(
str(runtime_receipt.get(field) or "").strip()
for field in (
"receipt_id",
"trace_id",
"run_id",
"work_item_id",
)
)
and requested_namespace
and runtime_namespace == requested_namespace
and runtime_kind in _KUBERNETES_KINDS
and runtime_name
and runtime_name == target_name
and (not kind or runtime_kind == kind)
and (not name or runtime_name == name)
)
method = (
"canonical_registry_namespace_kind_name_exact"
if registry_exact
else "kubernetes_runtime_uid_readback"
if runtime_uid_verified
else "unverified"
)
evidence: dict[str, Any] = {
"schema_version": "kubernetes_asset_identity_evidence_v1",
"status": "verified" if registry_exact or runtime_uid_verified else "unverified",
"method": method,
"namespace": requested_namespace or None,
"kind": kind or runtime_kind or None,
"name": name or runtime_name or target_name or None,
"registry_identity_exact": registry_exact,
"runtime_uid_verified": runtime_uid_verified,
"required": [
"canonical_registry_identity_plus_exact_namespace_kind_name",
"or_kubernetes_api_runtime_uid_readback",
],
}
if runtime_uid_verified:
evidence["runtime_uid_sha256"] = hashlib.sha256(
runtime_uid.encode("utf-8")
).hexdigest()
evidence["runtime_identity_source"] = _KUBERNETES_RUNTIME_IDENTITY_SOURCE
return evidence
def _host_scope(*values: str | None) -> tuple[str, str] | None:
"""Resolve only an explicit, known host identity."""
for value in values:
normalized = _text_token(value)
for suffix, scope in _HOSTS.items():
patterns = (
rf"(?:^|[^0-9]){suffix}(?:[^0-9]|$)",
rf"192\.168\.0\.{suffix}(?:[^0-9]|$)",
rf"host[_-]?{suffix}(?:[^0-9]|$)",
)
if any(re.search(pattern, normalized) for pattern in patterns):
return scope
return None
def _agent99_bridge(*, execution_role: str, dispatch_allowed: bool) -> dict[str, Any]:
return {
"schema_version": "agent99_host_operations_bridge_v1",
"integration_status": "contract_required_runtime_receipt_pending",
"execution_role": execution_role,
"dispatch_allowed": dispatch_allowed,
"dispatch_identity": "agent99_controlled_dispatch_identity_v1",
"command_envelope": "agent99_controlled_dispatch_input_v1",
"dispatch_receipt": "agent99_controlled_dispatch_receipt_v1",
"outcome_contract": "agent99_outcome_contract_v1",
"completion_callback": "/api/v1/agents/agent99/completion-callback",
"required_correlation": ["trace_id", "run_id", "work_item_id"],
"required_controls": [
"canonical_asset_id",
"typed_domain",
"allowlisted_command_id",
"ttl",
"idempotency_key",
"check_receipt",
"bounded_apply_receipt",
"independent_verifier_receipt",
],
"forbidden": [
"arbitrary_command",
"secret_read",
"cross_domain_fallback",
"unresolved_asset_dispatch",
"mark_resolved_without_callback_and_verifier",
],
}
def _typed_route(
*,
resolution_status: str,
target_kind: str,
target_resource: str,
namespace: str,
canonical_asset_id: str | None,
host: str | None,
executor: str | None,
verifier: str | None,
risk_class: str,
allowed_catalog_ids: list[str] | None = None,
allowed_inventory_hosts: list[str] | None = None,
drift_work_item_id: str | None = None,
stateful_level: str | None = None,
controlled_apply_allowed: bool | None = None,
break_glass_executor: str | None = None,
execution_role: str = "policy_observer_and_no_secret_dispatch_relay",
) -> dict[str, Any]:
resolved = resolution_status == "resolved"
apply_allowed = (
resolved and risk_class != "critical"
if controlled_apply_allowed is None
else resolved and risk_class != "critical" and controlled_apply_allowed
)
agent99_dispatch = resolved and target_kind in {
"control_plane_recovery",
"windows_vmware",
# Agent99 is only the read-only evidence collector for this domain;
# backup/restore mutation remains on the critical break-glass executor.
"backup_restore",
}
return {
"schema_version": "typed_domain_target_route_v2",
"resolution_status": resolution_status,
"route_id": (
f"typed:{target_kind}:{_asset_token(canonical_asset_id or target_resource)}"
),
"target_kind": target_kind,
"target_resource": target_resource,
"canonical_asset_id": canonical_asset_id,
"source_namespace": namespace or None,
"host": host,
"stateful_level": stateful_level,
"executor": executor,
"break_glass_executor": break_glass_executor,
"verifier": verifier,
"risk_class": risk_class,
"allowed_catalog_ids": list(allowed_catalog_ids or []),
"allowed_inventory_hosts": list(allowed_inventory_hosts or []),
"controlled_apply_allowed": apply_allowed,
"critical_break_glass_required": risk_class == "critical",
"cross_domain_fallback_allowed": False,
"circuit_open_behavior": "stay_in_same_domain_and_create_repair_work_item",
"drift_work_item_id": drift_work_item_id,
"investigator_contract": {
"active": "PreDecisionInvestigator",
"holmesgpt": "shadow_pending_internal_immutable_artifact",
"required_evidence": [
"source_receipt",
"canonical_asset_identity",
"metrics_logs_traces",
"recent_change_and_related_run",
],
},
"reasoning_contract": {
"rca": "ollama_provider_chain",
"provider_order": list(PRODUCTION_PROVIDER_ORDER),
"critic": "gemini_final_fallback_explicit_cost_gate",
"paid_call_allowed": False,
},
"agent99_bridge": _agent99_bridge(
execution_role=execution_role,
dispatch_allowed=agent99_dispatch,
),
}
def resolve_typed_alert_target(
*,
alertname: str,
target_resource: str,
namespace: str,
labels: Mapping[str, Any] | None = None,
alert_category: str = "",
registry: ServiceRegistryClient | None = None,
runtime_identity_evidence: Mapping[str, Any] | None = None,
) -> dict[str, Any]:
"""Resolve exactly one domain route; unknown assets never fall back AUTO."""
labels = labels or {}
registry = registry or get_service_registry()
text = " ".join(
(
_text_token(alertname),
_text_token(target_resource),
_text_token(alert_category),
)
)
compact_alert = _compact_token(alertname)
legacy_recovery = resolve_controlled_alert_target(
alertname=alertname,
target_resource=target_resource,
namespace=namespace,
)
if legacy_recovery is not None:
route = _typed_route(
resolution_status="resolved",
target_kind="control_plane_recovery",
target_resource=target_resource or "cold-start-gate",
namespace=namespace,
canonical_asset_id="control-plane:cold-start-gate",
host=None,
executor="Agent99",
verifier="cold_start_independent_scorecard_verifier",
risk_class="high",
execution_role="single_writer_control_plane_recovery_executor",
)
route["route_id"] = "agent99_recover_after_owner_review"
return route
if compact_alert in {
"host112guestreadbackmissing",
"host112guestnotready",
} or "host112guest" in _compact_token(target_resource):
return _typed_route(
resolution_status="resolved",
target_kind="control_plane_recovery",
target_resource=target_resource or "host112-guest-readiness",
namespace=namespace,
canonical_asset_id="host-guest:host_112",
host=_HOSTS["112"][1],
executor="Agent99",
verifier="host112_guest_independent_readiness_verifier",
risk_class="high",
allowed_inventory_hosts=["host_112"],
execution_role="single_writer_host112_guest_recovery_orchestrator",
)
if compact_alert in _ALERTMANAGER_DELIVERY_ALERTS:
return _typed_route(
resolution_status="resolved",
target_kind="docker_container",
target_resource="alertmanager",
namespace=namespace,
canonical_asset_id="container:host_110:alertmanager",
host=_HOSTS["110"][1],
executor="host_ansible_executor",
verifier="alert_chain_independent_delivery_verifier",
risk_class="medium",
allowed_catalog_ids=["ansible:110-alertmanager-delivery-recovery"],
allowed_inventory_hosts=["host_110"],
execution_role="single_writer_alert_chain_container_executor",
)
if (
compact_alert in _PROVIDER_FRESHNESS_ALERTS
or "source_provider_freshness_triage" in text
or "alertchain_provider_freshness" in text
):
provider = _text_token(
str(
labels.get("source")
or labels.get("provider")
or labels.get("service")
or ""
)
)
if provider not in {"sentry", "signoz"}:
provider = "sentry-signoz"
route = _typed_route(
resolution_status="resolved",
target_kind="observability_source_ingestion",
target_resource=target_resource or "source-provider-ingestion",
namespace=namespace,
canonical_asset_id=f"observability-provider:{provider}",
host=None,
executor="provider_freshness_readonly_executor",
verifier="provider_freshness_independent_verifier",
risk_class="low",
controlled_apply_allowed=True,
execution_role=(
"single_writer_observability_source_ingestion_readonly_executor"
),
)
route["provider_freshness_contract"] = {
"schema_version": "provider_freshness_typed_route_v1",
"execution_mode": "read_only",
"required_providers": ["sentry", "signoz"],
"allowed_actions": ["provider_specific_coverage_readback"],
"mutating_actions_allowed": False,
"generic_recurrence_can_resolve": False,
}
return route
if compact_alert in _OLLAMA_LOCAL_RUNTIME_ALERTS:
return _typed_route(
resolution_status="resolved",
target_kind="host_launchagent",
target_resource="ollama-local",
namespace=namespace,
canonical_asset_id="ai-provider:ollama_local",
host=_HOSTS["111"][1],
executor="host_ansible_executor",
verifier="ollama111_k3s_path_independent_verifier",
risk_class="medium",
allowed_catalog_ids=["ansible:111-ollama-fallback"],
allowed_inventory_hosts=["host_111", "host_120", "host_121"],
execution_role="single_writer_host111_launchagent_executor",
)
if compact_alert in _AGENT99_CONTROL_PLANE_HEALTH_ALERTS:
# Agent99's own health signal is a read-only control-plane route. It
# is never permission to recover a guest, power a VM, reboot a host,
# or fall through to an arbitrary Windows command. An explicit host
# identity that is not host_99 is drift and must fail closed.
explicit_host_scope = _host_scope(
str(labels.get("host") or ""),
str(labels.get("instance") or ""),
str(labels.get("node") or ""),
target_resource,
)
if (
explicit_host_scope is not None
and explicit_host_scope != _HOSTS["99"]
):
return _typed_route(
resolution_status="asset_identity_unresolved",
target_kind="unknown",
target_resource=target_resource,
namespace=namespace,
canonical_asset_id=None,
host=None,
executor=None,
verifier=None,
risk_class="high",
drift_work_item_id=_kubernetes_drift_work_item_id(
alertname,
target_resource,
explicit_host_scope[0],
),
stateful_level=StatefulLevel.UNRESOLVED.value,
)
return _typed_route(
resolution_status="resolved",
target_kind="windows_vmware",
target_resource=target_resource or "Agent99",
namespace=namespace,
canonical_asset_id="windows-vmware:host_99",
host=_HOSTS["99"][1],
executor="Agent99",
verifier="agent99_control_plane_selfcheck_independent_verifier",
risk_class="high",
allowed_inventory_hosts=["host_99"],
controlled_apply_allowed=False,
execution_role="single_writer_agent99_control_plane_selfcheck_executor",
)
if any(marker in text for marker in _BACKUP_RESTORE_MARKERS):
identity = registry.resolve_identity(target_resource)
generic_backup_signal = _compact_token(target_resource) in {
"",
"backuprestore",
}
if (
identity.get("resolution_status") != "resolved"
and not generic_backup_signal
):
return _typed_route(
resolution_status="asset_identity_unresolved",
target_kind="unknown",
target_resource=target_resource,
namespace=namespace,
canonical_asset_id=None,
host=None,
executor=None,
verifier=None,
risk_class="critical",
drift_work_item_id=identity["drift_work_item_id"],
stateful_level=StatefulLevel.UNRESOLVED.value,
)
return _typed_route(
resolution_status="resolved",
target_kind="backup_restore",
target_resource=target_resource or "backup_restore",
namespace=namespace,
canonical_asset_id=(
identity.get("canonical_id") or "data-protection:backup-restore"
),
host=identity.get("host"),
executor="Agent99",
break_glass_executor="backup_restore_break_glass",
verifier="backup_restore_readback_verifier",
risk_class="critical",
stateful_level=identity.get("stateful_level"),
execution_role="single_writer_agent99_backup_readback_collector",
)
if any(marker in text for marker in _WINDOWS_VMWARE_MARKERS):
host_scope = _host_scope(
str(labels.get("host") or ""),
str(labels.get("instance") or ""),
target_resource,
alertname,
)
explicit_agent99_target = any(
marker in _text_token(target_resource)
for marker in ("agent99", "windows99", "windows 99")
)
if host_scope is None and explicit_agent99_target:
host_scope = _HOSTS["99"]
if host_scope is None:
identity = registry.resolve_identity(target_resource or alertname)
return _typed_route(
resolution_status="asset_identity_unresolved",
target_kind="unknown",
target_resource=target_resource,
namespace=namespace,
canonical_asset_id=None,
host=None,
executor=None,
verifier=None,
risk_class="high",
drift_work_item_id=identity["drift_work_item_id"],
stateful_level=StatefulLevel.UNRESOLVED.value,
)
return _typed_route(
resolution_status="resolved",
target_kind="windows_vmware",
target_resource=target_resource or host_scope[0],
namespace=namespace,
canonical_asset_id=f"windows-vmware:{host_scope[0]}",
host=host_scope[1],
executor="Agent99",
verifier="agent99_independent_runtime_verifier",
risk_class="high",
allowed_inventory_hosts=[host_scope[0]],
execution_role="single_writer_windows_vmware_executor",
)
if compact_alert == "wazuhalertmanagerintegrationconvergence":
identity = registry.resolve_identity(target_resource)
inventory_host = _host_scope(str(identity.get("host") or ""))
if (
identity.get("resolution_status") == "resolved"
and identity.get("canonical_id") == "service:wazuh-manager"
and inventory_host is not None
and inventory_host[0] == "host_112"
):
return _typed_route(
resolution_status="resolved",
target_kind="host_systemd",
target_resource=target_resource,
namespace=namespace,
canonical_asset_id="service:wazuh-manager",
host=inventory_host[1],
executor="host_ansible_executor",
verifier="wazuh_alert_ingress_independent_verifier",
risk_class="high",
allowed_catalog_ids=[
"ansible:wazuh-alertmanager-integration"
],
allowed_inventory_hosts=["host_112"],
stateful_level=str(identity.get("stateful_level") or ""),
)
host_scope = _host_scope(
str(labels.get("host") or ""),
str(labels.get("instance") or ""),
str(labels.get("node") or ""),
target_resource,
alertname,
)
host110_pressure_alert = compact_alert.startswith(
"host110sustainedmoderatepressure"
)
allowed_catalog_ids: list[str] = []
if compact_alert in _DISK_ALERTS and host_scope is not None:
if host_scope[0] in {"host_110", "host_188"}:
allowed_catalog_ids = [
f"ansible:{host_scope[0].removeprefix('host_')}-disk-pressure"
]
elif compact_alert in _NODE_EXPORTER_ALERTS and host_scope is not None:
if host_scope[0] == "host_110":
allowed_catalog_ids = ["ansible:110-devops"]
elif (
host110_pressure_alert
and host_scope is not None
and host_scope[0] == "host_110"
):
allowed_catalog_ids = ["ansible:110-host-pressure-readonly"]
elif (
"wazuh" in text
and host_scope is not None
and host_scope[0] == "host_112"
):
allowed_catalog_ids = ["ansible:wazuh-manager-posture-readback"]
if host_scope is not None and (
compact_alert in _DISK_ALERTS
or compact_alert in _NODE_EXPORTER_ALERTS
or host110_pressure_alert
or "wazuh" in text
):
return _typed_route(
resolution_status="resolved",
target_kind="host_systemd",
target_resource=target_resource or host_scope[0],
namespace=namespace,
canonical_asset_id=f"host:{host_scope[1]}",
host=host_scope[1],
executor="host_ansible_executor",
verifier="host_runtime_independent_verifier",
risk_class="medium",
allowed_catalog_ids=allowed_catalog_ids,
allowed_inventory_hosts=[host_scope[0]],
)
identity = registry.resolve_identity(target_resource)
if identity["resolution_status"] == "resolved":
domain = str(identity["asset_domain"])
if domain == "kubernetes_workload":
identity_evidence = _kubernetes_identity_evidence(
identity=identity,
labels=labels,
namespace=namespace,
target_resource=target_resource,
runtime_identity_evidence=runtime_identity_evidence,
)
if identity_evidence["status"] != "verified":
route = _typed_route(
resolution_status="asset_identity_unresolved",
target_kind="unknown",
target_resource=target_resource,
namespace=namespace,
canonical_asset_id=None,
host=None,
executor=None,
verifier=None,
risk_class="high",
drift_work_item_id=_kubernetes_drift_work_item_id(
namespace,
target_resource,
str(labels.get("workload_kind") or ""),
),
stateful_level=StatefulLevel.UNRESOLVED.value,
)
route["resolution_reason"] = (
"kubernetes_identity_evidence_missing_or_mismatch"
)
route["identity_evidence"] = identity_evidence
return route
route = _typed_route(
resolution_status="resolved",
target_kind="kubernetes_workload",
target_resource=target_resource,
namespace=namespace,
canonical_asset_id=str(identity["canonical_id"]),
host="k3s",
executor=str(identity.get("executor") or "") or None,
verifier=str(identity.get("verifier") or "") or None,
risk_class="medium",
stateful_level=str(identity.get("stateful_level") or ""),
controlled_apply_allowed=bool(
identity.get("controlled_apply_allowed")
),
)
route["identity_evidence"] = identity_evidence
return route
inventory_host = _host_scope(str(identity.get("host") or ""))
catalog_ids = list(identity.get("allowed_catalog_ids") or [])
exact_sentry_consumer = (
_text_token(target_resource)
== "sentry-self-hosted-snuba-profiling-functions-consumer-1"
)
exact_registry_catalog_target = bool(
domain == "host_systemd"
and _text_token(target_resource)
== _text_token(str(identity.get("service_name") or ""))
)
exact_openclaw_compose_target = bool(
domain == "docker_compose_container"
and identity.get("canonical_id") == "service:openclaw:host188"
and compact_alert in _TELEGRAM_CALLBACK_FORWARDER_ALERTS
and _text_token(target_resource)
in {"openclaw", "clawbot", "clawbot.service"}
)
if not (
exact_sentry_consumer
or exact_registry_catalog_target
or exact_openclaw_compose_target
):
catalog_ids = []
stateful_level = str(identity.get("stateful_level") or "")
risk = (
"critical"
if domain in {"backup_restore"}
else "high"
if domain in {"database", "storage"}
or stateful_level
in {StatefulLevel.BLOCK.value, StatefulLevel.CRITICAL_HITL.value}
else "medium"
)
read_only_backup_route = domain == "backup_restore"
return _typed_route(
resolution_status="resolved",
target_kind=domain,
target_resource=target_resource,
namespace=namespace,
canonical_asset_id=str(identity["canonical_id"]),
host=identity.get("host"),
executor=(
"Agent99"
if read_only_backup_route
else str(identity.get("executor") or "") or None
),
break_glass_executor=(
str(identity.get("executor") or "backup_restore_break_glass")
if read_only_backup_route
else None
),
verifier=str(identity.get("verifier") or "") or None,
risk_class=risk,
allowed_catalog_ids=catalog_ids,
allowed_inventory_hosts=[inventory_host[0]] if inventory_host else [],
stateful_level=stateful_level,
controlled_apply_allowed=bool(
identity.get("controlled_apply_allowed")
),
execution_role=(
"single_writer_agent99_backup_readback_collector"
if read_only_backup_route
else "policy_observer_and_no_secret_dispatch_relay"
),
)
identity_evidence = _kubernetes_identity_evidence(
identity=identity,
labels=labels,
namespace=namespace,
target_resource=target_resource,
runtime_identity_evidence=runtime_identity_evidence,
)
if identity_evidence["runtime_uid_verified"] is True:
kubernetes_kind = str(identity_evidence["kind"])
route = _typed_route(
resolution_status="resolved",
target_kind="kubernetes_workload",
target_resource=target_resource,
namespace=namespace,
canonical_asset_id=(
f"k8s:{namespace or 'unbound'}:{kubernetes_kind}:"
f"{_asset_token(target_resource)}"
),
host="k3s",
executor="kubernetes_controlled_executor",
verifier="kubernetes_rollout_verifier",
risk_class="medium",
)
route["identity_evidence"] = identity_evidence
return route
route = _typed_route(
resolution_status="asset_identity_unresolved",
target_kind="unknown",
target_resource=target_resource,
namespace=namespace,
canonical_asset_id=None,
host=None,
executor=None,
verifier=None,
risk_class="high",
drift_work_item_id=identity["drift_work_item_id"],
stateful_level=StatefulLevel.UNRESOLVED.value,
)
if any(
_text_token(str(labels.get(key) or ""))
for key in ("workload_kind", "kubernetes_kind", "runtime_uid")
):
route["resolution_reason"] = (
"kubernetes_identity_evidence_missing_or_mismatch"
)
route["identity_evidence"] = identity_evidence
return route
def resolve_typed_incident_target(
incident: Mapping[str, Any] | None,
*,
registry: ServiceRegistryClient | None = None,
) -> dict[str, Any]:
"""Extract public incident fields and resolve one canonical typed route."""
incident = incident or {}
signals = [
signal
for signal in incident.get("signals") or []
if isinstance(signal, Mapping)
]
first_signal = signals[0] if signals else {}
labels = (
first_signal.get("labels")
if isinstance(first_signal.get("labels"), Mapping)
else {}
)
runtime_identity_evidence = (
first_signal.get("runtime_identity_evidence")
if isinstance(first_signal.get("runtime_identity_evidence"), Mapping)
else None
)
alertname = str(
incident.get("alertname")
or first_signal.get("alert_name")
or labels.get("alertname")
or ""
)
registry = registry or get_service_registry()
candidates = [
str(incident.get("target_resource") or ""),
*[
str(value)
for value in incident.get("affected_services") or []
if str(value).strip()
],
*[
str(labels.get(key) or "")
for key in (
"container_name",
"container",
"component",
"service",
"deployment",
"pod",
"target_resource",
)
],
]
target_resource = next((value for value in candidates if value.strip()), "")
for candidate in candidates:
if candidate and registry.get_service(candidate) is not None:
target_resource = candidate
break
return resolve_typed_alert_target(
alertname=alertname,
target_resource=target_resource,
namespace=str(labels.get("namespace") or ""),
labels=labels,
alert_category=str(incident.get("alert_category") or ""),
registry=registry,
runtime_identity_evidence=runtime_identity_evidence,
)
def resolve_controlled_alert_target(
*,
alertname: str,
target_resource: str,
namespace: str,
) -> dict[str, Any] | None:
"""Return the dedicated executor route for a known non-K8s target."""
text = " ".join(
(
_text_token(alertname),
_text_token(target_resource),
)
)
if not any(marker in text for marker in _RECOVERY_MARKERS):
return None
return {
"schema_version": "controlled_alert_target_route_v1",
"target_kind": "control_plane_recovery",
"target_resource": target_resource or "cold-start-gate",
"source_namespace": namespace or None,
"normalized_execution_namespace": "host_recovery",
"kubernetes_namespace_applicable": False,
"executor": "Agent99",
"route_id": "agent99_recover_after_owner_review",
"check_route": "agent99-mode Status controlledApply=false",
"controlled_apply_route": "agent99-mode Recover controlledApply=true",
"rollback_route": "agent99-mode Status controlledApply=false",
"runtime_execution_authorized": False,
"runtime_write_allowed": False,
}
def build_controlled_recovery_promotion_contract(
*,
alertname: str,
target_resource: str,
namespace: str,
incident_id: str = "",
) -> dict[str, Any] | None:
"""Build the no-false-green check/apply/verifier/learning contract."""
route = resolve_controlled_alert_target(
alertname=alertname,
target_resource=target_resource,
namespace=namespace,
)
if route is None:
return None
source_id = _asset_token(incident_id or f"{alertname}:{target_resource}")
fields = [
{
"field": "target_selector",
"status": "ready",
"source": "domain_target_normalizer",
"value": {
"target_kind": route["target_kind"],
"target_resource": route["target_resource"],
"normalized_execution_namespace": route[
"normalized_execution_namespace"
],
},
},
{
"field": "source_sensor_receipt",
"status": "pending",
"source": "alertmanager+cold_start_scorecard",
"value": "require_current_firing_or_recurrence_and_scorecard_ref",
},
{
"field": "check_mode_receipt",
"status": "pending",
"source": "Agent99 Status + read-only cold-start sensor",
"value": route["check_route"],
},
{
"field": "controlled_apply_route",
"status": "ready",
"source": "domain_target_normalizer",
"value": route["controlled_apply_route"],
},
{
"field": "dispatch_receipt",
"status": "pending",
"source": "agent99_sre_dispatch_receipt_v1",
"value": "accepted_and_inbox_triggered_required",
},
{
"field": "rollback_route",
"status": "ready",
"source": "domain_target_normalizer",
"value": route["rollback_route"],
},
{
"field": "post_apply_verifier",
"status": "pending",
"source": "Agent99 outcome + cold-start scorecard",
"value": [
"agent99_recover_outcome_terminal_readback",
"full_stack_cold_start_scorecard_rerun",
"required_hosts_harbor_k3s_routes_green",
"original_fingerprint_resolved_or_recurrence_decreased",
],
},
{
"field": "incident_closure_receipt",
"status": "pending",
"source": "incident_timeline",
"value": "terminal_only_after_independent_verifier",
},
{
"field": "telegram_receipt",
"status": "pending",
"source": "telegram_outbound_mirror",
"value": "lifecycle_result_delivery_ack_required",
},
{
"field": "km_writeback",
"status": "pending",
"source": "Hermes",
"value": f"km:cold-start-recovery:{source_id}",
},
{
"field": "playbook_trust_writeback",
"status": "pending",
"source": "OpenClaw/NemoTron",
"value": f"playbook-trust:cold-start-recovery:{source_id}",
},
]
ready_fields = [row["field"] for row in fields if row["status"] == "ready"]
blocked_fields = [row["field"] for row in fields if row["status"] != "ready"]
return {
"schema_version": "repair_candidate_promotion_contract_v1",
"status": "controlled_route_candidate_pending_receipts",
"route_id": route["route_id"],
"executor": route["executor"],
"incident_id": incident_id or None,
"target_route": route,
"ready_count": len(ready_fields),
"total_count": len(fields),
"blocked_count": len(blocked_fields),
"ready_fields": ready_fields,
"blocked_fields": blocked_fields,
"fields": fields,
"runtime_execution_authorized": False,
"runtime_write_allowed": False,
"controlled_playbook_queue": True,
"owner_review_required": False,
"owner_review_gate": "auto_waived_for_low_medium_high",
"needs_human": False,
"completion_status": "partial",
"completion_blocker": "dispatch_verifier_and_learning_receipts_missing",
"forbidden_operations": [
"kubectl_for_host_control_plane_target",
"generic_ansible_keyword_fallback",
"host_reboot",
"node_drain",
"backup_restore",
"secret_read",
"mark_resolved_without_post_verifier",
],
"controlled_automation_closure": {
"schema_version": "controlled_recovery_closure_contract_v1",
"trace_id_required": True,
"status": "pending_source_check_dispatch_verify_writeback",
"stages": [
{"stage": "sensor", "status": "pending", "writes_runtime_state": False},
{"stage": "normalize", "status": "passed", "writes_runtime_state": False},
{"stage": "check", "status": "pending", "writes_runtime_state": False},
{"stage": "controlled_apply", "status": "pending", "writes_runtime_state": False},
{"stage": "verify", "status": "pending", "writes_runtime_state": False},
{"stage": "learn_writeback", "status": "pending", "writes_runtime_state": False},
],
"required_receipts": [row["field"] for row in fields],
"automation_asset_ledger": [
{
"asset_type": "Sensor",
"asset_id": "scripts/reboot-recovery/full-stack-cold-start-check.sh",
"owner": "ColdStartMonitor",
"status": "read_only_receipt_pending",
},
{
"asset_type": "PlayBook",
"asset_id": "agent99:Recover",
"owner": "Agent99",
"status": "dispatch_receipt_pending",
},
{
"asset_type": "Verifier",
"asset_id": f"verifier:cold-start-recovery:{source_id}",
"owner": "PostExecutionVerifier",
"status": "pending",
},
{
"asset_type": "KM",
"asset_id": f"km:cold-start-recovery:{source_id}",
"owner": "Hermes",
"status": "pending_after_verifier",
},
],
},
}
def build_controlled_recovery_handoff(
*,
alertname: str,
target_resource: str,
namespace: str,
incident_id: str = "",
) -> dict[str, Any] | None:
contract = build_controlled_recovery_promotion_contract(
alertname=alertname,
target_resource=target_resource,
namespace=namespace,
incident_id=incident_id,
)
if contract is None:
return None
return {
"schema_version": "ai_decision_controlled_executor_handoff_v1",
"status": "agent99_dispatch_receipt_readback_pending",
"queued": False,
"side_effect_performed": False,
"single_writer_executor": "Agent99",
"route_id": contract["route_id"],
"active_blockers": [
"agent99_dispatch_receipt_readback_pending",
"post_apply_verifier_missing",
"km_playbook_writeback_missing",
],
"safe_next_action": "read_agent99_dispatch_then_run_independent_cold_start_verifier",
"repair_candidate_promotion_contract": contract,
"runtime_execution_authorized": False,
"runtime_write_allowed": False,
"owner_review_required": False,
"owner_review_gate": "auto_waived_for_low_medium_high",
"needs_human": False,
}