fix(agent99): bind control plane health selfcheck

apps/api/src/services/agent99_controlled_dispatch_ledger.py

@@ -303,6 +303,36 @@ def attach_agent99_dispatch_identity(
     return enriched
 
 
+def build_agent99_dispatch_scope(
+    *,
+    identity: Agent99DispatchIdentity,
+    kind: Any,
+    suggested_mode: Any,
+    target_resource: Any,
+    controlled_apply_requested: Any,
+    canonical_asset_id: Any = "",
+    typed_domain: Any = "",
+    executor: Any = "",
+    verifier: Any = "",
+) -> dict[str, Any]:
+    """Build the bounded typed scope persisted with one durable run."""
+
+    return {
+        "schema_version": "agent99_dispatch_scope_v1",
+        "kind": str(kind or "")[:120],
+        "suggested_mode": str(suggested_mode or "")[:80],
+        "target_resource": str(target_resource or "")[:240],
+        "controlled_apply_requested": bool(controlled_apply_requested is True),
+        "canonical_asset_id": str(canonical_asset_id or "")[:240],
+        "typed_domain": str(typed_domain or "")[:120],
+        "executor": str(executor or "")[:160],
+        "verifier": str(verifier or "")[:160],
+        # Never trust a transport-supplied route projection.  The identity is
+        # the same-run source of truth for route binding.
+        "route_id": identity.route_id,
+    }
+
+
 def build_agent99_dispatch_receipt_envelope(
     *,
     identity: Agent99DispatchIdentity,
@@ -337,6 +367,24 @@ def build_agent99_dispatch_receipt_envelope(
         and delivery_certainty == "not_delivered"
     )
     public_identity = identity.public_dict()
+    supplied_scope = (
+        dispatch_receipt.get("dispatch_scope")
+        if isinstance(dispatch_receipt.get("dispatch_scope"), dict)
+        else {}
+    )
+    dispatch_scope = build_agent99_dispatch_scope(
+        identity=identity,
+        kind=dispatch_receipt.get("kind"),
+        suggested_mode=dispatch_receipt.get("suggested_mode"),
+        target_resource=dispatch_receipt.get("target_resource"),
+        controlled_apply_requested=dispatch_receipt.get(
+            "controlled_apply_requested"
+        ),
+        canonical_asset_id=supplied_scope.get("canonical_asset_id"),
+        typed_domain=supplied_scope.get("typed_domain"),
+        executor=supplied_scope.get("executor"),
+        verifier=supplied_scope.get("verifier"),
+    )
     stage_identity = {
         "run_id": str(identity.run_id),
         "trace_id": identity.trace_id,
@@ -352,15 +400,7 @@ def build_agent99_dispatch_receipt_envelope(
             else "dispatch_delivery_unknown_reconcile_only"
         ),
         "identity": public_identity,
-        "dispatch_scope": {
-            "schema_version": "agent99_dispatch_scope_v1",
-            "kind": str(dispatch_receipt.get("kind") or ""),
-            "suggested_mode": str(dispatch_receipt.get("suggested_mode") or ""),
-            "target_resource": str(dispatch_receipt.get("target_resource") or ""),
-            "controlled_apply_requested": bool(
-                dispatch_receipt.get("controlled_apply_requested") is True
-            ),
-        },
+        "dispatch_scope": dispatch_scope,
         "dispatch_receipt": {
             "schema_version": str(
                 dispatch_receipt.get("schema_version")
@@ -508,13 +548,40 @@ class PostgresAgent99DispatchLedger:
         payload: dict[str, Any],
     ) -> dict[str, Any]:
         identity_payload = identity.public_dict()
+        routing = (
+            payload.get("routing")
+            if isinstance(payload.get("routing"), dict)
+            else {}
+        )
+        typed_target_route = (
+            routing.get("typedTargetRoute")
+            if isinstance(routing.get("typedTargetRoute"), dict)
+            else {}
+        )
+        awoooi = (
+            payload.get("awoooi")
+            if isinstance(payload.get("awoooi"), dict)
+            else {}
+        )
+        dispatch_scope = build_agent99_dispatch_scope(
+            identity=identity,
+            kind=payload.get("kind"),
+            suggested_mode=payload.get("suggestedMode") or "Status",
+            target_resource=awoooi.get("targetResource"),
+            controlled_apply_requested=payload.get("controlledApply"),
+            canonical_asset_id=typed_target_route.get("canonical_asset_id"),
+            typed_domain=typed_target_route.get("target_kind"),
+            executor=typed_target_route.get("executor"),
+            verifier=typed_target_route.get("verifier"),
+        )
         input_payload = {
             "schema_version": "agent99_controlled_dispatch_input_v1",
             "identity": identity_payload,
             "kind": str(payload.get("kind") or ""),
             "suggested_mode": str(payload.get("suggestedMode") or "Status"),
             "controlled_apply": bool(payload.get("controlledApply") is True),
-            "target_resource": str((payload.get("awoooi") or {}).get("targetResource") or ""),
+            "target_resource": str(awoooi.get("targetResource") or ""),
+            "dispatch_scope": dispatch_scope,
             "stores_secrets": False,
         }
         input_json = _stable_json(input_payload)
@@ -529,13 +596,7 @@ class PostgresAgent99DispatchLedger:
             "schema_version": "agent99_controlled_dispatch_receipt_v1",
             "status": "dispatch_reserved",
             "identity": identity_payload,
-            "dispatch_scope": {
-                "schema_version": "agent99_dispatch_scope_v1",
-                "kind": input_payload["kind"],
-                "suggested_mode": input_payload["suggested_mode"],
-                "target_resource": input_payload["target_resource"],
-                "controlled_apply_requested": input_payload["controlled_apply"],
-            },
+            "dispatch_scope": dispatch_scope,
             "dispatch_accepted": False,
             "controlled_apply_requested": bool(input_payload["controlled_apply"]),
             "controlled_apply_authorized": False,

apps/api/src/services/agent99_sre_bridge.py

@@ -29,7 +29,13 @@ from src.utils.timezone import now_taipei
 logger = get_logger("awoooi.agent99_sre_bridge")
 
 AGENT99_SRE_SINGLE_FLIGHT_TTL_SECONDS = 600
+AGENT99_CONTROL_PLANE_HEALTH_KIND = "agent99_control_plane_health"
+AGENT99_CONTROL_PLANE_HEALTH_MODE = "SelfCheck"
+AGENT99_CONTROL_PLANE_HEALTH_ROUTE_ID = (
+    "agent99:agent99_control_plane_health:SelfCheck"
+)
 AGENT99_DURABLE_ROUTE_KINDS = {
+    AGENT99_CONTROL_PLANE_HEALTH_KIND,
     "provider_freshness_signal",
     "public_route_502",
     "performance_pressure",
@@ -116,6 +122,12 @@ def resolve_agent99_alert_kind(
         annotations=annotations or {},
     )
 
+    if (
+        re.sub(r"[^a-z0-9]+", "", alertname.lower())
+        == "agent99serviceunhealthy"
+    ):
+        return AGENT99_CONTROL_PLANE_HEALTH_KIND
+
     # Explicit backup identities win over generic freshness words. Backup
     # failure cards intentionally mention freshness, but must stay in the
     # read-only BackupCheck lane with its stricter no-delete/no-restore policy.
@@ -204,6 +216,7 @@ def resolve_agent99_alert_kind(
 
 def _suggested_mode_for_kind(kind: str) -> str | None:
     return {
+        AGENT99_CONTROL_PLANE_HEALTH_KIND: AGENT99_CONTROL_PLANE_HEALTH_MODE,
         "provider_freshness_signal": "ProviderFreshness",
         "public_route_502": "Recover",
         "performance_pressure": "Perf",
@@ -254,6 +267,7 @@ def resolve_agent99_durable_route(
 
 def _resolution_policy_for_kind(kind: str) -> str:
     if kind in {
+        AGENT99_CONTROL_PLANE_HEALTH_KIND,
         "provider_freshness_signal",
         "public_route_502",
         "performance_pressure",
@@ -479,6 +493,13 @@ def build_agent99_sre_alert(
             "LightDM, VMware Tools, Wazuh services and recovery timer; do not reboot "
             "the host, change VM power, reset the VM, read secrets or change firewall."
         )
+    elif kind == AGENT99_CONTROL_PLANE_HEALTH_KIND:
+        instruction = (
+            "Run Agent99 SelfCheck only for canonical host_99 in the "
+            "windows_vmware domain; controlledApply=false; do not reboot, "
+            "change VM power, execute arbitrary shell, or fall back to another "
+            "host or domain."
+        )
 
     payload: dict[str, Any] = {
         "id": f"awoooi-alertmanager-{_safe_file_token(alert_id)}",
@@ -490,7 +511,7 @@ def build_agent99_sre_alert(
         "title": _safe_text(f"Alertmanager {alertname}", max_length=240),
         "message": _safe_text(message, max_length=2000),
         "labels": safe_labels,
-        "force": True,
+        "force": kind != AGENT99_CONTROL_PLANE_HEALTH_KIND,
         "controlledApply": suggested_mode
         in {
             "Recover",
@@ -552,6 +573,17 @@ def build_agent99_sre_alert(
     }
     if suggested_mode:
         payload["suggestedMode"] = suggested_mode
+    if kind == AGENT99_CONTROL_PLANE_HEALTH_KIND:
+        payload["routing"]["actionPolicy"] = {
+            "schemaVersion": "agent99_control_plane_action_policy_v1",
+            "allowedModes": [AGENT99_CONTROL_PLANE_HEALTH_MODE],
+            "controlledApplyAllowed": False,
+            "rebootAllowed": False,
+            "vmPowerChangeAllowed": False,
+            "arbitraryShellAllowed": False,
+            "crossHostFallbackAllowed": False,
+            "crossDomainFallbackAllowed": False,
+        }
     if instruction:
         payload["instruction"] = instruction
     return payload
@@ -862,6 +894,55 @@ def dispatch_agent99_sre_alert(payload: dict[str, Any]) -> str:
     return str(receipt["transport"])
 
 
+def _agent99_dispatch_scope_from_payload(
+    payload: dict[str, Any],
+) -> dict[str, Any]:
+    """Project the public-safe typed scope carried by one Agent99 run."""
+
+    routing = (
+        payload.get("routing")
+        if isinstance(payload.get("routing"), dict)
+        else {}
+    )
+    typed_target = (
+        routing.get("typedTargetRoute")
+        if isinstance(routing.get("typedTargetRoute"), dict)
+        else {}
+    )
+    awoooi = (
+        payload.get("awoooi")
+        if isinstance(payload.get("awoooi"), dict)
+        else {}
+    )
+    identity = (
+        awoooi.get("agent99DispatchIdentity")
+        if isinstance(awoooi.get("agent99DispatchIdentity"), dict)
+        else {}
+    )
+    return {
+        "schema_version": "agent99_dispatch_scope_v1",
+        "kind": _safe_text(payload.get("kind"), max_length=120),
+        "suggested_mode": _safe_text(
+            payload.get("suggestedMode"), max_length=80
+        ),
+        "target_resource": _safe_text(
+            awoooi.get("targetResource"), max_length=240
+        ),
+        "controlled_apply_requested": bool(
+            payload.get("controlledApply") is True
+        ),
+        "canonical_asset_id": _safe_text(
+            typed_target.get("canonical_asset_id"), max_length=240
+        ),
+        "typed_domain": _safe_text(
+            typed_target.get("target_kind"), max_length=120
+        ),
+        "executor": _safe_text(typed_target.get("executor"), max_length=160),
+        "verifier": _safe_text(typed_target.get("verifier"), max_length=160),
+        "route_id": _safe_text(identity.get("route_id"), max_length=240),
+    }
+
+
 def dispatch_agent99_sre_alert_with_receipt(
     payload: dict[str, Any],
 ) -> dict[str, Any]:
@@ -877,6 +958,7 @@ def dispatch_agent99_sre_alert_with_receipt(
     target_resource = _safe_text(awoooi.get("targetResource"), max_length=240)
     suggested_mode = _safe_text(payload.get("suggestedMode"), max_length=80)
     controlled_apply_requested = bool(payload.get("controlledApply") is True)
+    dispatch_scope = _agent99_dispatch_scope_from_payload(payload)
     if settings.AGENT99_SRE_ALERT_RELAY_URL:
         raw_receipt = post_agent99_sre_alert(payload) or {}
         response_payload: dict[str, Any] = {}
@@ -966,6 +1048,7 @@ def dispatch_agent99_sre_alert_with_receipt(
             "target_resource": target_resource,
             "suggested_mode": suggested_mode,
             "controlled_apply_requested": controlled_apply_requested,
+            "dispatch_scope": dispatch_scope,
             "http_status": http_status,
             "transport_accepted": transport_accepted,
             "accepted": accepted,
@@ -1001,6 +1084,7 @@ def dispatch_agent99_sre_alert_with_receipt(
                 "target_resource": target_resource,
                 "suggested_mode": suggested_mode,
                 "controlled_apply_requested": controlled_apply_requested,
+                "dispatch_scope": dispatch_scope,
                 "transport_accepted": True,
                 "accepted": False,
                 "inbox_triggered": False,
@@ -1031,6 +1115,7 @@ def dispatch_agent99_sre_alert_with_receipt(
         "target_resource": target_resource,
         "suggested_mode": suggested_mode,
         "controlled_apply_requested": controlled_apply_requested,
+        "dispatch_scope": dispatch_scope,
         "transport_accepted": False,
         "accepted": False,
         "inbox_triggered": False,
@@ -1130,6 +1215,48 @@ async def bridge_alertmanager_to_agent99(
             str(route_id or "").strip()
             or f"agent99:{str(payload.get('kind') or 'monitoring_alert')}:{suggested_mode}"
         )
+        if payload.get("kind") == AGENT99_CONTROL_PLANE_HEALTH_KIND:
+            typed_target_route = (
+                routing.get("typedTargetRoute")
+                if isinstance(routing.get("typedTargetRoute"), dict)
+                else {}
+            )
+            exact_control_plane_scope = bool(
+                suggested_mode == AGENT99_CONTROL_PLANE_HEALTH_MODE
+                and mutating is False
+                and payload.get("force") is False
+                and resolved_route_id == AGENT99_CONTROL_PLANE_HEALTH_ROUTE_ID
+                and typed_target_route.get("resolution_status") == "resolved"
+                and typed_target_route.get("canonical_asset_id") == "host_99"
+                and typed_target_route.get("target_kind") == "windows_vmware"
+                and typed_target_route.get("executor") == "Agent99"
+                and typed_target_route.get("verifier")
+                == "agent99_control_plane_selfcheck_independent_verifier"
+                and typed_target_route.get("cross_domain_fallback_allowed")
+                is False
+            )
+            if not exact_control_plane_scope:
+                logger.warning(
+                    "agent99_control_plane_scope_denied_fail_closed",
+                    alert_id=alert_id,
+                    alertname=alertname,
+                    route_id=resolved_route_id,
+                    canonical_asset_id=typed_target_route.get(
+                        "canonical_asset_id"
+                    ),
+                    target_kind=typed_target_route.get("target_kind"),
+                    reason="agent99_control_plane_exact_scope_required",
+                )
+                return {
+                    "status": "failed",
+                    "reason": "agent99_control_plane_exact_scope_required",
+                    "dispatchPerformed": False,
+                    "runtimeClosureVerified": False,
+                    "typedTargetRoute": typed_target_route,
+                    "driftWorkItemId": typed_target_route.get(
+                        "drift_work_item_id"
+                    ),
+                }
         durable_route = bool(
             str(incident_id or "").strip()
             or str(route_id or "").strip()
@@ -1453,7 +1580,19 @@ async def bridge_alertmanager_to_agent99(
                     "status": "transport_exception",
                     "transport": "unknown",
                     "alert_id": str(dispatch_identity.run_id),
-                    "kind": "host_recovery",
+                    "kind": str(payload.get("kind") or "monitoring_alert"),
+                    "suggested_mode": str(
+                        payload.get("suggestedMode") or "Status"
+                    ),
+                    "target_resource": str(
+                        (payload.get("awoooi") or {}).get("targetResource") or ""
+                    ),
+                    "controlled_apply_requested": bool(
+                        payload.get("controlledApply") is True
+                    ),
+                    "dispatch_scope": _agent99_dispatch_scope_from_payload(
+                        payload
+                    ),
                     "accepted": False,
                     "inbox_triggered": False,
                     "delivery_certainty": "unknown",

apps/api/src/services/controlled_alert_target_router.py

@@ -51,6 +51,7 @@ _WINDOWS_VMWARE_MARKERS = (
     "vmware",
     "vmx",
 )
+_AGENT99_CONTROL_PLANE_HEALTH_ALERTS = {"agent99serviceunhealthy"}
 _DISK_ALERTS = {
     "hostoutofdiskspace",
     "hostdiskusagehigh",
@@ -489,6 +490,53 @@ def resolve_typed_alert_target(
             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="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 {