diff --git a/apps/api/src/api/v1/agents.py b/apps/api/src/api/v1/agents.py
index 0a08b9f92..3eb5c9733 100644
--- a/apps/api/src/api/v1/agents.py
+++ b/apps/api/src/api/v1/agents.py
@@ -505,6 +505,9 @@ from src.services.service_health_failure_notification_policy import (
 from src.services.service_health_gap_matrix import (
     load_latest_service_health_gap_matrix,
 )
+from src.services.sre_k3s_controlled_automation_work_items import (
+    load_sre_k3s_controlled_automation_work_items,
+)
 from src.services.stockplatform_public_api_controlled_recovery_preflight import (
     load_latest_stockplatform_public_api_controlled_recovery_preflight,
 )
@@ -1193,6 +1196,41 @@ async def get_agent99_enterprise_ai_automation_work_items() -> dict[str, Any]:
         ) from exc
 
 
+@router.get(
+    "/sre-k3s-controlled-automation-work-items",
+    response_model=dict[str, Any],
+    summary="取得 SRE/K3s AI controlled automation 架構與工作總帳",
+    description=(
+        "讀取固定的 Alert→Canonical Asset→Typed Router→Investigator→RCA/Critic→"
+        "Policy→Executor→Verifier→Closure/Learning 架構、Agent99 Host Operations "
+        "Bridge、provider/cost gate、domain routes 與完整工作項。此端點只讀 committed "
+        "snapshot，不呼叫模型、不執行修復、不切換 provider、不讀 secret、不觸發 CD。"
+    ),
+)
+async def get_sre_k3s_controlled_automation_work_items() -> dict[str, Any]:
+    """回傳 SRE/K3s AI controlled automation 唯一工作總帳。"""
+
+    try:
+        payload = await asyncio.to_thread(
+            load_sre_k3s_controlled_automation_work_items
+        )
+        return redact_public_lan_topology(payload)
+    except FileNotFoundError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=str(exc),
+        ) from exc
+    except (json.JSONDecodeError, ValueError) as exc:
+        logger.error(
+            "sre_k3s_controlled_automation_work_items_invalid",
+            error=str(exc),
+        )
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="SRE/K3s AI controlled automation 工作總帳無效",
+        ) from exc
+
+
 @router.post(
     "/agent99/telegram-lifecycle",
     response_model=dict[str, Any],
diff --git a/apps/api/src/services/agent99_sre_bridge.py b/apps/api/src/services/agent99_sre_bridge.py
index f94f19c6a..143b77b97 100644
--- a/apps/api/src/services/agent99_sre_bridge.py
+++ b/apps/api/src/services/agent99_sre_bridge.py
@@ -23,6 +23,7 @@ from src.services.agent99_public_receipts import (
     extract_agent99_outcome_evidence_refs,
     normalize_agent99_public_receipt_ref,
 )
+from src.services.controlled_alert_target_router import resolve_typed_alert_target
 from src.utils.timezone import now_taipei
 
 logger = get_logger("awoooi.agent99_sre_bridge")
@@ -406,6 +407,13 @@ def build_agent99_sre_alert(
     labels = labels or {}
     annotations = annotations or {}
     normalized_severity = (severity or "warning").lower()
+    typed_target_route = resolve_typed_alert_target(
+        alertname=alertname,
+        target_resource=target_resource,
+        namespace=namespace,
+        labels=labels,
+        alert_category=alert_category,
+    )
     kind = resolve_agent99_alert_kind(
         alertname=alertname,
         severity=normalized_severity,
@@ -502,6 +510,14 @@ def build_agent99_sre_alert(
             "correlationKey": route_group_key,
             "sourceFingerprint": _safe_text(fingerprint, max_length=160),
             "singleFlightWindowSeconds": AGENT99_SRE_SINGLE_FLIGHT_TTL_SECONDS,
+            "typedTargetRoute": typed_target_route,
+            "typedDispatchAllowed": bool(
+                (typed_target_route.get("agent99_bridge") or {}).get(
+                    "dispatch_allowed"
+                )
+                is True
+            ),
+            "crossDomainFallbackAllowed": False,
         },
         "resolution": {
             "schemaVersion": "agent99_source_event_resolution_v1",
@@ -517,6 +533,21 @@ def build_agent99_sre_alert(
             "alertCategory": _safe_text(alert_category, max_length=160),
             "notificationType": _safe_text(notification_type, max_length=160),
             "sourceUrl": _safe_text(source_url or "", max_length=600),
+            "canonicalAssetId": _safe_text(
+                typed_target_route.get("canonical_asset_id"),
+                max_length=240,
+            ),
+            "typedDomain": _safe_text(
+                typed_target_route.get("target_kind"),
+                max_length=120,
+            ),
+            "assetResolutionStatus": _safe_text(
+                typed_target_route.get("resolution_status"),
+                max_length=120,
+            ),
+            "agent99HostOperationsBridge": typed_target_route.get(
+                "agent99_bridge"
+            ),
         },
     }
     if suggested_mode:
@@ -1063,6 +1094,38 @@ async def bridge_alertmanager_to_agent99(
         )
         suggested_mode = str(payload.get("suggestedMode") or "Status")
         mutating = bool(payload.get("controlledApply") is True)
+        typed_dispatch_allowed = bool(
+            routing.get("typedDispatchAllowed") is True
+        )
+        if mutating and not typed_dispatch_allowed:
+            typed_target_route = (
+                routing.get("typedTargetRoute")
+                if isinstance(routing.get("typedTargetRoute"), dict)
+                else {}
+            )
+            logger.warning(
+                "agent99_mutating_dispatch_typed_domain_denied",
+                alert_id=alert_id,
+                alertname=alertname,
+                target_kind=typed_target_route.get("target_kind"),
+                canonical_asset_id=typed_target_route.get(
+                    "canonical_asset_id"
+                ),
+                drift_work_item_id=typed_target_route.get(
+                    "drift_work_item_id"
+                ),
+                reason="agent99_typed_dispatch_not_allowed",
+            )
+            return {
+                "status": "failed",
+                "reason": "agent99_typed_dispatch_not_allowed",
+                "dispatchPerformed": False,
+                "runtimeClosureVerified": False,
+                "typedTargetRoute": typed_target_route,
+                "driftWorkItemId": typed_target_route.get(
+                    "drift_work_item_id"
+                ),
+            }
         resolved_route_id = (
             str(route_id or "").strip()
             or f"agent99:{str(payload.get('kind') or 'monitoring_alert')}:{suggested_mode}"
diff --git a/apps/api/src/services/auto_repair_service.py b/apps/api/src/services/auto_repair_service.py
index 62c1846a3..06e3745a2 100644
--- a/apps/api/src/services/auto_repair_service.py
+++ b/apps/api/src/services/auto_repair_service.py
@@ -22,9 +22,9 @@ Phase 8: 自動化層實作
 - CRITICAL / break-glass / 不可逆資料動作仍需明確 break-glass gate
 """
 
+import re
 from collections.abc import Callable
 from dataclasses import dataclass
-import re
 from typing import Any, Protocol
 from uuid import NAMESPACE_URL, UUID, uuid5
 
@@ -44,10 +44,15 @@ from src.services.global_repair_cooldown import (
     check_global_repair_cooldown,
     record_global_repair_action,
 )
-# Sprint 5.1: Service Registry Guardrail (ADR-062)
-from src.services.service_registry import StatefulLevel, get_service_registry
 from src.services.playbook_service import IPlaybookService, get_playbook_service
 
+# Sprint 5.1: Service Registry Guardrail (ADR-062)
+from src.services.service_registry import (
+    ServiceRegistryClient,
+    StatefulLevel,
+    get_service_registry,
+)
+
 logger = structlog.get_logger(__name__)
 
 
@@ -232,10 +237,12 @@ class AutoRepairService:
         self,
         playbook_service: IPlaybookService | None = None,
         cooldown_checker: Callable | None = None,
+        service_registry: ServiceRegistryClient | None = None,
     ):
         # 2026-04-01 ogt: 注入 cooldown_checker 支援測試隔離 (DI 原則)
         self._playbook_service = playbook_service or get_playbook_service()
         self._cooldown_checker = cooldown_checker or check_global_repair_cooldown
+        self._service_registry = service_registry or get_service_registry()
         # 2026-04-04 Claude Code: Phase 25 P1 — 持有 runbook_generator task 引用，防 GC 回收
         import asyncio
         self._pending_tasks: set[asyncio.Task] = set()
@@ -315,22 +322,37 @@ class AutoRepairService:
         # 全域熔斷之後、嚴重度之前，BLOCK 等級直接拒絕
         # 保守原則：Registry 讀取失敗也 block（優先安全，不放行）
         try:
-            _registry = get_service_registry()
+            _registry = self._service_registry
             _service_name = (incident.target_resource or "") if hasattr(incident, "target_resource") else ""
             if not _service_name and incident.affected_services:
                 _service_name = incident.affected_services[0]
             _stateful_level = _registry.get_stateful_level(_service_name)
-            if _stateful_level == StatefulLevel.BLOCK:
+            if _stateful_level in {
+                StatefulLevel.BLOCK,
+                StatefulLevel.UNRESOLVED,
+            }:
+                _identity = _registry.resolve_identity(_service_name)
+                _is_unresolved = _stateful_level == StatefulLevel.UNRESOLVED
                 logger.warning(
                     "auto_repair_blocked_guardrail",
                     incident_id=incident.incident_id,
                     service_name=_service_name,
-                    stateful_level="BLOCK",
+                    stateful_level=_stateful_level.value,
+                    drift_work_item_id=_identity.get("drift_work_item_id"),
                 )
                 return AutoRepairDecision(
                     can_auto_repair=False,
-                    reason=f"GUARDRAIL_BLOCK: 服務 '{_service_name}' 屬於禁止自動修復清單（資料安全，見 service-registry.yaml）",
-                    blocked_by="SERVICE_REGISTRY_BLOCK",
+                    reason=(
+                        f"ASSET_IDENTITY_UNRESOLVED: '{_service_name}' 未命中 canonical registry；"
+                        f"已建立 {_identity.get('drift_work_item_id')}，禁止 fallback AUTO"
+                        if _is_unresolved
+                        else f"GUARDRAIL_BLOCK: 服務 '{_service_name}' 屬於禁止自動修復清單（資料安全，見 service-registry.yaml）"
+                    ),
+                    blocked_by=(
+                        "ASSET_IDENTITY_UNRESOLVED"
+                        if _is_unresolved
+                        else "SERVICE_REGISTRY_BLOCK"
+                    ),
                 )
         except Exception as _guardrail_err:
             # S1-3 修正: Registry 失敗時保守拒絕，不允許穿透（ADR-062 審查修正 2026-04-08）
@@ -574,7 +596,9 @@ class AutoRepairService:
 
             # 2026-04-08 Claude Code: 統帥指令「所有操作都必須被記錄，寫入資料庫」
             try:
-                from src.repositories.audit_log_repository import get_auto_repair_execution_repository
+                from src.repositories.audit_log_repository import (
+                    get_auto_repair_execution_repository,
+                )
                 max_risk = self._get_max_risk_level(playbook)
                 await get_auto_repair_execution_repository().create(
                     incident_id=incident.incident_id,
@@ -626,8 +650,11 @@ class AutoRepairService:
             # verifier 有 10s warmup + 30s timeout，不能阻塞在主路徑
             try:
                 import asyncio as _asyncio
-                from src.services.post_execution_verifier import get_post_execution_verifier
+
                 from src.services.learning_service import get_learning_service
+                from src.services.post_execution_verifier import (
+                    get_post_execution_verifier,
+                )
 
                 _action_taken = _build_verification_action_taken(
                     playbook.playbook_id,
@@ -667,9 +694,13 @@ class AutoRepairService:
                                 )
                                 return
                             try:
-                                from src.services.rollback_manager import get_rollback_manager
-                                from src.services.declarative_remediation import DeclarativeRemediation
                                 from src.core.metrics import ROLLBACK_EXECUTED_TOTAL
+                                from src.services.declarative_remediation import (
+                                    DeclarativeRemediation,
+                                )
+                                from src.services.rollback_manager import (
+                                    get_rollback_manager,
+                                )
 
                                 # 從 Incident 推導 target / namespace / action
                                 _rb_target = (incident.affected_services or ["unknown"])[0]
@@ -758,7 +789,9 @@ class AutoRepairService:
 
             # 2026-04-08 Claude Code: 失敗也必須寫入 DB
             try:
-                from src.repositories.audit_log_repository import get_auto_repair_execution_repository
+                from src.repositories.audit_log_repository import (
+                    get_auto_repair_execution_repository,
+                )
                 max_risk = self._get_max_risk_level(playbook)
                 await get_auto_repair_execution_repository().create(
                     incident_id=incident.incident_id,
@@ -780,8 +813,9 @@ class AutoRepairService:
             # 2026-04-04 Claude Code: Phase 25 P1 — 失敗修復後 fire-and-forget 生成 ANTI_PATTERN
             # 2026-04-05 Claude Code: I1 修正 — 補齊 _pending_tasks GC 防護（對稱化）
             try:
-                from src.services.runbook_generator import get_runbook_generator
                 import asyncio as _asyncio
+
+                from src.services.runbook_generator import get_runbook_generator
                 symptoms = self._extract_symptoms(incident)
                 symptoms_hash = symptoms.compute_hash()
                 gen = get_runbook_generator()
@@ -1359,7 +1393,11 @@ class AutoRepairService:
 
         try:
             from src.db.base import get_db_context
-            from src.plugins.mcp.gateway import GatewayContext, McpGateway, McpGatewayError
+            from src.plugins.mcp.gateway import (
+                GatewayContext,
+                McpGateway,
+                McpGatewayError,
+            )
             from src.services.mcp_audit_context import with_mcp_audit_context
 
             incident_id = incident.incident_id
diff --git a/apps/api/src/services/awooop_ansible_audit_service.py b/apps/api/src/services/awooop_ansible_audit_service.py
index 7155ba6e1..1914d71e8 100644
--- a/apps/api/src/services/awooop_ansible_audit_service.py
+++ b/apps/api/src/services/awooop_ansible_audit_service.py
@@ -19,6 +19,7 @@ from sqlalchemy import text
 from src.db.base import get_db_context
 from src.services.controlled_alert_target_router import (
     resolve_controlled_alert_target,
+    resolve_typed_incident_target,
 )
 
 logger = structlog.get_logger(__name__)
@@ -524,6 +525,27 @@ _CATALOG: tuple[dict[str, Any], ...] = (
         "risk_level": "low",
         "catalog_revision": "2026-07-14-node-exporter-bounded-v5",
     },
+    {
+        "catalog_id": "ansible:110-sentry-profiling-consumer-recovery",
+        "playbook_path": (
+            "infra/ansible/playbooks/110-sentry-profiling-consumer-recovery.yml"
+        ),
+        "check_mode_playbook_path": (
+            "infra/ansible/playbooks/110-sentry-profiling-consumer-recovery.yml"
+        ),
+        "inventory_hosts": ["host_110"],
+        "domains": ["docker_container", "sentry", "host_110"],
+        "keywords": [
+            "dockercontainerunhealthy",
+            "sentry-self-hosted-snuba-profiling-functions-consumer-1",
+        ],
+        "supports_check_mode": True,
+        "auto_apply_enabled": True,
+        "approval_required": False,
+        "risk_level": "medium",
+        "canonical_asset_id": "service:sentry",
+        "catalog_revision": "2026-07-15-sentry-exact-container-v1",
+    },
     {
         "catalog_id": "ansible:110-devops-full-convergence",
         "playbook_path": "infra/ansible/playbooks/110-devops.yml",
@@ -970,6 +992,103 @@ def _domain_controlled_executor_route(
 
 
 def _catalog_hints(incident: dict[str, Any] | None, drift: dict[str, Any] | None) -> dict[str, Any]:
+    typed_route = resolve_typed_incident_target(incident)
+    allowed_catalog_ids = {
+        str(value)
+        for value in typed_route.get("allowed_catalog_ids") or []
+        if str(value)
+    }
+    candidates: list[dict[str, Any]] = []
+    unmatched: list[str] = []
+    for item in _CATALOG:
+        catalog_id = str(item["catalog_id"])
+        if catalog_id not in allowed_catalog_ids:
+            unmatched.append(catalog_id)
+            continue
+        public_item = {
+            key: value
+            for key, value in item.items()
+            if key
+            in {
+                "catalog_id",
+                "playbook_path",
+                "inventory_hosts",
+                "domains",
+                "supports_check_mode",
+                "check_mode_playbook_path",
+                "auto_apply_enabled",
+                "approval_required",
+                "risk_level",
+                "no_write_only",
+                "catalog_revision",
+            }
+        }
+        allowed_hosts = {
+            str(value)
+            for value in typed_route.get("allowed_inventory_hosts") or []
+            if str(value)
+        }
+        catalog_hosts = {
+            str(value)
+            for value in item.get("inventory_hosts") or []
+            if str(value)
+        }
+        if not catalog_hosts or catalog_hosts != allowed_hosts:
+            unmatched.append(catalog_id)
+            continue
+        candidates.append(
+            {
+                **public_item,
+                "match_score": 100,
+                "matched_keywords": ["typed_domain_exact_match"],
+                "semantic_match": True,
+                "matched_activation_keywords": [],
+                "matched_activation_keyword_groups": [],
+                "canonical_asset_id": typed_route.get("canonical_asset_id"),
+                "typed_domain": typed_route.get("target_kind"),
+                "target_scope_validated": True,
+            }
+        )
+
+    candidates.sort(key=lambda row: str(row["catalog_id"]))
+    resolution_status = str(typed_route.get("resolution_status") or "")
+    if resolution_status == "asset_identity_unresolved":
+        decision_effect = "create_asset_drift_work_item"
+        not_used_reason = "asset_identity_unresolved_no_fallback"
+    elif typed_route.get("executor") != "host_ansible_executor":
+        decision_effect = "delegate_to_domain_executor"
+        not_used_reason = "typed_non_ansible_domain_route"
+    elif not candidates:
+        decision_effect = "create_domain_playbook_work_item"
+        not_used_reason = "typed_domain_has_no_exact_bounded_playbook"
+    else:
+        decision_effect = "bounded_domain_candidate"
+        not_used_reason = None
+
+    result = {
+        "match_mode": "typed_domain_router_v2",
+        "decision_effect": decision_effect,
+        "available_count": len(_CATALOG),
+        "candidates": candidates,
+        "unmatched_catalog_ids": sorted(set(unmatched)),
+        "typed_target_route": typed_route,
+        "cross_domain_fallback_allowed": False,
+    }
+    if not_used_reason:
+        result["not_used_reason"] = not_used_reason
+    if decision_effect == "delegate_to_domain_executor":
+        result["controlled_executor"] = typed_route
+    if resolution_status == "asset_identity_unresolved":
+        result["drift_work_item_id"] = typed_route.get("drift_work_item_id")
+    return result
+
+
+def _legacy_catalog_hints(
+    incident: dict[str, Any] | None,
+    drift: dict[str, Any] | None,
+) -> dict[str, Any]:
+    """Retained for offline comparison only; runtime routing uses v2 above."""
+
     controlled_route = _domain_controlled_executor_route(incident)
     if controlled_route is not None:
         return {
@@ -1155,6 +1274,7 @@ def build_ansible_decision_audit_payload(
         execution_priority = 100
     execution_priority = max(0, min(execution_priority, 100))
     idempotency_key = f"ansible-candidate:{project_id}:{incident_id}"
+    typed_target_route = hints["typed_target_route"]
     input_payload = {
         "incident_id": incident_id,
         "project_id": project_id,
@@ -1178,8 +1298,10 @@ def build_ansible_decision_audit_payload(
         .lower(),
         "check_mode": True,
         "apply_enabled": False,
-        "approval_required": True,
+        "approval_required": proposal_risk_level == "critical",
         "candidate_catalog_schema": hints["match_mode"],
+        "typed_target_route": typed_target_route,
+        "cross_domain_fallback_allowed": False,
         "executor_candidates": [
             {
                 "catalog_id": row["catalog_id"],
@@ -1190,6 +1312,10 @@ def build_ansible_decision_audit_payload(
                 "auto_apply_enabled": row["auto_apply_enabled"],
                 "approval_required": row["approval_required"],
                 "risk_level": row["risk_level"],
+                "catalog_revision": row.get("catalog_revision"),
+                "canonical_asset_id": row.get("canonical_asset_id"),
+                "typed_domain": row.get("typed_domain"),
+                "target_scope_validated": row.get("target_scope_validated"),
                 "match_score": row["match_score"],
                 "matched_keywords": row["matched_keywords"],
                 "semantic_match": row["semantic_match"],
@@ -1208,9 +1334,12 @@ def build_ansible_decision_audit_payload(
             or ""
         )[:240],
         "target_selector": {
-            "schema_version": "ai_decision_target_selector_v1",
+            "schema_version": "ai_decision_target_selector_v2",
             "project_id": project_id,
             "incident_id": incident_id,
+            "canonical_asset_id": typed_target_route.get("canonical_asset_id"),
+            "typed_domain": typed_target_route.get("target_kind"),
+            "resolution_status": typed_target_route.get("resolution_status"),
             "affected_services": affected_services,
             "namespaces": namespaces,
             "catalog_ids": [str(row["catalog_id"]) for row in candidates[:5]],
@@ -1219,6 +1348,7 @@ def build_ansible_decision_audit_payload(
                 for row in candidates[:5]
                 for host in row.get("inventory_hosts") or []
             }),
+            "cross_domain_fallback_allowed": False,
         },
         "source_truth_diff": {
             "required_before_apply": True,
diff --git a/apps/api/src/services/awooop_ansible_check_mode_service.py b/apps/api/src/services/awooop_ansible_check_mode_service.py
index 8439a5716..271f598d0 100644
--- a/apps/api/src/services/awooop_ansible_check_mode_service.py
+++ b/apps/api/src/services/awooop_ansible_check_mode_service.py
@@ -222,6 +222,21 @@ def _safe_candidate(input_payload: dict[str, Any]) -> dict[str, Any]:
     if not isinstance(candidates, list) or not candidates:
         raise ValueError("missing_executor_candidates")
 
+    typed_schema = (
+        str(input_payload.get("candidate_catalog_schema") or "")
+        == "typed_domain_router_v2"
+    )
+    typed_route = input_payload.get("typed_target_route")
+    if typed_schema and not isinstance(typed_route, dict):
+        raise ValueError("typed_target_route_missing")
+    if typed_schema:
+        if typed_route.get("resolution_status") != "resolved":
+            raise ValueError("typed_target_route_unresolved")
+        if typed_route.get("executor") != "host_ansible_executor":
+            raise ValueError("typed_target_route_wrong_executor")
+        if typed_route.get("cross_domain_fallback_allowed") is not False:
+            raise ValueError("cross_domain_fallback_not_explicitly_blocked")
+
     for candidate in candidates:
         if not isinstance(candidate, dict):
             continue
@@ -241,6 +256,30 @@ def _safe_candidate(input_payload: dict[str, Any]) -> dict[str, Any]:
         if candidate_playbook_path not in {catalog_playbook_path, check_mode_playbook_path}:
             continue
         inventory_hosts = candidate.get("inventory_hosts") or catalog_item.get("inventory_hosts") or []
+        if typed_schema:
+            allowed_catalog_ids = {
+                str(value)
+                for value in typed_route.get("allowed_catalog_ids") or []
+                if str(value)
+            }
+            allowed_inventory_hosts = {
+                str(value)
+                for value in typed_route.get("allowed_inventory_hosts") or []
+                if str(value)
+            }
+            if catalog_id not in allowed_catalog_ids:
+                continue
+            if set(inventory_hosts) != allowed_inventory_hosts:
+                continue
+            if (
+                candidate.get("canonical_asset_id")
+                != typed_route.get("canonical_asset_id")
+            ):
+                continue
+            if candidate.get("typed_domain") != typed_route.get("target_kind"):
+                continue
+            if candidate.get("target_scope_validated") is not True:
+                continue
         if (
             isinstance(inventory_hosts, list)
             and inventory_hosts
@@ -262,6 +301,17 @@ def _safe_candidate(input_payload: dict[str, Any]) -> dict[str, Any]:
                 "auto_apply_enabled": bool(catalog_item.get("auto_apply_enabled") is True),
                 "approval_required": bool(catalog_item.get("approval_required") is True),
                 "break_glass_required": bool(catalog_item.get("break_glass_required") is True),
+                "canonical_asset_id": (
+                    typed_route.get("canonical_asset_id")
+                    if typed_schema
+                    else candidate.get("canonical_asset_id")
+                ),
+                "typed_domain": (
+                    typed_route.get("target_kind")
+                    if typed_schema
+                    else candidate.get("typed_domain")
+                ),
+                "target_scope_validated": typed_schema,
             }
     raise ValueError("no_safe_check_mode_candidate")
 
@@ -549,6 +599,8 @@ def build_ansible_check_mode_claim_input(
         .strip()
         .lower(),
         "target_selector": candidate_input.get("target_selector") or {},
+        "typed_target_route": candidate_input.get("typed_target_route") or {},
+        "cross_domain_fallback_allowed": False,
         "source_truth_diff": candidate_input.get("source_truth_diff") or {},
         "risk_policy_decision": candidate_input.get("risk_policy_decision") or {},
         "playbook_trust_gate": candidate_input.get("playbook_trust_gate") or {},
@@ -562,6 +614,9 @@ def build_ansible_check_mode_claim_input(
         "check_mode_playbook_path": safe["check_mode_playbook_path"],
         "inventory_hosts": list(safe["inventory_hosts"]),
         "risk_level": safe["risk_level"],
+        "canonical_asset_id": safe["canonical_asset_id"],
+        "typed_domain": safe["typed_domain"],
+        "target_scope_validated": safe["target_scope_validated"],
     }
 
 
diff --git a/apps/api/src/services/awooop_ansible_post_verifier.py b/apps/api/src/services/awooop_ansible_post_verifier.py
index c822f6e16..c3a21d668 100644
--- a/apps/api/src/services/awooop_ansible_post_verifier.py
+++ b/apps/api/src/services/awooop_ansible_post_verifier.py
@@ -153,6 +153,24 @@ _POSTCONDITION_REGISTRY: dict[str, tuple[AssetPostcondition, ...]] = {
             "sleep 5; done; exit 1",
         ),
     ),
+    "ansible:110-sentry-profiling-consumer-recovery": (
+        AssetPostcondition(
+            "host_110_sentry_profiling_consumer_running",
+            "service",
+            "host_110",
+            "test \"$(docker inspect --format '{{.State.Running}}' "
+            "sentry-self-hosted-snuba-profiling-functions-consumer-1 "
+            "2>/dev/null)\" = true",
+        ),
+        AssetPostcondition(
+            "host_110_sentry_profiling_consumer_healthy",
+            "service",
+            "host_110",
+            "test \"$(docker inspect --format '{{.State.Health.Status}}' "
+            "sentry-self-hosted-snuba-profiling-functions-consumer-1 "
+            "2>/dev/null)\" = healthy",
+        ),
+    ),
     "ansible:110-devops-full-convergence": (
         AssetPostcondition(
             "host_110_docker_runtime",
diff --git a/apps/api/src/services/controlled_alert_target_router.py b/apps/api/src/services/controlled_alert_target_router.py
index 0b2265cbf..ed2f1e9ec 100644
--- a/apps/api/src/services/controlled_alert_target_router.py
+++ b/apps/api/src/services/controlled_alert_target_router.py
@@ -14,8 +14,15 @@ runtime state.
 from __future__ import annotations
 
 import re
+from collections.abc import Mapping
 from typing import Any
 
+from src.services.service_registry import (
+    ServiceRegistryClient,
+    StatefulLevel,
+    get_service_registry,
+)
+
 _RECOVERY_MARKERS = (
     "cold-start",
     "cold_start",
@@ -28,6 +35,49 @@ _RECOVERY_MARKERS = (
     "reboot auto recovery",
 )
 
+_BACKUP_RESTORE_MARKERS = (
+    "backup",
+    "restore",
+    "escrow",
+    "offsite",
+    "retention",
+)
+_WINDOWS_VMWARE_MARKERS = (
+    "agent99",
+    "windows99",
+    "windows 99",
+    "vmware",
+    "vmx",
+)
+_DISK_ALERTS = {
+    "hostoutofdiskspace",
+    "hostdiskusagehigh",
+    "hostdiskusagecritical",
+    "hostdiskwillfillin24hours",
+}
+_NODE_EXPORTER_ALERTS = {
+    "nodeexporterdown",
+    "nodeexporterscrapedown",
+    "nodeexporterunhealthy",
+}
+_KUBERNETES_KINDS = {
+    "cronjob",
+    "daemonset",
+    "deployment",
+    "job",
+    "pod",
+    "statefulset",
+}
+_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())
@@ -38,6 +88,407 @@ def _asset_token(value: str) -> str:
     return normalized[:96] or "unbound"
 
 
+def _compact_token(value: str | None) -> str:
+    return re.sub(r"[^a-z0-9]+", "", _text_token(value))
+
+
+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,
+    execution_role: str = "policy_observer_and_no_secret_dispatch_relay",
+) -> dict[str, Any]:
+    resolved = resolution_status == "resolved"
+    agent99_dispatch = resolved and target_kind in {
+        "control_plane_recovery",
+        "windows_vmware",
+    }
+    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,
+        "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": resolved and risk_class != "critical",
+        "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": ["ollama_gcp_a", "ollama_gcp_b", "ollama_local_111"],
+            "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,
+) -> 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 any(marker in text for marker in _BACKUP_RESTORE_MARKERS):
+        identity = registry.resolve_identity(target_resource)
+        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="backup_restore_break_glass",
+            verifier="backup_restore_readback_verifier",
+            risk_class="critical",
+            stateful_level=identity.get("stateful_level"),
+        )
+
+    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,
+        ) or _HOSTS["99"]
+        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",
+        )
+
+    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"])
+        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"
+        )
+        if not exact_sentry_consumer:
+            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"
+        )
+        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=str(identity.get("executor") or "") or 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,
+        )
+
+    kubernetes_kind = _text_token(
+        str(labels.get("workload_kind") or labels.get("kubernetes_kind") or "")
+    )
+    if kubernetes_kind in _KUBERNETES_KINDS:
+        return _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",
+        )
+
+    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,
+    )
+
+
+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 {}
+    )
+    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,
+    )
+
+
 def resolve_controlled_alert_target(
     *,
     alertname: str,
diff --git a/apps/api/src/services/service_registry.py b/apps/api/src/services/service_registry.py
index fdd1b5189..8c3c56a6c 100644
--- a/apps/api/src/services/service_registry.py
+++ b/apps/api/src/services/service_registry.py
@@ -7,11 +7,12 @@
 
 from __future__ import annotations
 
-import structlog
+import hashlib
 from enum import Enum
 from pathlib import Path
 from typing import Any
 
+import structlog
 import yaml
 
 logger = structlog.get_logger(__name__)
@@ -21,6 +22,7 @@ logger = structlog.get_logger(__name__)
 # 功能降級: 找不到 YAML 時 get_service_registry() 回傳空 registry
 _DEFAULT_REGISTRY_PATH: Path | None = None
 
+
 def _find_registry_path() -> Path | None:
     """安全搜尋 service-registry.yaml，找不到回傳 None"""
     current = Path(__file__).resolve().parent
@@ -36,6 +38,7 @@ def _find_registry_path() -> Path | None:
 
 
 class StatefulLevel(str, Enum):
+    UNRESOLVED = "UNRESOLVED"       # 資產身分未解析，禁止任何 fallback apply
     BLOCK = "BLOCK"                 # 禁止，僅告警
     CRITICAL_HITL = "CRITICAL_HITL"  # 2 票 MultiSig
     STANDARD_HITL = "STANDARD_HITL"  # 1 票
@@ -45,21 +48,49 @@ class StatefulLevel(str, Enum):
 class ServiceInfo:
     def __init__(self, data: dict[str, Any]) -> None:
         self.name: str = data["name"]
+        self.canonical_id: str = data.get("canonical_id", f"service:{self.name}")
         self.display_name: str = data.get("display_name", self.name)
         self.host: str = data.get("host", "unknown")
         self.stateful_level: StatefulLevel = StatefulLevel(data.get("stateful_level", "AUTO"))
+        self.asset_domain: str = data.get(
+            "asset_domain",
+            "kubernetes_workload" if self.host == "k3s" else "docker_container",
+        )
+        self.executor: str = data.get(
+            "executor",
+            {
+                "kubernetes_workload": "kubernetes_controlled_executor",
+                "database": "db_bounded_executor",
+                "backup_restore": "backup_restore_break_glass",
+                "storage": "storage_bounded_executor",
+            }.get(self.asset_domain, "host_ansible_executor"),
+        )
+        self.verifier: str = data.get(
+            "verifier",
+            {
+                "kubernetes_workload": "kubernetes_rollout_verifier",
+                "database": "db_independent_verifier",
+                "backup_restore": "backup_restore_readback_verifier",
+                "storage": "storage_independent_verifier",
+            }.get(self.asset_domain, "container_runtime_independent_verifier"),
+        )
         self.reason: str = data.get("reason", "")
         self.alert_only: bool = data.get("alert_only", False)
         self.requires_pre_backup: bool = data.get("requires_pre_backup", False)
-        self.restart_command: str = data.get("restart_command", "docker restart")
+        self.restart_command: str = data.get("restart_command", "")
         self.containers: list[str] = data.get("containers", [])
+        self.aliases: list[str] = data.get("aliases", [])
+        self.allowed_catalog_ids: list[str] = data.get("allowed_catalog_ids", [])
 
 
 class ServiceRegistryClient:
     """
     Service Registry 客戶端
     讀取 ops/config/service-registry.yaml，提供服務 Stateful 分級查詢
-    設計原則: 純讀取，不寫入；失敗時 fallback AUTO（防護不應阻擋告警流程）
+    設計原則: 純讀取，不寫入；失敗或未知資產一律 fail closed。
+
+    Registry 缺口不應阻擋告警收件，但必須阻擋執行並建立 deterministic
+    drift work item。不得把未知資產猜成 AUTO，也不得提供泛用 restart 命令。
     """
 
     def __init__(self, registry_path: Path | None = None) -> None:
@@ -67,13 +98,29 @@ class ServiceRegistryClient:
         self._services: dict[str, ServiceInfo] = {}
         self._backup_policies: dict[str, Any] = {}
         self._multisig_config: dict[str, Any] = {}
+        self._load_error: str | None = None
         self._loaded = False
 
+    @staticmethod
+    def _identity_key(value: str) -> str:
+        return str(value or "").strip().lower()
+
+    @classmethod
+    def _drift_work_item_id(cls, value: str) -> str:
+        key = cls._identity_key(value) or "missing-target"
+        digest = hashlib.sha256(key.encode("utf-8")).hexdigest()[:12].upper()
+        return f"AIA-ASSET-DRIFT-{digest}"
+
     def _load(self) -> None:
         if self._loaded:
             return
         if self._path is None or not self._path.exists():
-            logger.warning("service_registry_not_found", path=str(self._path), fallback="all services treated as AUTO")
+            self._load_error = "service_registry_not_found"
+            logger.warning(
+                "service_registry_not_found",
+                path=str(self._path),
+                fallback="asset_identity_unresolved",
+            )
             self._loaded = True
             return
         try:
@@ -81,32 +128,89 @@ class ServiceRegistryClient:
                 data = yaml.safe_load(f)
             for svc in data.get("services", []):
                 info = ServiceInfo(svc)
-                self._services[info.name] = info
-                # 也按 container 名稱建立索引
-                for container in info.containers:
-                    self._services[container] = info
+                aliases = [info.name, *info.containers, *info.aliases]
+                for alias in aliases:
+                    key = self._identity_key(alias)
+                    existing = self._services.get(key)
+                    if existing is not None and existing.canonical_id != info.canonical_id:
+                        raise ValueError(
+                            f"duplicate service registry alias: {alias}"
+                        )
+                    self._services[key] = info
             self._backup_policies = data.get("backup_policies", {})
             self._multisig_config = data.get("multisig", {})
             self._loaded = True
             logger.info(f"Service Registry 載入完成: {len(self._services)} 個服務")
         except Exception as e:
-            logger.error(f"Service Registry 載入失敗: {e}，所有服務 fallback AUTO")
+            self._services = {}
+            self._load_error = "service_registry_load_failed"
+            logger.error(
+                "service_registry_load_failed",
+                error=str(e),
+                fallback="asset_identity_unresolved",
+            )
             self._loaded = True  # 防止重複嘗試
 
     def get_service(self, name: str) -> ServiceInfo | None:
         self._load()
-        return self._services.get(name)
+        return self._services.get(self._identity_key(name))
+
+    def resolve_identity(self, name: str) -> dict[str, Any]:
+        """Resolve an exact service/container alias without fuzzy fallback."""
+
+        info = self.get_service(name)
+        if info is None:
+            return {
+                "schema_version": "canonical_asset_identity_v2",
+                "resolution_status": "asset_identity_unresolved",
+                "input_identity": str(name or ""),
+                "normalized_identity": self._identity_key(name),
+                "canonical_id": None,
+                "asset_domain": "unknown",
+                "host": None,
+                "stateful_level": StatefulLevel.UNRESOLVED.value,
+                "executor": None,
+                "verifier": None,
+                "allowed_catalog_ids": [],
+                "controlled_apply_allowed": False,
+                "drift_work_item_id": self._drift_work_item_id(name),
+                "registry_error": self._load_error,
+            }
+        return {
+            "schema_version": "canonical_asset_identity_v2",
+            "resolution_status": "resolved",
+            "input_identity": str(name or ""),
+            "normalized_identity": self._identity_key(name),
+            "canonical_id": info.canonical_id,
+            "service_name": info.name,
+            "asset_domain": info.asset_domain,
+            "host": info.host,
+            "stateful_level": info.stateful_level.value,
+            "executor": info.executor,
+            "verifier": info.verifier,
+            "allowed_catalog_ids": list(info.allowed_catalog_ids),
+            "controlled_apply_allowed": info.stateful_level == StatefulLevel.AUTO,
+            "drift_work_item_id": None,
+            "registry_error": self._load_error,
+        }
 
     def get_stateful_level(self, service_name: str) -> StatefulLevel:
-        """查詢服務分級，未知服務 fallback AUTO"""
+        """查詢服務分級；未知服務 fail closed 為 UNRESOLVED。"""
         info = self.get_service(service_name)
         if info is None:
-            logger.warning(f"未知服務 '{service_name}'，fallback AUTO")
-            return StatefulLevel.AUTO
+            logger.warning(
+                "service_registry_identity_unresolved",
+                service_name=service_name,
+                drift_work_item_id=self._drift_work_item_id(service_name),
+            )
+            return StatefulLevel.UNRESOLVED
         return info.stateful_level
 
     def is_blocked(self, service_name: str) -> bool:
-        return self.get_stateful_level(service_name) == StatefulLevel.BLOCK
+        return self.get_stateful_level(service_name) in {
+            StatefulLevel.BLOCK,
+            StatefulLevel.UNRESOLVED,
+        }
 
     def requires_multisig(self, service_name: str) -> bool:
         return self.get_stateful_level(service_name) == StatefulLevel.CRITICAL_HITL
@@ -124,7 +228,7 @@ class ServiceRegistryClient:
 
     def get_restart_command(self, service_name: str) -> str:
         info = self.get_service(service_name)
-        return info.restart_command if info else "docker restart"
+        return info.restart_command if info else ""
 
 
 # Singleton
diff --git a/apps/api/src/services/sre_k3s_controlled_automation_work_items.py b/apps/api/src/services/sre_k3s_controlled_automation_work_items.py
new file mode 100644
index 000000000..0f5caed3f
--- /dev/null
+++ b/apps/api/src/services/sre_k3s_controlled_automation_work_items.py
@@ -0,0 +1,230 @@
+"""Canonical SRE/K3s controlled-automation architecture and work ledger."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any
+
+from src.services.snapshot_paths import default_operations_dir
+
+_DEFAULT_OPERATIONS_DIR = default_operations_dir(Path(__file__))
+_SNAPSHOT_FILE = "sre-k3s-controlled-automation-work-items.snapshot.json"
+_SCHEMA_VERSION = "sre_k3s_controlled_automation_work_items_v1"
+_PROGRAM_ID = "AIA-SRE-P0-20260715"
+_PIPELINE = [
+    "Alert",
+    "Canonical Asset Normalize",
+    "Typed Domain Router",
+    "HolmesGPT Investigator",
+    "Ollama RCA / Gemini Critic",
+    "Deterministic Policy",
+    "Single Controlled Executor",
+    "Independent Verifier",
+    "Incident Closure + KM/RAG/MCP/PlayBook",
+]
+_PROVIDER_ORDER = [
+    "ollama_gcp_a",
+    "ollama_gcp_b",
+    "ollama_local_111",
+    "gemini_api",
+]
+_REQUIRED_DOMAINS = {
+    "kubernetes_workload",
+    "host_systemd",
+    "docker_container",
+    "windows_vmware",
+    "database",
+    "backup_restore",
+    "unknown",
+}
+_VALID_PRIORITIES = {"P0", "P1", "P2"}
+_VALID_RISKS = {"low", "medium", "high", "critical"}
+_VALID_STATUSES = {
+    "source_implemented_runtime_pending",
+    "in_progress",
+    "planned",
+    "policy_active",
+    "runtime_closed",
+}
+_REQUIRED_ITEM_FIELDS = {
+    "id",
+    "order",
+    "priority",
+    "phase",
+    "title",
+    "owner_lane",
+    "risk",
+    "status",
+    "dependencies",
+    "source_refs",
+    "executor",
+    "verifier",
+    "rollback",
+    "exit_condition",
+    "next_action",
+}
+
+
+def load_sre_k3s_controlled_automation_work_items(
+    operations_dir: Path | None = None,
+) -> dict[str, Any]:
+    """Load and validate the single canonical architecture/work ledger."""
+
+    directory = operations_dir or _DEFAULT_OPERATIONS_DIR
+    path = directory / _SNAPSHOT_FILE
+    with path.open(encoding="utf-8") as handle:
+        payload = json.load(handle)
+
+    if not isinstance(payload, dict):
+        raise ValueError(f"{path}: expected JSON object")
+    if payload.get("schema_version") != _SCHEMA_VERSION:
+        raise ValueError(f"{path}: unexpected schema_version")
+    if payload.get("program_id") != _PROGRAM_ID:
+        raise ValueError(f"{path}: unexpected program_id")
+
+    architecture = _dict(payload, "architecture", path)
+    if architecture.get("pipeline") != _PIPELINE:
+        raise ValueError(f"{path}: architecture.pipeline order changed")
+    if architecture.get("cross_domain_fallback_allowed") is not False:
+        raise ValueError(f"{path}: cross-domain fallback must remain false")
+    if architecture.get("unknown_asset_terminal") != "asset_identity_unresolved":
+        raise ValueError(f"{path}: unknown asset terminal must fail closed")
+
+    provider = _dict(payload, "provider_policy", path)
+    if provider.get("ordered_route") != _PROVIDER_ORDER:
+        raise ValueError(f"{path}: provider order changed")
+    if provider.get("gemini_paid_call_allowed") is not False:
+        raise ValueError(f"{path}: Gemini paid calls require an explicit cost cap")
+    if provider.get("production_provider_route_switch_allowed") is not False:
+        raise ValueError(f"{path}: production provider switch must remain gated")
+
+    bridge = _dict(payload, "agent99_host_operations_bridge", path)
+    if bridge.get("arbitrary_command_allowed") is not False:
+        raise ValueError(f"{path}: Agent99 arbitrary commands must remain forbidden")
+    if bridge.get("unknown_asset_dispatch_allowed") is not False:
+        raise ValueError(f"{path}: Agent99 unknown-asset dispatch must remain forbidden")
+    if bridge.get("cross_domain_fallback_allowed") is not False:
+        raise ValueError(f"{path}: Agent99 cross-domain fallback must remain forbidden")
+    if bridge.get("completion_callback") != (
+        "/api/v1/agents/agent99/completion-callback"
+    ):
+        raise ValueError(f"{path}: Agent99 completion callback drift")
+
+    domain_rows = _list(payload, "domain_routes", path)
+    domains = {str(row.get("domain") or "") for row in domain_rows if isinstance(row, dict)}
+    if domains != _REQUIRED_DOMAINS:
+        raise ValueError(f"{path}: domain route coverage mismatch")
+    unknown = next(row for row in domain_rows if row.get("domain") == "unknown")
+    if unknown.get("executor") is not None or unknown.get("fallback") != "forbidden":
+        raise ValueError(f"{path}: unknown domain must have no executor/fallback")
+    backup = next(
+        row for row in domain_rows if row.get("domain") == "backup_restore"
+    )
+    if backup.get("critical_read_only_default") is not True:
+        raise ValueError(f"{path}: backup/restore must remain critical read-only")
+
+    items = _list(payload, "work_items", path)
+    ids: list[str] = []
+    priority_counts: dict[str, int] = {}
+    status_counts: dict[str, int] = {}
+    for index, item in enumerate(items, start=1):
+        if not isinstance(item, dict):
+            raise ValueError(f"{path}: work_items[{index - 1}] must be an object")
+        missing = sorted(_REQUIRED_ITEM_FIELDS - item.keys())
+        if missing:
+            raise ValueError(f"{path}: work item missing fields: {missing}")
+        item_id = str(item.get("id") or "")
+        if not item_id:
+            raise ValueError(f"{path}: work item id required")
+        if item.get("order") != index:
+            raise ValueError(f"{path}: {item_id}.order must equal {index}")
+        if item.get("priority") not in _VALID_PRIORITIES:
+            raise ValueError(f"{path}: {item_id}.priority invalid")
+        if item.get("risk") not in _VALID_RISKS:
+            raise ValueError(f"{path}: {item_id}.risk invalid")
+        if item.get("status") not in _VALID_STATUSES:
+            raise ValueError(f"{path}: {item_id}.status invalid")
+        ids.append(item_id)
+        priority = str(item["priority"])
+        item_status = str(item["status"])
+        priority_counts[priority] = priority_counts.get(priority, 0) + 1
+        status_counts[item_status] = status_counts.get(item_status, 0) + 1
+    if len(ids) != len(set(ids)):
+        raise ValueError(f"{path}: duplicate work item id")
+    known_ids = set(ids)
+    for item in items:
+        unknown_dependencies = set(item["dependencies"]) - known_ids
+        if unknown_dependencies:
+            raise ValueError(
+                f"{path}: {item['id']} has unknown dependencies: "
+                f"{sorted(unknown_dependencies)}"
+            )
+
+    current_p0 = _dict(payload, "current_p0", path)
+    if current_p0.get("id") not in known_ids:
+        raise ValueError(f"{path}: current_p0.id must be a work item")
+    current_item = next(item for item in items if item["id"] == current_p0["id"])
+    if current_item["priority"] != "P0" or current_item["status"] == "runtime_closed":
+        raise ValueError(f"{path}: current_p0 must be an active P0 item")
+
+    rollups = _dict(payload, "rollups", path)
+    if rollups.get("total_items") != len(items):
+        raise ValueError(f"{path}: rollups.total_items mismatch")
+    if rollups.get("by_priority") != priority_counts:
+        raise ValueError(f"{path}: rollups.by_priority mismatch")
+    if rollups.get("by_status") != status_counts:
+        raise ValueError(f"{path}: rollups.by_status mismatch")
+    source_implemented = sum(
+        item["status"] == "source_implemented_runtime_pending" for item in items
+    )
+    runtime_closed = sum(item["status"] == "runtime_closed" for item in items)
+    if rollups.get("source_implemented_items") != source_implemented:
+        raise ValueError(f"{path}: source implemented count mismatch")
+    if rollups.get("runtime_closed_items") != runtime_closed:
+        raise ValueError(f"{path}: runtime closed count mismatch")
+    expected_percent = round((runtime_closed / len(items)) * 100) if items else 0
+    if rollups.get("program_completion_percent") != expected_percent:
+        raise ValueError(f"{path}: program completion must use runtime closure")
+
+    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")
+    return payload
+
+
+def build_sre_k3s_program_projection(payload: dict[str, Any]) -> dict[str, Any]:
+    """Return a bounded cockpit projection without dropping truth labels."""
+
+    return {
+        "schema_version": payload["schema_version"],
+        "program_id": payload["program_id"],
+        "generated_at": payload["generated_at"],
+        "status": payload["status"],
+        "scope_complete": payload["scope_complete"],
+        "current_p0": payload["current_p0"],
+        "rollups": payload["rollups"],
+        "provider_order": payload["provider_policy"]["ordered_route"],
+        "gemini_paid_call_allowed": payload["provider_policy"][
+            "gemini_paid_call_allowed"
+        ],
+        "domain_count": len(payload["domain_routes"]),
+        "agent99_bridge": payload["agent99_host_operations_bridge"],
+        "full_readback_api": (
+            "/api/v1/agents/sre-k3s-controlled-automation-work-items"
+        ),
+    }
+
+
+def _dict(payload: dict[str, Any], key: str, path: Path) -> dict[str, Any]:
+    value = payload.get(key)
+    if not isinstance(value, dict):
+        raise ValueError(f"{path}: {key} must be an object")
+    return value
+
+
+def _list(payload: dict[str, Any], key: str, path: Path) -> list[Any]:
+    value = payload.get(key)
+    if not isinstance(value, list):
+        raise ValueError(f"{path}: {key} must be an array")
+    return value
diff --git a/apps/api/tests/test_auto_repair_service.py b/apps/api/tests/test_auto_repair_service.py
index 0812fd3fb..59c9cf2ef 100644
--- a/apps/api/tests/test_auto_repair_service.py
+++ b/apps/api/tests/test_auto_repair_service.py
@@ -21,6 +21,7 @@ from src.models.playbook import (
 )
 from src.plugins.mcp.interfaces import MCPToolResult
 from src.services.auto_repair_service import AutoRepairService
+from src.services.service_registry import StatefulLevel
 from src.utils.timezone import now_taipei
 
 
@@ -129,6 +130,20 @@ class _DbContext:
         return None
 
 
+class _ResolvedTestServiceRegistry:
+    """Keep recommendation tests focused while production unknowns fail closed."""
+
+    def get_stateful_level(self, _service_name: str) -> StatefulLevel:
+        return StatefulLevel.AUTO
+
+    def resolve_identity(self, service_name: str) -> dict:
+        return {
+            "resolution_status": "resolved",
+            "canonical_id": f"test-service:{service_name}",
+            "drift_work_item_id": None,
+        }
+
+
 class TestAutoRepairService:
     """Auto Repair Service unit tests"""
 
@@ -142,6 +157,7 @@ class TestAutoRepairService:
         return AutoRepairService(
             playbook_service=mock_playbook_service,
             cooldown_checker=_no_cooldown,
+            service_registry=_ResolvedTestServiceRegistry(),
         )
 
     @pytest.mark.asyncio
diff --git a/apps/api/tests/test_awooop_truth_chain_service.py b/apps/api/tests/test_awooop_truth_chain_service.py
index 47c7b7ba3..583fd7920 100644
--- a/apps/api/tests/test_awooop_truth_chain_service.py
+++ b/apps/api/tests/test_awooop_truth_chain_service.py
@@ -1337,13 +1337,16 @@ def test_ansible_truth_surfaces_audited_check_mode_record() -> None:
     assert truth["summary"]["latest_catalog_id"] == "ansible:188-momo-backup-user"
     assert truth["summary"]["latest_returncode"] == 0
     assert "ansible_check_mode_executed" in truth["audit_contract"]["operation_types"]
-    assert truth["candidate_catalog"]["decision_effect"] == "none"
-    assert truth["candidate_catalog"]["candidates"][0]["catalog_id"] == "ansible:188-momo-backup-user"
-    assert truth["candidate_catalog"]["candidates"][0]["auto_apply_enabled"] is True
-    assert (
-        truth["candidate_catalog"]["candidates"][0]["check_mode_playbook_path"]
-        == "infra/ansible/playbooks/188-momo-backup-user.yml"
+    assert truth["candidate_catalog"]["decision_effect"] == (
+        "delegate_to_domain_executor"
     )
+    assert truth["candidate_catalog"]["candidates"] == []
+    assert truth["candidate_catalog"]["typed_target_route"]["target_kind"] == (
+        "backup_restore"
+    )
+    assert truth["candidate_catalog"]["typed_target_route"][
+        "critical_break_glass_required"
+    ] is True
 
 
 def test_ansible_truth_marks_worker_apply_as_controlled_apply() -> None:
@@ -1397,16 +1400,17 @@ def test_ansible_truth_keeps_catalog_hint_separate_from_runtime_use() -> None:
 
     assert truth["considered"] is False
     assert truth["records"] == []
-    assert truth["not_used_reason"].startswith("no automation_operation_log row")
-    assert truth["candidate_catalog"]["candidates"][0]["catalog_id"] == "ansible:nginx-sync"
-    assert truth["candidate_catalog"]["candidates"][0]["check_mode_playbook_path"] == (
-        "infra/ansible/playbooks/nginx-sync-readonly.yml"
+    assert truth["not_used_reason"] == "asset_identity_unresolved_no_fallback"
+    assert truth["candidate_catalog"]["candidates"] == []
+    assert truth["candidate_catalog"]["decision_effect"] == (
+        "create_asset_drift_work_item"
+    )
+    assert truth["candidate_catalog"]["drift_work_item_id"].startswith(
+        "AIA-ASSET-DRIFT-"
     )
-    assert truth["candidate_catalog"]["candidates"][0]["approval_required"] is False
-    assert truth["candidate_catalog"]["decision_effect"] == "none"
 
 
-def test_ansible_decision_audit_payload_is_dry_run_only() -> None:
+def test_unknown_docker_asset_does_not_emit_generic_ansible_candidate() -> None:
     incident = SimpleNamespace(
         incident_id="INC-DOCKER",
         project_id="awoooi",
@@ -1430,17 +1434,7 @@ def test_ansible_decision_audit_payload_is_dry_run_only() -> None:
         not_used_reason="manual approval required; Ansible check-mode is not wired yet",
     )
 
-    assert payload is not None
-    assert payload["operation_type"] == "ansible_candidate_matched"
-    assert payload["status"] == "dry_run"
-    assert payload["input"]["executor"] == "ansible"
-    assert payload["input"]["check_mode"] is True
-    assert payload["input"]["apply_enabled"] is False
-    assert payload["input"]["approval_required"] is True
-    assert payload["input"]["execution_priority"] == 100
-    assert payload["input"]["executor_candidates"]
-    assert payload["output"]["decision_effect"] == "audit_only"
-    assert payload["dry_run_result"]["check_mode_executed"] is False
+    assert payload is None
 
 
 def test_ansible_decision_audit_rejects_host_number_only_catalog_match() -> None:
@@ -1475,7 +1469,7 @@ def test_ansible_decision_audit_rejects_host_number_only_catalog_match() -> None
     assert payload is None
 
 
-def test_ansible_canary_candidate_preserves_approval_and_controlled_apply() -> None:
+def test_kubernetes_canary_does_not_enter_ansible_candidate_queue() -> None:
     approval_id = "00000000-0000-0000-0000-000000000044"
     incident = SimpleNamespace(
         incident_id="INC-CONTROLLED-CANARY",
@@ -1509,21 +1503,7 @@ def test_ansible_canary_candidate_preserves_approval_and_controlled_apply() -> N
         not_used_reason="queue bounded no-traffic canary",
     )
 
-    assert payload is not None
-    assert payload["input"]["approval_id"] == approval_id
-    candidate = payload["input"]["executor_candidates"][0]
-    assert candidate["catalog_id"] == "ansible:awoooi-auto-repair-canary"
-    assert candidate["inventory_hosts"] == ["host_121"]
-    assert candidate["approval_required"] is False
-    claim = build_ansible_check_mode_claim_input(
-        source_candidate_op_id="00000000-0000-0000-0000-000000000045",
-        candidate_input=payload["input"],
-    )
-    assert claim["approval_id"] == approval_id
-    assert claim["controlled_apply_allowed"] is True
-    assert claim["apply_playbook_path"] == (
-        "infra/ansible/playbooks/awoooi-auto-repair-canary.yml"
-    )
+    assert payload is None
 
 
 @pytest.mark.asyncio
@@ -1587,7 +1567,7 @@ def test_controlled_apply_bridges_terminal_projection_and_telegram_receipt() ->
     assert "_finalize_controlled_approval_projection" in source
 
 
-def test_ansible_decision_audit_payload_exposes_check_mode_safety_flags(
+def test_backup_candidate_is_excluded_from_ansible_check_mode_queue(
     monkeypatch,
 ) -> None:
     monkeypatch.setenv("AWOOOI_BUILD_COMMIT_SHA", "source-sha-123")
@@ -1614,15 +1594,7 @@ def test_ansible_decision_audit_payload_exposes_check_mode_safety_flags(
         not_used_reason="candidate audit",
     )
 
-    candidate = payload["input"]["executor_candidates"][0]
-    assert candidate["catalog_id"] == "ansible:188-momo-backup-user"
-    assert candidate["supports_check_mode"] is True
-    assert candidate["playbook_path"] == "infra/ansible/playbooks/188-momo-backup-user.yml"
-    assert candidate["check_mode_playbook_path"] == "infra/ansible/playbooks/188-momo-backup-user.yml"
-    assert candidate["auto_apply_enabled"] is True
-    assert candidate["approval_required"] is False
-    assert candidate["risk_level"] == "low"
-    assert payload["input"]["router_source_sha"] == "source-sha-123"
+    assert payload is None
 
 
 def test_ansible_check_mode_claim_input_authorizes_controlled_apply() -> None:
@@ -3106,13 +3078,12 @@ async def test_ai_decision_handoff_queues_evidence_backed_single_writer_candidat
         evidence_verifier=verify,
     )
 
-    assert handoff["status"] == "controlled_check_mode_queued"
-    assert handoff["queued"] is True
+    assert handoff["status"] == "no_allowlisted_ansible_candidate"
+    assert handoff["queued"] is False
     assert handoff["side_effect_performed"] is False
     assert handoff["single_writer_executor"] == "awoooi-ansible-executor-broker"
-    assert handoff["source_truth_diff"]["required_before_apply"] is True
-    assert handoff["risk_policy_decision"]["risk_level"] == "low"
-    assert recorded[0]["decision_path"] == "repair_candidate_controlled_queue"
+    assert handoff["active_blockers"] == ["no_allowlisted_ansible_candidate"]
+    assert recorded == []
 
 
 @pytest.mark.asyncio
diff --git a/apps/api/tests/test_controlled_alert_target_router.py b/apps/api/tests/test_controlled_alert_target_router.py
index 9de0cd5b9..11fa81e60 100644
--- a/apps/api/tests/test_controlled_alert_target_router.py
+++ b/apps/api/tests/test_controlled_alert_target_router.py
@@ -97,7 +97,7 @@ def test_cold_start_contract_requires_same_run_receipts_before_closure() -> None
 def test_cold_start_is_excluded_from_generic_ansible_keyword_catalog() -> None:
     hints = _catalog_hints(_cold_start_incident(), None)
 
-    assert hints["match_mode"] == "domain_controlled_executor_route_v1"
+    assert hints["match_mode"] == "typed_domain_router_v2"
     assert hints["decision_effect"] == "delegate_to_domain_executor"
     assert hints["candidates"] == []
     assert hints["controlled_executor"]["executor"] == "Agent99"
@@ -105,7 +105,7 @@ def test_cold_start_is_excluded_from_generic_ansible_keyword_catalog() -> None:
     assert "ansible:188-momo-backup-user" in hints["unmatched_catalog_ids"]
 
     truth = build_ansible_truth([], incident=_cold_start_incident(), drift=None)
-    assert truth["not_used_reason"] == "non_kubernetes_control_plane_route"
+    assert truth["not_used_reason"] == "typed_non_ansible_domain_route"
     assert truth["candidate_catalog"]["controlled_executor"]["route_id"] == (
         "agent99_recover_after_owner_review"
     )
diff --git a/apps/api/tests/test_postgres_slow_query_automation_guard.py b/apps/api/tests/test_postgres_slow_query_automation_guard.py
index f44bcbd3e..713cc97a4 100644
--- a/apps/api/tests/test_postgres_slow_query_automation_guard.py
+++ b/apps/api/tests/test_postgres_slow_query_automation_guard.py
@@ -82,7 +82,7 @@ def test_slow_query_cannot_activate_momo_backup_or_ai_web_catalog() -> None:
     assert payload is None
 
 
-def test_momo_backup_failure_still_activates_bounded_backup_catalog() -> None:
+def test_momo_backup_failure_is_critical_readback_not_ansible_apply() -> None:
     incident = _incident(
         "MomoPostgresBackupFailed",
         {
@@ -104,10 +104,4 @@ def test_momo_backup_failure_still_activates_bounded_backup_catalog() -> None:
         not_used_reason="candidate backfill",
     )
 
-    assert payload is not None
-    candidate = payload["input"]["executor_candidates"][0]
-    assert candidate["catalog_id"] == "ansible:188-momo-backup-user"
-    assert candidate["semantic_match"] is True
-    assert "momopostgresbackupfailed" in candidate[
-        "matched_activation_keywords"
-    ]
+    assert payload is None
diff --git a/apps/api/tests/test_sre_k3s_controlled_automation_work_items_api.py b/apps/api/tests/test_sre_k3s_controlled_automation_work_items_api.py
new file mode 100644
index 000000000..d4345df4c
--- /dev/null
+++ b/apps/api/tests/test_sre_k3s_controlled_automation_work_items_api.py
@@ -0,0 +1,124 @@
+from __future__ import annotations
+
+# ruff: noqa: E402, I001
+
+import json
+import os
+from pathlib import Path
+
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+os.environ.setdefault("DATABASE_URL", "postgresql+asyncpg://test:test@localhost/test")
+
+from src.api.v1.agents import router
+from src.services.sre_k3s_controlled_automation_work_items import (
+    build_sre_k3s_program_projection,
+    load_sre_k3s_controlled_automation_work_items,
+)
+
+ROOT = Path(__file__).resolve().parents[3]
+OPERATIONS = ROOT / "docs" / "operations"
+SNAPSHOT = OPERATIONS / "sre-k3s-controlled-automation-work-items.snapshot.json"
+
+
+def test_loader_returns_fixed_architecture_provider_order_and_agent99_bridge() -> None:
+    payload = load_sre_k3s_controlled_automation_work_items()
+
+    assert payload["program_id"] == "AIA-SRE-P0-20260715"
+    assert payload["current_p0"]["id"] == "AIA-SRE-004"
+    assert payload["architecture"]["pipeline"] == [
+        "Alert",
+        "Canonical Asset Normalize",
+        "Typed Domain Router",
+        "HolmesGPT Investigator",
+        "Ollama RCA / Gemini Critic",
+        "Deterministic Policy",
+        "Single Controlled Executor",
+        "Independent Verifier",
+        "Incident Closure + KM/RAG/MCP/PlayBook",
+    ]
+    assert payload["provider_policy"]["ordered_route"] == [
+        "ollama_gcp_a",
+        "ollama_gcp_b",
+        "ollama_local_111",
+        "gemini_api",
+    ]
+    assert payload["provider_policy"]["gemini_paid_call_allowed"] is False
+    assert payload["agent99_host_operations_bridge"]["single_executor_domains"] == [
+        "windows_vmware",
+        "control_plane_recovery",
+    ]
+    assert payload["agent99_host_operations_bridge"][
+        "unknown_asset_dispatch_allowed"
+    ] is False
+    assert payload["rollups"] == {
+        "total_items": 18,
+        "by_priority": {"P0": 16, "P1": 2},
+        "by_status": {
+            "source_implemented_runtime_pending": 6,
+            "in_progress": 6,
+            "planned": 5,
+            "policy_active": 1,
+        },
+        "source_implemented_items": 6,
+        "runtime_closed_items": 0,
+        "program_completion_percent": 0,
+        "asset_coverage_status": "partial",
+        "runtime_closure_status": "not_started_for_this_program_sha",
+    }
+
+
+def test_projection_keeps_program_asset_and_runtime_truth_separate() -> None:
+    payload = load_sre_k3s_controlled_automation_work_items()
+    projection = build_sre_k3s_program_projection(payload)
+
+    assert projection["rollups"]["source_implemented_items"] == 6
+    assert projection["rollups"]["runtime_closed_items"] == 0
+    assert projection["rollups"]["program_completion_percent"] == 0
+    assert projection["domain_count"] == 7
+    assert projection["gemini_paid_call_allowed"] is False
+    assert "work_items" not in projection
+
+
+def test_loader_rejects_gemini_paid_enablement_without_cost_gate(
+    tmp_path: Path,
+) -> None:
+    payload = json.loads(SNAPSHOT.read_text(encoding="utf-8"))
+    payload["provider_policy"]["gemini_paid_call_allowed"] = True
+    target = tmp_path / SNAPSHOT.name
+    target.write_text(json.dumps(payload), encoding="utf-8")
+
+    with pytest.raises(ValueError, match="explicit cost cap"):
+        load_sre_k3s_controlled_automation_work_items(tmp_path)
+
+
+def test_loader_rejects_unknown_asset_executor(tmp_path: Path) -> None:
+    payload = json.loads(SNAPSHOT.read_text(encoding="utf-8"))
+    unknown = next(
+        row for row in payload["domain_routes"] if row["domain"] == "unknown"
+    )
+    unknown["executor"] = "generic_fallback"
+    target = tmp_path / SNAPSHOT.name
+    target.write_text(json.dumps(payload), encoding="utf-8")
+
+    with pytest.raises(ValueError, match="unknown domain"):
+        load_sre_k3s_controlled_automation_work_items(tmp_path)
+
+
+def test_endpoint_returns_full_work_ledger() -> None:
+    app = FastAPI()
+    app.include_router(router, prefix="/api/v1")
+    client = TestClient(app)
+
+    response = client.get("/api/v1/agents/sre-k3s-controlled-automation-work-items")
+
+    assert response.status_code == 200
+    payload = response.json()
+    assert payload["program_id"] == "AIA-SRE-P0-20260715"
+    assert len(payload["work_items"]) == 18
+    assert payload["rollups"]["runtime_closed_items"] == 0
+    assert payload["agent99_host_operations_bridge"][
+        "completion_callback"
+    ] == "/api/v1/agents/agent99/completion-callback"
diff --git a/apps/api/tests/test_sre_typed_domain_router.py b/apps/api/tests/test_sre_typed_domain_router.py
new file mode 100644
index 000000000..c9be4b7df
--- /dev/null
+++ b/apps/api/tests/test_sre_typed_domain_router.py
@@ -0,0 +1,364 @@
+from __future__ import annotations
+
+# ruff: noqa: E402, I001
+
+import hashlib
+import os
+from pathlib import Path
+
+import pytest
+import yaml
+
+os.environ.setdefault("DATABASE_URL", "postgresql+asyncpg://test:test@localhost/test")
+
+from src.services.awooop_ansible_audit_service import (  # noqa: E402
+    _catalog_hints,
+    build_ansible_decision_audit_payload,
+    get_ansible_catalog_item,
+)
+from src.services.awooop_ansible_check_mode_service import (  # noqa: E402
+    build_ansible_check_mode_claim_input,
+)
+from src.services.awooop_ansible_post_verifier import (  # noqa: E402
+    postconditions_for_catalog,
+)
+from src.services.agent99_sre_bridge import build_agent99_sre_alert  # noqa: E402
+from src.services.agent99_sre_bridge import (  # noqa: E402
+    bridge_alertmanager_to_agent99,
+)
+from src.services.controlled_alert_target_router import (  # noqa: E402
+    resolve_typed_alert_target,
+)
+from src.services.service_registry import (  # noqa: E402
+    ServiceRegistryClient,
+    StatefulLevel,
+)
+
+
+ROOT = Path(__file__).resolve().parents[3]
+REGISTRY = ROOT / "ops" / "config" / "service-registry.yaml"
+CONFIGMAP = ROOT / "k8s" / "awoooi-prod" / "15-service-registry-configmap.yaml"
+PLAYBOOK = (
+    ROOT
+    / "infra"
+    / "ansible"
+    / "playbooks"
+    / "110-sentry-profiling-consumer-recovery.yml"
+)
+SENTRY_CONTAINER = "sentry-self-hosted-snuba-profiling-functions-consumer-1"
+SENTRY_CATALOG = "ansible:110-sentry-profiling-consumer-recovery"
+
+
+def _sentry_incident() -> dict:
+    return {
+        "incident_id": "INC-20260714-471307",
+        "project_id": "awoooi",
+        "alertname": "DockerContainerUnhealthy",
+        "alert_category": "infrastructure",
+        "affected_services": [SENTRY_CONTAINER],
+        "signals": [
+            {
+                "alert_name": "DockerContainerUnhealthy",
+                "labels": {
+                    "container_name": SENTRY_CONTAINER,
+                    "namespace": "default",
+                    "host": "110",
+                },
+            }
+        ],
+    }
+
+
+def test_unknown_registry_identity_fails_closed_without_restart_fallback() -> None:
+    registry = ServiceRegistryClient(REGISTRY)
+
+    identity = registry.resolve_identity("not-in-canonical-registry")
+
+    assert identity["resolution_status"] == "asset_identity_unresolved"
+    assert identity["stateful_level"] == "UNRESOLVED"
+    assert identity["controlled_apply_allowed"] is False
+    assert identity["drift_work_item_id"].startswith("AIA-ASSET-DRIFT-")
+    assert registry.get_stateful_level("not-in-canonical-registry") == (
+        StatefulLevel.UNRESOLVED
+    )
+    assert registry.is_blocked("not-in-canonical-registry") is True
+    assert registry.get_restart_command("not-in-canonical-registry") == ""
+
+
+def test_sentry_container_resolves_to_exact_host110_domain_route() -> None:
+    route = resolve_typed_alert_target(
+        alertname="DockerContainerUnhealthy",
+        target_resource=SENTRY_CONTAINER,
+        namespace="default",
+        labels={"host": "110", "container_name": SENTRY_CONTAINER},
+        alert_category="infrastructure",
+        registry=ServiceRegistryClient(REGISTRY),
+    )
+
+    assert route["resolution_status"] == "resolved"
+    assert route["target_kind"] == "docker_container"
+    assert route["canonical_asset_id"] == "service:sentry"
+    assert route["executor"] == "host_ansible_executor"
+    assert route["allowed_catalog_ids"] == [SENTRY_CATALOG]
+    assert route["allowed_inventory_hosts"] == ["host_110"]
+    assert route["cross_domain_fallback_allowed"] is False
+    assert route["reasoning_contract"]["provider_order"] == [
+        "ollama_gcp_a",
+        "ollama_gcp_b",
+        "ollama_local_111",
+    ]
+    assert route["reasoning_contract"]["paid_call_allowed"] is False
+
+
+def test_sentry_incident_cannot_fall_to_host188_or_full_convergence() -> None:
+    hints = _catalog_hints(_sentry_incident(), None)
+
+    assert hints["match_mode"] == "typed_domain_router_v2"
+    assert hints["decision_effect"] == "bounded_domain_candidate"
+    assert [row["catalog_id"] for row in hints["candidates"]] == [
+        SENTRY_CATALOG
+    ]
+    assert hints["candidates"][0]["inventory_hosts"] == ["host_110"]
+    assert hints["cross_domain_fallback_allowed"] is False
+    assert "ansible:188-ai-web" in hints["unmatched_catalog_ids"]
+    assert "ansible:110-devops-full-convergence" in hints[
+        "unmatched_catalog_ids"
+    ]
+
+
+def test_unknown_and_kubernetes_assets_never_enter_ansible_fallback() -> None:
+    unknown = _catalog_hints(
+        {
+            "alertname": "DockerContainerUnhealthy",
+            "affected_services": ["mystery-container"],
+        },
+        None,
+    )
+    kubernetes = _catalog_hints(
+        {
+            "alertname": "DeploymentUnavailable",
+            "affected_services": ["awoooi-api"],
+            "signals": [
+                {
+                    "labels": {
+                        "namespace": "awoooi-prod",
+                        "workload_kind": "deployment",
+                    }
+                }
+            ],
+        },
+        None,
+    )
+
+    assert unknown["decision_effect"] == "create_asset_drift_work_item"
+    assert unknown["candidates"] == []
+    assert unknown["drift_work_item_id"].startswith("AIA-ASSET-DRIFT-")
+    assert kubernetes["decision_effect"] == "delegate_to_domain_executor"
+    assert kubernetes["candidates"] == []
+    assert kubernetes["typed_target_route"]["executor"] == (
+        "kubernetes_controlled_executor"
+    )
+
+
+def test_backup_restore_is_critical_readback_only() -> None:
+    route = resolve_typed_alert_target(
+        alertname="BackupRestoreEscrowMissing",
+        target_resource="backup_restore",
+        namespace="",
+        labels={},
+        registry=ServiceRegistryClient(REGISTRY),
+    )
+
+    assert route["target_kind"] == "backup_restore"
+    assert route["risk_class"] == "critical"
+    assert route["critical_break_glass_required"] is True
+    assert route["controlled_apply_allowed"] is False
+    assert route["allowed_catalog_ids"] == []
+    assert route["agent99_bridge"]["dispatch_allowed"] is False
+
+
+def test_agent99_payload_carries_same_typed_host_operations_contract() -> None:
+    payload = build_agent99_sre_alert(
+        alert_id="INC-COLD-START-99",
+        alertname="ColdStartGateBlocked",
+        severity="warning",
+        namespace="default",
+        target_resource="cold-start-gate",
+        message="controlled recovery required",
+        labels={"host": "99"},
+        fingerprint="cold-start-fingerprint",
+        alert_category="host_recovery",
+    )
+
+    route = payload["routing"]["typedTargetRoute"]
+    bridge = payload["awoooi"]["agent99HostOperationsBridge"]
+    assert route["target_kind"] == "control_plane_recovery"
+    assert payload["routing"]["typedDispatchAllowed"] is True
+    assert payload["routing"]["crossDomainFallbackAllowed"] is False
+    assert payload["awoooi"]["canonicalAssetId"] == (
+        "control-plane:cold-start-gate"
+    )
+    assert bridge["dispatch_identity"] == (
+        "agent99_controlled_dispatch_identity_v1"
+    )
+    assert bridge["command_envelope"] == "agent99_controlled_dispatch_input_v1"
+    assert bridge["completion_callback"] == (
+        "/api/v1/agents/agent99/completion-callback"
+    )
+    assert bridge["required_correlation"] == [
+        "trace_id",
+        "run_id",
+        "work_item_id",
+    ]
+
+
+def test_host112_guest_recovery_is_an_exact_agent99_domain_route() -> None:
+    payload = build_agent99_sre_alert(
+        alert_id="INC-HOST112-GUEST",
+        alertname="Host112GuestNotReady",
+        severity="critical",
+        namespace="host112",
+        target_resource="host112-guest-recovery",
+        message="Host 112 guest readiness is degraded",
+        labels={"host": "112", "service": "host112-guest-recovery"},
+        fingerprint="host112-guest-readiness",
+        alert_category="host_recovery",
+    )
+
+    route = payload["routing"]["typedTargetRoute"]
+    assert route["canonical_asset_id"] == "host-guest:host_112"
+    assert route["executor"] == "Agent99"
+    assert route["allowed_inventory_hosts"] == ["host_112"]
+    assert route["agent99_bridge"]["dispatch_allowed"] is True
+    assert payload["routing"]["typedDispatchAllowed"] is True
+    assert payload["routing"]["crossDomainFallbackAllowed"] is False
+
+
+@pytest.mark.asyncio
+async def test_agent99_rejects_cross_domain_mutating_dispatch_before_transport(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    dispatched: list[dict] = []
+    monkeypatch.setattr(
+        "src.services.agent99_sre_bridge.dispatch_agent99_sre_alert_with_receipt",
+        lambda payload: dispatched.append(payload) or {},
+    )
+
+    result = await bridge_alertmanager_to_agent99(
+        alert_id="INC-K3S-CROSS-DOMAIN",
+        alertname="KubePodCrashLooping",
+        severity="critical",
+        namespace="awoooi-prod",
+        target_resource="awoooi-api",
+        message="K3s workload requires its domain executor",
+        labels={"workload_kind": "deployment"},
+        fingerprint="k3s-cross-domain",
+        alert_category="kubernetes",
+        project_id="awoooi",
+        incident_id="INC-K3S-CROSS-DOMAIN",
+        work_item_id="AIA-SRE-K3S-CROSS-DOMAIN",
+    )
+
+    assert result["status"] == "failed"
+    assert result["reason"] == "agent99_typed_dispatch_not_allowed"
+    assert result["dispatchPerformed"] is False
+    assert result["typedTargetRoute"]["target_kind"] == "kubernetes_workload"
+    assert dispatched == []
+
+
+def test_typed_candidate_scope_rejects_cross_host_tampering() -> None:
+    class Incident:
+        incident_id = "INC-20260714-471307"
+        project_id = "awoooi"
+        alertname = "DockerContainerUnhealthy"
+        alert_category = "infrastructure"
+        notification_type = "TYPE-3"
+        severity = None
+        affected_services = [SENTRY_CONTAINER]
+        signals = []
+
+    payload = build_ansible_decision_audit_payload(
+        incident=Incident(),
+        proposal_data={"source": "test", "risk_level": "medium"},
+        decision_path="repair_candidate_controlled_queue",
+        not_used_reason="test",
+    )
+    assert payload is not None
+    candidate_input = payload["input"]
+    candidate_input["executor_candidates"][0]["inventory_hosts"] = ["host_188"]
+
+    with pytest.raises(ValueError, match="no_safe_check_mode_candidate"):
+        build_ansible_check_mode_claim_input(
+            source_candidate_op_id="00000000-0000-0000-0000-000000471307",
+            candidate_input=candidate_input,
+        )
+
+
+def test_sentry_candidate_reaches_bounded_check_mode_with_exact_scope() -> None:
+    class Incident:
+        incident_id = "INC-20260714-471307"
+        project_id = "awoooi"
+        alertname = "DockerContainerUnhealthy"
+        alert_category = "infrastructure"
+        notification_type = "TYPE-3"
+        severity = None
+        affected_services = [SENTRY_CONTAINER]
+        signals = []
+
+    payload = build_ansible_decision_audit_payload(
+        incident=Incident(),
+        proposal_data={"source": "test", "risk_level": "medium"},
+        decision_path="repair_candidate_controlled_queue",
+        not_used_reason="test",
+    )
+    assert payload is not None
+
+    claim = build_ansible_check_mode_claim_input(
+        source_candidate_op_id="00000000-0000-0000-0000-000000471307",
+        candidate_input=payload["input"],
+    )
+
+    assert claim["catalog_id"] == SENTRY_CATALOG
+    assert claim["inventory_hosts"] == ["host_110"]
+    assert claim["typed_domain"] == "docker_container"
+    assert claim["canonical_asset_id"] == "service:sentry"
+    assert claim["target_scope_validated"] is True
+    assert claim["cross_domain_fallback_allowed"] is False
+    assert claim["controlled_apply_allowed"] is True
+
+
+def test_sentry_playbook_and_post_verifier_are_exact_container_bounded() -> None:
+    payload = yaml.safe_load(PLAYBOOK.read_text(encoding="utf-8"))
+    source = PLAYBOOK.read_text(encoding="utf-8")
+    catalog = get_ansible_catalog_item(SENTRY_CATALOG)
+    postconditions = postconditions_for_catalog(SENTRY_CATALOG)
+
+    assert payload[0]["hosts"] == "host_110"
+    assert payload[0]["become"] is False
+    assert catalog is not None
+    assert catalog["inventory_hosts"] == ["host_110"]
+    assert catalog["auto_apply_enabled"] is True
+    assert catalog["risk_level"] == "medium"
+    assert len(postconditions) == 2
+    assert all(row.inventory_host == "host_110" for row in postconditions)
+    assert source.count(SENTRY_CONTAINER) == 1
+    for forbidden in (
+        "docker compose",
+        "docker system prune",
+        "host_188",
+        "reboot",
+        "systemctl restart",
+    ):
+        assert forbidden not in source.lower()
+
+
+def test_runtime_registry_is_exact_generated_mirror() -> None:
+    source_text = REGISTRY.read_text(encoding="utf-8")
+    configmap = yaml.safe_load(CONFIGMAP.read_text(encoding="utf-8"))
+    runtime_text = configmap["data"]["service-registry.yaml"]
+    annotation = configmap["metadata"]["annotations"][
+        "awoooi.wooo.work/source-sha256"
+    ]
+
+    assert runtime_text == source_text
+    assert annotation == hashlib.sha256(source_text.encode("utf-8")).hexdigest()
diff --git a/docs/operations/sre-k3s-controlled-automation-work-items.snapshot.json b/docs/operations/sre-k3s-controlled-automation-work-items.snapshot.json
new file mode 100644
index 000000000..7ee535c6b
--- /dev/null
+++ b/docs/operations/sre-k3s-controlled-automation-work-items.snapshot.json
@@ -0,0 +1,562 @@
+{
+  "schema_version": "sre_k3s_controlled_automation_work_items_v1",
+  "governance_version": "global_product_governance_v2",
+  "program_id": "AIA-SRE-P0-20260715",
+  "generated_at": "2026-07-15T15:30:00+08:00",
+  "status": "phase_1_canonical_identity_and_typed_routing_in_progress",
+  "scope_complete": false,
+  "current_p0": {
+    "id": "AIA-SRE-004",
+    "title": "Canonical Asset Normalize 與 Typed Domain Router production closure",
+    "phase": "phase_1",
+    "terminal_condition": "同一 source SHA 通過 focused/full tests、Gitea CD、production registry parity、typed route replay 與 Sentry exact-container verifier readback"
+  },
+  "architecture": {
+    "pipeline": [
+      "Alert",
+      "Canonical Asset Normalize",
+      "Typed Domain Router",
+      "HolmesGPT Investigator",
+      "Ollama RCA / Gemini Critic",
+      "Deterministic Policy",
+      "Single Controlled Executor",
+      "Independent Verifier",
+      "Incident Closure + KM/RAG/MCP/PlayBook"
+    ],
+    "cross_domain_fallback_allowed": false,
+    "unknown_asset_terminal": "asset_identity_unresolved",
+    "circuit_open_terminal": "stay_in_same_domain_and_create_repair_work_item",
+    "same_run_identity_required": [
+      "trace_id",
+      "run_id",
+      "work_item_id"
+    ]
+  },
+  "provider_policy": {
+    "ordered_route": [
+      "ollama_gcp_a",
+      "ollama_gcp_b",
+      "ollama_local_111",
+      "gemini_api"
+    ],
+    "ollama_role": "primary_rca_and_action_planning",
+    "gemini_role": "final_fallback_and_optional_critic_only",
+    "gemini_paid_call_allowed": false,
+    "gemini_cost_cap_status": "explicit_cost_cap_not_approved",
+    "gemini_credential_status": "runtime_secret_reference_verifier_pending_no_secret_value_read",
+    "production_provider_route_switch_allowed": false,
+    "required_before_gemini_enablement": [
+      "offline_replay_scorecard",
+      "shadow_comparison",
+      "bounded_canary",
+      "daily_and_monthly_cost_cap",
+      "per_incident_token_cap",
+      "rate_limit_and_circuit_breaker",
+      "rollback_to_ollama_chain",
+      "production_cost_readback"
+    ]
+  },
+  "agent99_host_operations_bridge": {
+    "role": "on_prem_policy_controlled_ai_execution_node",
+    "single_executor_domains": [
+      "windows_vmware",
+      "control_plane_recovery"
+    ],
+    "relay_only_domains": [
+      "host_systemd",
+      "docker_container"
+    ],
+    "dispatch_identity_schema": "agent99_controlled_dispatch_identity_v1",
+    "command_envelope_schema": "agent99_controlled_dispatch_input_v1",
+    "dispatch_receipt_schema": "agent99_controlled_dispatch_receipt_v1",
+    "outcome_schema": "agent99_outcome_contract_v1",
+    "completion_callback": "/api/v1/agents/agent99/completion-callback",
+    "arbitrary_command_allowed": false,
+    "unknown_asset_dispatch_allowed": false,
+    "cross_domain_fallback_allowed": false,
+    "runtime_terminal": "callback_plus_independent_verifier_plus_learning_writeback"
+  },
+  "domain_routes": [
+    {
+      "domain": "kubernetes_workload",
+      "executor": "kubernetes_controlled_executor",
+      "verifier": "kubernetes_rollout_verifier",
+      "fallback": "forbidden"
+    },
+    {
+      "domain": "host_systemd",
+      "executor": "host_ansible_executor",
+      "verifier": "host_runtime_independent_verifier",
+      "fallback": "forbidden"
+    },
+    {
+      "domain": "docker_container",
+      "executor": "host_ansible_executor_with_exact_container_playbook",
+      "verifier": "container_runtime_independent_verifier",
+      "fallback": "forbidden"
+    },
+    {
+      "domain": "windows_vmware",
+      "executor": "Agent99",
+      "verifier": "agent99_independent_runtime_verifier",
+      "fallback": "forbidden"
+    },
+    {
+      "domain": "database",
+      "executor": "db_bounded_executor",
+      "verifier": "db_independent_verifier",
+      "fallback": "forbidden"
+    },
+    {
+      "domain": "backup_restore",
+      "executor": "backup_restore_break_glass",
+      "verifier": "backup_restore_readback_verifier",
+      "fallback": "forbidden",
+      "critical_read_only_default": true
+    },
+    {
+      "domain": "unknown",
+      "executor": null,
+      "verifier": null,
+      "terminal": "asset_identity_unresolved",
+      "fallback": "forbidden"
+    }
+  ],
+  "work_items": [
+    {
+      "id": "AIA-SRE-001",
+      "order": 1,
+      "priority": "P0",
+      "phase": "phase_1",
+      "title": "建立唯一 machine-readable 架構與工作總帳",
+      "owner_lane": "AIControlPlane",
+      "risk": "low",
+      "status": "source_implemented_runtime_pending",
+      "dependencies": [],
+      "source_refs": [
+        "docs/operations/sre-k3s-controlled-automation-work-items.snapshot.json",
+        "apps/api/src/services/sre_k3s_controlled_automation_work_items.py"
+      ],
+      "executor": "Gitea controlled CD",
+      "verifier": "ledger schema and API readback verifier",
+      "rollback": "revert source commit",
+      "exit_condition": "production API returns the exact committed ledger and rollups",
+      "next_action": "complete loader/API tests then deploy"
+    },
+    {
+      "id": "AIA-SRE-002",
+      "order": 2,
+      "priority": "P0",
+      "phase": "phase_1",
+      "title": "Canonical Service Registry 單一來源與 runtime exact mirror",
+      "owner_lane": "AssetIdentity",
+      "risk": "medium",
+      "status": "source_implemented_runtime_pending",
+      "dependencies": [
+        "AIA-SRE-001"
+      ],
+      "source_refs": [
+        "ops/config/service-registry.yaml",
+        "scripts/ops/render-service-registry-configmap.py",
+        "k8s/awoooi-prod/15-service-registry-configmap.yaml"
+      ],
+      "executor": "service registry renderer plus Gitea CD",
+      "verifier": "source hash and exact embedded YAML parity",
+      "rollback": "reapply previous ConfigMap from prior Gitea SHA",
+      "exit_condition": "production ConfigMap source hash and API identity lookup match Gitea main",
+      "next_action": "deploy and read back ConfigMap annotation"
+    },
+    {
+      "id": "AIA-SRE-003",
+      "order": 3,
+      "priority": "P0",
+      "phase": "phase_1",
+      "title": "Unknown asset fail-closed 與 deterministic drift work item",
+      "owner_lane": "AssetIdentity",
+      "risk": "medium",
+      "status": "source_implemented_runtime_pending",
+      "dependencies": [
+        "AIA-SRE-002"
+      ],
+      "source_refs": [
+        "apps/api/src/services/service_registry.py",
+        "apps/api/src/services/auto_repair_service.py"
+      ],
+      "executor": "none for unresolved identity",
+      "verifier": "unknown asset replay must return UNRESOLVED and zero candidates",
+      "rollback": "not applicable; no-write terminal",
+      "exit_condition": "unknown assets create a stable drift ID and never return AUTO/restart fallback",
+      "next_action": "production replay one unknown asset without runtime apply"
+    },
+    {
+      "id": "AIA-SRE-004",
+      "order": 4,
+      "priority": "P0",
+      "phase": "phase_1",
+      "title": "Typed Domain Router 與跨 domain fallback 禁止",
+      "owner_lane": "AutomationRouter",
+      "risk": "high",
+      "status": "source_implemented_runtime_pending",
+      "dependencies": [
+        "AIA-SRE-003"
+      ],
+      "source_refs": [
+        "apps/api/src/services/controlled_alert_target_router.py",
+        "apps/api/src/services/awooop_ansible_audit_service.py",
+        "apps/api/src/services/awooop_ansible_check_mode_service.py"
+      ],
+      "executor": "typed policy router",
+      "verifier": "domain/catalog/host/canonical ID scope verifier",
+      "rollback": "revert router source while keeping unknown fail-closed",
+      "exit_condition": "every candidate is exact-domain scoped and circuit-open never selects another host/domain",
+      "next_action": "run full Ansible replay regression"
+    },
+    {
+      "id": "AIA-SRE-005",
+      "order": 5,
+      "priority": "P0",
+      "phase": "phase_1",
+      "title": "Sentry profiling consumer exact-container bounded recovery",
+      "owner_lane": "HostContainerAutomation",
+      "risk": "medium",
+      "status": "source_implemented_runtime_pending",
+      "dependencies": [
+        "AIA-SRE-004"
+      ],
+      "source_refs": [
+        "infra/ansible/playbooks/110-sentry-profiling-consumer-recovery.yml",
+        "apps/api/src/services/awooop_ansible_post_verifier.py"
+      ],
+      "executor": "awoooi-ansible-executor-broker",
+      "verifier": "exact container running plus healthy readback",
+      "rollback": "stop further writes, same-domain check replay and cooldown",
+      "exit_condition": "INC-20260714-471307 replay selects only host_110 exact playbook and post-verifier passes",
+      "next_action": "deploy then execute one bounded same-fingerprint controlled replay"
+    },
+    {
+      "id": "AIA-SRE-006",
+      "order": 6,
+      "priority": "P0",
+      "phase": "phase_2",
+      "title": "Agent99 Host Operations Bridge typed enforcement",
+      "owner_lane": "Agent99",
+      "risk": "high",
+      "status": "source_implemented_runtime_pending",
+      "dependencies": [
+        "AIA-SRE-004"
+      ],
+      "source_refs": [
+        "apps/api/src/services/agent99_sre_bridge.py",
+        "apps/api/src/services/agent99_controlled_dispatch_ledger.py",
+        "apps/api/src/services/agent99_completion_callback.py"
+      ],
+      "executor": "Agent99 for Windows/VMware/control-plane; relay-only elsewhere",
+      "verifier": "Agent99 callback plus external same-run verifier",
+      "rollback": "disable typed dispatch route and retain no-write Status readback",
+      "exit_condition": "all Agent99 dispatches carry canonical typed scope and no legacy catch-all can mutate",
+      "next_action": "deploy then replay one allowlisted Host112 or cold-start run plus one cross-domain denied candidate"
+    },
+    {
+      "id": "AIA-SRE-007",
+      "order": 7,
+      "priority": "P0",
+      "phase": "phase_2",
+      "title": "K3s workload 專屬 executor 與 rollout verifier",
+      "owner_lane": "KubernetesAutomation",
+      "risk": "high",
+      "status": "planned",
+      "dependencies": [
+        "AIA-SRE-004"
+      ],
+      "source_refs": [
+        "apps/api/src/services/executor.py",
+        "apps/api/src/services/post_execution_verifier.py"
+      ],
+      "executor": "kubernetes_controlled_executor",
+      "verifier": "kubernetes_rollout_verifier",
+      "rollback": "declarative rollout undo or previous immutable revision",
+      "exit_condition": "K3s incidents never enter Ansible/Agent99 fallback and close with rollout evidence",
+      "next_action": "inventory current kubernetes actions and bind allowlisted declarative routes"
+    },
+    {
+      "id": "AIA-SRE-008",
+      "order": 8,
+      "priority": "P0",
+      "phase": "phase_2",
+      "title": "Host/systemd 專屬 Ansible executor inventory",
+      "owner_lane": "HostAutomation",
+      "risk": "high",
+      "status": "in_progress",
+      "dependencies": [
+        "AIA-SRE-004"
+      ],
+      "source_refs": [
+        "infra/ansible/inventory/hosts.yml",
+        "apps/api/src/services/awooop_ansible_audit_service.py"
+      ],
+      "executor": "host_ansible_executor",
+      "verifier": "host_runtime_independent_verifier",
+      "rollback": "catalog-specific compensating action",
+      "exit_condition": "every host action has one exact inventory host, check mode, bounded apply and postcondition",
+      "next_action": "close catalogs still backed by broad convergence playbooks"
+    },
+    {
+      "id": "AIA-SRE-009",
+      "order": 9,
+      "priority": "P0",
+      "phase": "phase_2",
+      "title": "Windows/VMware Agent99 single-executor closure",
+      "owner_lane": "Agent99",
+      "risk": "high",
+      "status": "in_progress",
+      "dependencies": [
+        "AIA-SRE-006"
+      ],
+      "source_refs": [
+        "apps/api/src/services/agent99_controlled_dispatch_ledger.py",
+        "apps/api/src/services/agent99_same_run_reconcile.py"
+      ],
+      "executor": "Agent99",
+      "verifier": "agent99_independent_runtime_verifier",
+      "rollback": "allowlisted Status/no-write reconciliation and bounded generation retry",
+      "exit_condition": "Windows/VMware runs use durable identity, callback, verifier and learning receipt",
+      "next_action": "reconcile remaining Agent99 enterprise work ledger with this program"
+    },
+    {
+      "id": "AIA-SRE-010",
+      "order": 10,
+      "priority": "P0",
+      "phase": "phase_3",
+      "title": "DB 專屬 verifier 與 bounded executor",
+      "owner_lane": "DatabaseAutomation",
+      "risk": "high",
+      "status": "planned",
+      "dependencies": [
+        "AIA-SRE-004"
+      ],
+      "source_refs": [
+        "ops/config/service-registry.yaml"
+      ],
+      "executor": "db_bounded_executor",
+      "verifier": "db_independent_verifier",
+      "rollback": "transactional or compensating action only; destructive restore remains break-glass",
+      "exit_condition": "DB incidents use DB-native preflight/postconditions and never generic host restart",
+      "next_action": "classify read-only, session, pool, replication and bounded maintenance actions"
+    },
+    {
+      "id": "AIA-SRE-011",
+      "order": 11,
+      "priority": "P0",
+      "phase": "phase_3",
+      "title": "Backup/restore readback 與 critical break-glass",
+      "owner_lane": "DataProtection",
+      "risk": "critical",
+      "status": "policy_active",
+      "dependencies": [
+        "AIA-SRE-004"
+      ],
+      "source_refs": [
+        "apps/api/src/services/backup_restore_signal_automation.py"
+      ],
+      "executor": "backup_restore_break_glass",
+      "verifier": "backup_restore_readback_verifier",
+      "rollback": "no-write default; destructive action requires incident-specific break-glass",
+      "exit_condition": "freshness, offsite, escrow and restore-drill evidence close without false green",
+      "next_action": "bind typed route to DR scorecard without running backup/restore"
+    },
+    {
+      "id": "AIA-SRE-012",
+      "order": 12,
+      "priority": "P0",
+      "phase": "phase_3",
+      "title": "HolmesGPT Investigator shadow integration",
+      "owner_lane": "Investigator",
+      "risk": "high",
+      "status": "planned",
+      "dependencies": [
+        "AIA-SRE-004"
+      ],
+      "source_refs": [
+        "apps/api/src/services/pre_decision_investigator.py"
+      ],
+      "executor": "none in shadow",
+      "verifier": "offline replay quality and prompt-injection safety scorecard",
+      "rollback": "remove shadow consumer; active PreDecisionInvestigator remains",
+      "exit_condition": "internal immutable artifact passes replay, shadow and canary before core replacement",
+      "next_action": "source HolmesGPT from approved internal mirror without GitHub access"
+    },
+    {
+      "id": "AIA-SRE-013",
+      "order": 13,
+      "priority": "P0",
+      "phase": "phase_3",
+      "title": "Ollama RCA 與 Gemini Critic/cost gate",
+      "owner_lane": "ModelRouter",
+      "risk": "critical",
+      "status": "in_progress",
+      "dependencies": [
+        "AIA-SRE-012"
+      ],
+      "source_refs": [
+        "apps/api/src/services/ai_router.py",
+        "apps/api/src/services/ollama_failover.py"
+      ],
+      "executor": "ordered model router",
+      "verifier": "provider-attributed replay accuracy, latency and cost readback",
+      "rollback": "Gemini disabled; remain on ordered Ollama chain",
+      "exit_condition": "provider order is enforced and Gemini cannot run without explicit cost cap",
+      "next_action": "verify secret metadata presence without reading value, then build offline comparison"
+    },
+    {
+      "id": "AIA-SRE-014",
+      "order": 14,
+      "priority": "P0",
+      "phase": "phase_4",
+      "title": "Independent verifier registry 完整覆蓋",
+      "owner_lane": "PostExecutionVerifier",
+      "risk": "high",
+      "status": "in_progress",
+      "dependencies": [
+        "AIA-SRE-005",
+        "AIA-SRE-007",
+        "AIA-SRE-008",
+        "AIA-SRE-009",
+        "AIA-SRE-010"
+      ],
+      "source_refs": [
+        "apps/api/src/services/awooop_ansible_post_verifier.py",
+        "apps/api/src/services/post_execution_verifier.py"
+      ],
+      "executor": "none",
+      "verifier": "domain-specific external readback",
+      "rollback": "failed verifier triggers same-domain retry or compensating action",
+      "exit_condition": "100 percent of mutating catalogs have independent postconditions",
+      "next_action": "reconcile catalog IDs against postcondition registry"
+    },
+    {
+      "id": "AIA-SRE-015",
+      "order": 15,
+      "priority": "P0",
+      "phase": "phase_4",
+      "title": "Incident closure 與 KM/RAG/MCP/PlayBook durable writeback",
+      "owner_lane": "LearningControlPlane",
+      "risk": "medium",
+      "status": "in_progress",
+      "dependencies": [
+        "AIA-SRE-014"
+      ],
+      "source_refs": [
+        "apps/api/src/services/learning_service.py",
+        "apps/api/src/services/awooop_ansible_learning_writeback.py"
+      ],
+      "executor": "learning writeback consumer",
+      "verifier": "durable same-run acknowledgement readback",
+      "rollback": "append-only correction event; immutable receipts are not overwritten",
+      "exit_condition": "closure requires all learning acknowledgements on the same run",
+      "next_action": "bind typed domain and verifier outcome into immutable learning receipt"
+    },
+    {
+      "id": "AIA-SRE-016",
+      "order": 16,
+      "priority": "P1",
+      "phase": "phase_5",
+      "title": "24h replay、MTTA/MTTR、誤判、復發與成本 scorecard",
+      "owner_lane": "AutomationSLO",
+      "risk": "low",
+      "status": "planned",
+      "dependencies": [
+        "AIA-SRE-015"
+      ],
+      "source_refs": [],
+      "executor": "offline replay worker",
+      "verifier": "production aggregate versus same-run receipts",
+      "rollback": "read-only scorecard",
+      "exit_condition": "all lanes publish MTTA, MTTR, false positive, recurrence, human intervention, verifier and rollback metrics",
+      "next_action": "replay the latest 24h alert/runs corpus after phase 1 deploy"
+    },
+    {
+      "id": "AIA-SRE-017",
+      "order": 17,
+      "priority": "P1",
+      "phase": "phase_5",
+      "title": "Telegram 與 cockpit 只呈現 canonical lifecycle truth",
+      "owner_lane": "ProductExperience",
+      "risk": "medium",
+      "status": "in_progress",
+      "dependencies": [
+        "AIA-SRE-015"
+      ],
+      "source_refs": [
+        "ops/config/telegram-routing-registry.yaml"
+      ],
+      "executor": "canonical Telegram gateway",
+      "verifier": "delivery receipt plus desktop/mobile visible smoke",
+      "rollback": "suppress duplicate projection; retain incident ledger",
+      "exit_condition": "no alert is sprayed across groups/bots and every card links one canonical run",
+      "next_action": "reconcile project/product/alert class routing against the typed domain"
+    },
+    {
+      "id": "AIA-SRE-018",
+      "order": 18,
+      "priority": "P0",
+      "phase": "phase_6",
+      "title": "Gitea CD、production runtime 與 visible closure",
+      "owner_lane": "ReleaseControl",
+      "risk": "high",
+      "status": "planned",
+      "dependencies": [
+        "AIA-SRE-001",
+        "AIA-SRE-002",
+        "AIA-SRE-003",
+        "AIA-SRE-004",
+        "AIA-SRE-005",
+        "AIA-SRE-006",
+        "AIA-SRE-014",
+        "AIA-SRE-015"
+      ],
+      "source_refs": [
+        ".gitea/workflows/cd.yaml"
+      ],
+      "executor": "Gitea controlled CD",
+      "verifier": "deploy marker, image SHA, API readback and desktop/mobile smoke",
+      "rollback": "previous immutable image revision and ConfigMap source SHA",
+      "exit_condition": "source/test/CD/runtime/UI evidence all match one deployed SHA",
+      "next_action": "commit phase 1, normal push after shared writer release, then terminal CD/readback"
+    }
+  ],
+  "rollups": {
+    "total_items": 18,
+    "by_priority": {
+      "P0": 16,
+      "P1": 2
+    },
+    "by_status": {
+      "source_implemented_runtime_pending": 6,
+      "in_progress": 6,
+      "planned": 5,
+      "policy_active": 1
+    },
+    "source_implemented_items": 6,
+    "runtime_closed_items": 0,
+    "program_completion_percent": 0,
+    "asset_coverage_status": "partial",
+    "runtime_closure_status": "not_started_for_this_program_sha"
+  },
+  "completion_contract": {
+    "required_stages": [
+      "sensor_source_receipt",
+      "canonical_asset_identity",
+      "source_truth_diff",
+      "ai_decision_candidate",
+      "risk_policy_decision",
+      "check_mode_receipt",
+      "bounded_execution_receipt",
+      "independent_verifier_or_no_write_terminal",
+      "incident_closure",
+      "telegram_receipt",
+      "km_rag_mcp_playbook_writeback_ack"
+    ],
+    "terminal_without_all_stages": "partial_or_degraded_with_safe_next_action",
+    "source_test_cd_green_is_runtime_closure": false
+  }
+}
diff --git a/docs/superpowers/specs/2026-04-15-MASTER-ai-autonomous-flywheel-v2.md b/docs/superpowers/specs/2026-04-15-MASTER-ai-autonomous-flywheel-v2.md
index 068ccecc8..1ab310e51 100644
--- a/docs/superpowers/specs/2026-04-15-MASTER-ai-autonomous-flywheel-v2.md
+++ b/docs/superpowers/specs/2026-04-15-MASTER-ai-autonomous-flywheel-v2.md
@@ -161,6 +161,20 @@ AWOOOI / AwoooP / IwoooS 不是單純監控頁、告警轉發器或資安清冊
 
 ---
 
+### 1.8 2026-07-15 SRE / K3s 固定 typed automation 架構
+
+統帥批准的固定主幹如下，後續不得因單一告警改回 keyword-first、generic fallback 或多 executor 競寫：
+
+`Alert → Canonical Asset Normalize → Typed Domain Router → HolmesGPT Investigator → Ollama RCA / Gemini Critic → Deterministic Policy → Single Controlled Executor → Independent Verifier → Incident Closure + KM/RAG/MCP/PlayBook`
+
+1. K3s workload 只走 Kubernetes executor / verifier；host / systemd 只走 Host Ansible；Docker Compose / Sentry 只走該 host 的 exact-container PlayBook；Windows / VMware 與 control-plane recovery 由 Agent99 執行；DB 只走 DB bounded executor / verifier；backup / restore 預設 readback-only 且屬 critical break-glass。
+2. Unknown asset 固定回 `asset_identity_unresolved` 並建立 deterministic drift work item；不得 fallback AUTO。Circuit-open 固定留在原 domain 建修復工作項；不得滑落到另一主機、另一 domain 或 Agent99 arbitrary command。
+3. Agent99 是既有 on-prem AI 執行節點，不另造孤島：沿用 `agent99_controlled_dispatch_identity_v1`、durable ledger、single-flight、outbox、outcome ingestion、completion callback 與 independent verifier；每次必須帶同一 `trace_id/run_id/work_item_id`。Linux host / Docker lane 中 Agent99 只能作 no-secret relay / coordination，不得取代專屬 Ansible executor。
+4. 模型順序固定為 GCP-A Ollama → GCP-B Ollama → 111 Ollama → Gemini API。Gemini 只作最後備援 / critic；未有明確日/月/per-incident 成本上限、replay、shadow、canary 與 rollback 前，production paid call 與 provider route switch 均維持禁止。
+5. 唯一 machine-readable 工作總帳為 `docs/operations/sre-k3s-controlled-automation-work-items.snapshot.json`；API readback 為 `/api/v1/agents/sre-k3s-controlled-automation-work-items`。Source、test、CD、runtime 與 visible evidence 必須分開計算，只有同一 run 完整 closure 才算完成。
+
+---
+
 ## §2 當前架構診斷（鐵證 — 2026-04-15 深層病灶掃描）
 
 ### 2.1 Q1-Q5 鐵證摘要表
@@ -5356,3 +5370,11 @@ Trigger commit `f5cd37b7` 與 deploy marker `0ba92357` 已把 governance UI 的
 **觸發**：110 壓力事故 fail-closed guard 將專用 `awoooi-cd-lane.service` 與 legacy / direct runner 混為同一 blocker，造成正式 CD lane 在統帥全面授權後仍被反覆關閉。
 
 **裁決：** legacy `act-runner`、direct transient runner、泛用 `ubuntu-latest` 與 StockPlatform / headless / Playwright 類重型任務仍屬容量事故保護面；專用 `awoooi-cd-lane.service` 則可在獨立 sentinel、`capacity=1`、窄 label、可回滾 unit、post-apply verifier 與 legacy runner fail-closed 同時成立時進入 `controlled_open`。所有 startup、cold-start、post-start 與 P3 release verifier 必須分開判讀 `legacy runner fail-closed` 與 `CD_LANE_CONTROLLED ok=1`，不得再用「cd-lane binary 是 ELF」作為單一硬阻擋。
+
+### 2026-07-15 15:30 (台北) — §1.8 / SRE-K3s typed automation Phase 1
+
+**觸發**：production `INC-20260714-471307` 的 Sentry container 告警被 keyword catalog 誤導到 host 188，再滑到 host 110 broad convergence；留下 check receipt，但 apply、verifier 與 learning closure 均為零。根因同時包含 unknown asset `fallback AUTO`、source/runtime Registry 漂移與 candidate 缺少 domain/host/canonical identity scope。
+
+**Source 已推進：** 建立 18 項 machine-readable 工作總帳與 API loader；Service Registry unknown 改為 `UNRESOLVED` 並產生 deterministic drift ID；ConfigMap 改由 canonical source deterministic render；新增 typed domain router、cross-domain fail-closed、Sentry exact-container host 110 PlayBook、independent postconditions；typed route 已投影到既有 Agent99 durable dispatch/receipt/callback contract，且 mutating dispatch 已在 transport/ledger 前強制檢查 `typedDispatchAllowed`。模型順序維持 GCP-A → GCP-B → 111 → Gemini，Gemini paid call 與 provider switch 維持 explicit cost gate。
+
+**未完成不得誤報：** 本段已有 source、typed/Agent99 focused tests 與 non-integration API `4941 passed` evidence；尚未 Gitea CD、production ConfigMap parity、Sentry controlled replay、Agent99 allowlisted/cross-domain runtime replay、HolmesGPT internal artifact、DB/K3s executor closure或 KM/RAG/MCP/PlayBook 同 run writeback；program runtime closure 仍為 `0/18`。
diff --git a/infra/ansible/playbooks/110-sentry-profiling-consumer-recovery.yml b/infra/ansible/playbooks/110-sentry-profiling-consumer-recovery.yml
new file mode 100644
index 000000000..232cb7c9b
--- /dev/null
+++ b/infra/ansible/playbooks/110-sentry-profiling-consumer-recovery.yml
@@ -0,0 +1,105 @@
+---
+- name: Bounded host 110 Sentry profiling consumer recovery
+  hosts: host_110
+  gather_facts: false
+  become: false
+  vars:
+    sentry_profiling_consumer: sentry-self-hosted-snuba-profiling-functions-consumer-1
+
+  tasks:
+    - name: Inspect exact Sentry profiling consumer running state
+      ansible.builtin.command:
+        argv:
+          - docker
+          - inspect
+          - --format
+          - "{{ '{{' }}.State.Running{{ '}}' }}"
+          - "{{ sentry_profiling_consumer }}"
+      register: sentry_consumer_running
+      check_mode: false
+      changed_when: false
+      failed_when: false
+
+    - name: Inspect exact Sentry profiling consumer health state
+      ansible.builtin.command:
+        argv:
+          - docker
+          - inspect
+          - --format
+          - "{{ '{{' }}.State.Health.Status{{ '}}' }}"
+          - "{{ sentry_profiling_consumer }}"
+      register: sentry_consumer_health
+      check_mode: false
+      changed_when: false
+      failed_when: false
+
+    - name: Decide whether the exact container needs bounded recovery
+      ansible.builtin.set_fact:
+        sentry_consumer_missing_or_stopped: >-
+          {{
+            sentry_consumer_running.rc != 0
+            or (sentry_consumer_running.stdout | trim) != 'true'
+          }}
+        sentry_consumer_unhealthy: >-
+          {{
+            sentry_consumer_running.rc == 0
+            and (sentry_consumer_running.stdout | trim) == 'true'
+            and (
+              sentry_consumer_health.rc != 0
+              or (sentry_consumer_health.stdout | trim) != 'healthy'
+            )
+          }}
+
+    - name: Start only the exact stopped Sentry profiling consumer
+      ansible.builtin.command:
+        argv:
+          - docker
+          - start
+          - "{{ sentry_profiling_consumer }}"
+      when: sentry_consumer_missing_or_stopped | bool
+      changed_when: true
+
+    - name: Restart only the exact unhealthy Sentry profiling consumer
+      ansible.builtin.command:
+        argv:
+          - docker
+          - restart
+          - "{{ sentry_profiling_consumer }}"
+      when: sentry_consumer_unhealthy | bool
+      changed_when: true
+
+    - name: Verify exact Sentry profiling consumer is running after apply
+      ansible.builtin.command:
+        argv:
+          - docker
+          - inspect
+          - --format
+          - "{{ '{{' }}.State.Running{{ '}}' }}"
+          - "{{ sentry_profiling_consumer }}"
+      register: sentry_consumer_running_after
+      retries: 6
+      delay: 5
+      until: >-
+        sentry_consumer_running_after.rc == 0
+        and (sentry_consumer_running_after.stdout | trim) == 'true'
+      check_mode: false
+      changed_when: false
+      when: not ansible_check_mode
+
+    - name: Verify exact Sentry profiling consumer is healthy after apply
+      ansible.builtin.command:
+        argv:
+          - docker
+          - inspect
+          - --format
+          - "{{ '{{' }}.State.Health.Status{{ '}}' }}"
+          - "{{ sentry_profiling_consumer }}"
+      register: sentry_consumer_health_after
+      retries: 12
+      delay: 5
+      until: >-
+        sentry_consumer_health_after.rc == 0
+        and (sentry_consumer_health_after.stdout | trim) == 'healthy'
+      check_mode: false
+      changed_when: false
+      when: not ansible_check_mode
diff --git a/k8s/awoooi-prod/15-service-registry-configmap.yaml b/k8s/awoooi-prod/15-service-registry-configmap.yaml
index 7a040908d..df9b37c7f 100644
--- a/k8s/awoooi-prod/15-service-registry-configmap.yaml
+++ b/k8s/awoooi-prod/15-service-registry-configmap.yaml
@@ -1,9 +1,5 @@
-# k8s/awoooi-prod/15-service-registry-configmap.yaml
-# Service Registry ConfigMap — 掛載 ops/config/service-registry.yaml 到 K8s 容器
-# 建立: 2026-04-08 Claude Sonnet 4.6
-# 目的: 解決 Docker 容器無法找到 service-registry.yaml 導致 Guardrail 降級 AUTO 問題
-# 掛載路徑: /app/ops/config/service-registry.yaml
-# 參考: ADR-062, ADR-063
+# Generated by scripts/ops/render-service-registry-configmap.py.
+# Source: ops/config/service-registry.yaml; do not edit this file by hand.
 apiVersion: v1
 kind: ConfigMap
 metadata:
@@ -12,13 +8,15 @@ metadata:
   labels:
     app: awoooi
     component: service-registry
+  annotations:
+    awoooi.wooo.work/source-sha256: "1662532ccf9c574e9853291a6f2881377c1a5b2519c4123ae05caf45be904121"
 data:
   service-registry.yaml: |
     # ops/config/service-registry.yaml
     # Service Registry — 服務 Stateful 分級清單
-    # 版本: 1.0.0
+    # 版本: 2.0.0
     # 建立: Claude Sonnet 4.6 / 2026-04-08 Asia/Taipei
-    # 維護: 修改需 PR + 統帥審核，禁止直接 push
+    # 維護: Gitea main + controlled CD；runtime ConfigMap 必須由 renderer 精確產生
     # 說明:
     #   BLOCK        = 系統禁止自動修復，僅告警（資料風險最高）
     #   CRITICAL_HITL = 允許 Playbook，但需 MultiSig 2票
@@ -26,12 +24,21 @@ data:
     #   AUTO         = 允許自動執行（無狀態服務）
     # 參考: ADR-062, ADR-063
 
+    schema_version: service_registry_v2
+    governance_version: global_product_governance_v2
+    identity_policy:
+      match_mode: exact_alias_only
+      unknown_asset_terminal: asset_identity_unresolved
+      unknown_asset_auto_fallback_allowed: false
+      cross_domain_fallback_allowed: false
+
     services:
       # ─── BLOCK：系統禁止（連 Playbook 都不提供）────────────────────────────
       - name: postgres
         display_name: "PostgreSQL 主庫 (awoooi_prod)"
         host: "192.168.0.188"
         stateful_level: BLOCK
+        asset_domain: database
         reason: "主要業務資料庫，重啟可能導致 WAL 截斷、事務回滾"
         alert_only: true
         containers: ["postgres"]
@@ -40,6 +47,7 @@ data:
         display_name: "PostgreSQL (momo_db)"
         host: "192.168.0.188"
         stateful_level: BLOCK
+        asset_domain: database
         reason: "momo 產品資料庫，禁止自動操作"
         alert_only: true
         containers: ["momo-db"]
@@ -48,6 +56,7 @@ data:
         display_name: "PostgreSQL (Langfuse)"
         host: "192.168.0.110"
         stateful_level: BLOCK
+        asset_domain: database
         reason: "LLM trace 資料庫，重啟導致追蹤資料遺失"
         alert_only: true
         containers: ["langfuse-db"]
@@ -56,6 +65,7 @@ data:
         display_name: "PostgreSQL (Harbor Registry)"
         host: "192.168.0.110"
         stateful_level: BLOCK
+        asset_domain: database
         reason: "Harbor Registry 資料庫，重啟可能損壞 image layer 索引"
         alert_only: true
         containers: ["harbor-db"]
@@ -64,6 +74,7 @@ data:
         display_name: "PostgreSQL (Sentry)"
         host: "192.168.0.110"
         stateful_level: BLOCK
+        asset_domain: database
         reason: "Sentry 錯誤追蹤資料庫"
         alert_only: true
         containers: ["sentry-postgres"]
@@ -72,6 +83,7 @@ data:
         display_name: "ClickHouse (SignOz)"
         host: "192.168.0.110"
         stateful_level: BLOCK
+        asset_domain: database
         reason: "列欄式 OLAP 資料庫；只允許具備 native backup、bounded apply、rollback 與 post-verifier 的受控重建"
         alert_only: true
         containers: ["signoz-clickhouse"]
@@ -81,6 +93,7 @@ data:
         display_name: "Redis (AWOOOI)"
         host: "192.168.0.188"
         stateful_level: CRITICAL_HITL
+        asset_domain: database
         reason: "AWOOOI 依賴 Redis 做冪等鎖與快取，重啟丟失鎖狀態"
         requires_pre_backup: false
         containers: ["redis"]
@@ -89,6 +102,7 @@ data:
         display_name: "Redis (Harbor)"
         host: "192.168.0.110"
         stateful_level: CRITICAL_HITL
+        asset_domain: database
         reason: "Harbor session 快取"
         containers: ["harbor-redis"]
 
@@ -96,6 +110,7 @@ data:
         display_name: "Redis (Sentry)"
         host: "192.168.0.110"
         stateful_level: CRITICAL_HITL
+        asset_domain: database
         reason: "Sentry 任務佇列"
         containers: ["sentry-redis"]
 
@@ -119,6 +134,7 @@ data:
         display_name: "MinIO (物件存儲)"
         host: "192.168.0.188"
         stateful_level: CRITICAL_HITL
+        asset_domain: storage
         reason: "寫入中重啟可能導致 multipart upload 中斷"
         requires_pre_backup: false
         containers: ["minio"]
@@ -159,12 +175,14 @@ data:
         display_name: "AWOOOI API (K3s)"
         host: "k3s"
         stateful_level: AUTO
+        asset_domain: kubernetes_workload
         containers: []
 
       - name: awoooi-web
         display_name: "AWOOOI Web (K3s)"
         host: "k3s"
         stateful_level: AUTO
+        asset_domain: kubernetes_workload
         containers: []
 
       - name: blackbox-exporter
@@ -173,12 +191,39 @@ data:
         stateful_level: AUTO
         containers: ["blackbox-exporter"]
 
+      - name: sentry
+        display_name: "Sentry (錯誤追蹤)"
+        host: "192.168.0.110"
+        stateful_level: AUTO
+        asset_domain: docker_container
+        reason: "Sentry container 只能走該 host 的 exact-container bounded PlayBook；禁止 compose-wide 或跨 host fallback"
+        containers:
+          - sentry-web
+          - sentry-worker
+          - sentry-cron
+          - sentry-self-hosted-snuba-profiling-functions-consumer-1
+        allowed_catalog_ids:
+          - ansible:110-sentry-profiling-consumer-recovery
+
       - name: langfuse
         display_name: "Langfuse (LLMOps)"
         host: "192.168.0.110"
         stateful_level: AUTO
         containers: ["langfuse-web", "langfuse-worker"]
 
+      - name: signoz
+        display_name: "SignOz (APM)"
+        host: "192.168.0.188"
+        stateful_level: AUTO
+        reason: "APM 無狀態查詢層，docker compose up -d 可恢復（ClickHouse 資料不受影響）"
+        containers: ["signoz"]
+
+      - name: bitan-app
+        display_name: "Bitan App"
+        host: "192.168.0.188"
+        stateful_level: AUTO
+        containers: ["bitan-app"]
+
       - name: ollama
         display_name: "Ollama (Local LLM)"
         host: "192.168.0.188"
@@ -205,13 +250,13 @@ data:
 
     # ─── 備份策略參考 ────────────────────────────────────────────────────────
     backup_policies:
-      velero_max_age_hours: 4
-      emergency_backup_timeout: 600
-      block_backup_on_high_io: true
+      velero_max_age_hours: 4       # Velero 備份過期閾值（Q2 決策）
+      emergency_backup_timeout: 600  # 緊急備份超時秒數
+      block_backup_on_high_io: true  # CPU/IO > 80% 時禁止觸發備份（Q4 決策）
       io_threshold_percent: 80
 
     # ─── MultiSig 設定 ───────────────────────────────────────────────────────
     multisig:
-      critical_required_votes: 2
-      standard_required_votes: 1
-      vote_expiry_minutes: 30
+      critical_required_votes: 2   # CRITICAL_HITL 需要幾票
+      standard_required_votes: 1   # STANDARD_HITL 需要幾票
+      vote_expiry_minutes: 30       # 投票有效期
diff --git a/ops/config/service-registry.yaml b/ops/config/service-registry.yaml
index 262520d69..a40397fed 100644
--- a/ops/config/service-registry.yaml
+++ b/ops/config/service-registry.yaml
@@ -1,8 +1,8 @@
 # ops/config/service-registry.yaml
 # Service Registry — 服務 Stateful 分級清單
-# 版本: 1.0.0
+# 版本: 2.0.0
 # 建立: Claude Sonnet 4.6 / 2026-04-08 Asia/Taipei
-# 維護: 修改需 PR + 統帥審核，禁止直接 push
+# 維護: Gitea main + controlled CD；runtime ConfigMap 必須由 renderer 精確產生
 # 說明:
 #   BLOCK        = 系統禁止自動修復，僅告警（資料風險最高）
 #   CRITICAL_HITL = 允許 Playbook，但需 MultiSig 2票
@@ -10,12 +10,21 @@
 #   AUTO         = 允許自動執行（無狀態服務）
 # 參考: ADR-062, ADR-063
 
+schema_version: service_registry_v2
+governance_version: global_product_governance_v2
+identity_policy:
+  match_mode: exact_alias_only
+  unknown_asset_terminal: asset_identity_unresolved
+  unknown_asset_auto_fallback_allowed: false
+  cross_domain_fallback_allowed: false
+
 services:
   # ─── BLOCK：系統禁止（連 Playbook 都不提供）────────────────────────────
   - name: postgres
     display_name: "PostgreSQL 主庫 (awoooi_prod)"
     host: "192.168.0.188"
     stateful_level: BLOCK
+    asset_domain: database
     reason: "主要業務資料庫，重啟可能導致 WAL 截斷、事務回滾"
     alert_only: true
     containers: ["postgres"]
@@ -24,6 +33,7 @@ services:
     display_name: "PostgreSQL (momo_db)"
     host: "192.168.0.188"
     stateful_level: BLOCK
+    asset_domain: database
     reason: "momo 產品資料庫，禁止自動操作"
     alert_only: true
     containers: ["momo-db"]
@@ -32,6 +42,7 @@ services:
     display_name: "PostgreSQL (Langfuse)"
     host: "192.168.0.110"
     stateful_level: BLOCK
+    asset_domain: database
     reason: "LLM trace 資料庫，重啟導致追蹤資料遺失"
     alert_only: true
     containers: ["langfuse-db"]
@@ -40,6 +51,7 @@ services:
     display_name: "PostgreSQL (Harbor Registry)"
     host: "192.168.0.110"
     stateful_level: BLOCK
+    asset_domain: database
     reason: "Harbor Registry 資料庫，重啟可能損壞 image layer 索引"
     alert_only: true
     containers: ["harbor-db"]
@@ -48,6 +60,7 @@ services:
     display_name: "PostgreSQL (Sentry)"
     host: "192.168.0.110"
     stateful_level: BLOCK
+    asset_domain: database
     reason: "Sentry 錯誤追蹤資料庫"
     alert_only: true
     containers: ["sentry-postgres"]
@@ -56,6 +69,7 @@ services:
     display_name: "ClickHouse (SignOz)"
     host: "192.168.0.110"
     stateful_level: BLOCK
+    asset_domain: database
     reason: "列欄式 OLAP 資料庫；只允許具備 native backup、bounded apply、rollback 與 post-verifier 的受控重建"
     alert_only: true
     containers: ["signoz-clickhouse"]
@@ -65,6 +79,7 @@ services:
     display_name: "Redis (AWOOOI)"
     host: "192.168.0.188"
     stateful_level: CRITICAL_HITL
+    asset_domain: database
     reason: "AWOOOI 依賴 Redis 做冪等鎖與快取，重啟丟失鎖狀態"
     requires_pre_backup: false
     containers: ["redis"]
@@ -73,6 +88,7 @@ services:
     display_name: "Redis (Harbor)"
     host: "192.168.0.110"
     stateful_level: CRITICAL_HITL
+    asset_domain: database
     reason: "Harbor session 快取"
     containers: ["harbor-redis"]
 
@@ -80,6 +96,7 @@ services:
     display_name: "Redis (Sentry)"
     host: "192.168.0.110"
     stateful_level: CRITICAL_HITL
+    asset_domain: database
     reason: "Sentry 任務佇列"
     containers: ["sentry-redis"]
 
@@ -103,6 +120,7 @@ services:
     display_name: "MinIO (物件存儲)"
     host: "192.168.0.188"
     stateful_level: CRITICAL_HITL
+    asset_domain: storage
     reason: "寫入中重啟可能導致 multipart upload 中斷"
     requires_pre_backup: false
     containers: ["minio"]
@@ -143,12 +161,14 @@ services:
     display_name: "AWOOOI API (K3s)"
     host: "k3s"
     stateful_level: AUTO
+    asset_domain: kubernetes_workload
     containers: []
 
   - name: awoooi-web
     display_name: "AWOOOI Web (K3s)"
     host: "k3s"
     stateful_level: AUTO
+    asset_domain: kubernetes_workload
     containers: []
 
   - name: blackbox-exporter
@@ -161,8 +181,15 @@ services:
     display_name: "Sentry (錯誤追蹤)"
     host: "192.168.0.110"
     stateful_level: AUTO
-    reason: "Web server 無狀態，docker compose up -d 即可恢復"
-    containers: ["sentry-web", "sentry-worker", "sentry-cron"]
+    asset_domain: docker_container
+    reason: "Sentry container 只能走該 host 的 exact-container bounded PlayBook；禁止 compose-wide 或跨 host fallback"
+    containers:
+      - sentry-web
+      - sentry-worker
+      - sentry-cron
+      - sentry-self-hosted-snuba-profiling-functions-consumer-1
+    allowed_catalog_ids:
+      - ansible:110-sentry-profiling-consumer-recovery
 
   - name: langfuse
     display_name: "Langfuse (LLMOps)"
diff --git a/scripts/ops/render-service-registry-configmap.py b/scripts/ops/render-service-registry-configmap.py
new file mode 100644
index 000000000..232b62788
--- /dev/null
+++ b/scripts/ops/render-service-registry-configmap.py
@@ -0,0 +1,67 @@
+#!/usr/bin/env python3
+"""Render the production Service Registry ConfigMap from one canonical source."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+from pathlib import Path
+
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+SOURCE = REPO_ROOT / "ops" / "config" / "service-registry.yaml"
+TARGET = REPO_ROOT / "k8s" / "awoooi-prod" / "15-service-registry-configmap.yaml"
+
+
+def render(source_text: str) -> str:
+    if not source_text.endswith("\n"):
+        source_text += "\n"
+    digest = hashlib.sha256(source_text.encode("utf-8")).hexdigest()
+    embedded = "".join(
+        f"    {line}" if line.strip() else line
+        for line in source_text.splitlines(keepends=True)
+    )
+    return (
+        "# Generated by scripts/ops/render-service-registry-configmap.py.\n"
+        "# Source: ops/config/service-registry.yaml; do not edit this file by hand.\n"
+        "apiVersion: v1\n"
+        "kind: ConfigMap\n"
+        "metadata:\n"
+        "  name: service-registry\n"
+        "  namespace: awoooi-prod\n"
+        "  labels:\n"
+        "    app: awoooi\n"
+        "    component: service-registry\n"
+        "  annotations:\n"
+        f"    awoooi.wooo.work/source-sha256: \"{digest}\"\n"
+        "data:\n"
+        "  service-registry.yaml: |\n"
+        f"{embedded}"
+    )
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "--check",
+        action="store_true",
+        help="fail if the committed ConfigMap is not an exact render",
+    )
+    args = parser.parse_args()
+
+    expected = render(SOURCE.read_text(encoding="utf-8"))
+    if args.check:
+        actual = TARGET.read_text(encoding="utf-8") if TARGET.exists() else ""
+        if actual != expected:
+            raise SystemExit(
+                "service registry ConfigMap drift: run "
+                "scripts/ops/render-service-registry-configmap.py"
+            )
+        return 0
+
+    TARGET.write_text(expected, encoding="utf-8")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())