diff --git a/apps/api/src/api/v1/auto_repair.py b/apps/api/src/api/v1/auto_repair.py
index cb808aa05..fd75b3e06 100644
--- a/apps/api/src/api/v1/auto_repair.py
+++ b/apps/api/src/api/v1/auto_repair.py
@@ -13,6 +13,8 @@ Phase 8.2: API Router 實作
 - 業務邏輯委託給 Service 層
 """
 
+from typing import Literal
+
 from fastapi import APIRouter, HTTPException, Query
 from pydantic import BaseModel, Field
 
@@ -62,6 +64,11 @@ class ExecuteResponse(BaseModel):
     """執行自動修復回應"""
 
     success: bool = Field(description="是否執行成功")
+    status: Literal["completed", "queued", "failed"] = Field(
+        description="completed | queued | failed"
+    )
+    repair_executed: bool = Field(description="是否已發生 runtime 修復")
+    repair_verified: bool = Field(description="是否已有獨立 verifier 成功證據")
     incident_id: str = Field(description="Incident ID")
     playbook_id: str = Field(description="Playbook ID")
     executed_steps: list[str] = Field(description="已執行的步驟")
@@ -164,6 +171,9 @@ async def execute_auto_repair(request: ExecuteRequest, _csrf_token: CSRFToken) -
 
     return ExecuteResponse(
         success=result.success,
+        status=result.status,
+        repair_executed=result.repair_executed,
+        repair_verified=result.repair_verified,
         incident_id=result.incident_id,
         playbook_id=result.playbook_id,
         executed_steps=result.executed_steps,
diff --git a/apps/api/src/plugins/mcp/providers/ssh_provider.py b/apps/api/src/plugins/mcp/providers/ssh_provider.py
index f5bca9313..4f144d6f9 100644
--- a/apps/api/src/plugins/mcp/providers/ssh_provider.py
+++ b/apps/api/src/plugins/mcp/providers/ssh_provider.py
@@ -19,10 +19,10 @@ SSH MCP Tool Provider — MCP Phase 2a
   群組 B (6 個安全操作工具，需 trust_score >= 0.8)
     ssh_docker_restart        — DockerContainerExited / HarborDown
     ssh_docker_compose_restart — SentryDown / SignOzDown / GiteaDown
-    ssh_systemctl_restart     — OllamaDown / KaliScannerDown
+    ssh_systemctl_restart     — compatibility name; direct execution disabled
     ssh_clear_docker_logs     — HostOutOfDiskSpace (log 佔用)
     ssh_renew_ssl             — TLSCertExpiringIn7Days
-    ssh_reload_nginx          — TLSProbeFailure / conf 更新後
+    ssh_reload_nginx          — compatibility name; direct execution disabled
 
 四層安全守衛 (缺一不可):
   1. tool_name 必須在白名單
@@ -358,7 +358,10 @@ class SSHProvider(MCPToolProvider):
             ),
             MCPTool(
                 name="ssh_systemctl_restart",
-                description="Restart a systemd service (systemctl restart <service>). Requires trust_score >= 0.8.",
+                description=(
+                    "Compatibility-only name. Direct SSH execution is disabled; "
+                    "use the incident-bound Host Ansible controlled executor."
+                ),
                 input_schema={"type": "object", "properties": {
                     "host": {"type": "string"},
                     "service": {"type": "string"},
@@ -388,7 +391,10 @@ class SSHProvider(MCPToolProvider):
             ),
             MCPTool(
                 name="ssh_reload_nginx",
-                description="Test and reload nginx config (nginx -t && systemctl reload nginx). Requires trust_score >= 0.8.",
+                description=(
+                    "Compatibility-only name. Direct reload is disabled; use "
+                    "the incident-bound Host Ansible controlled executor."
+                ),
                 input_schema={"type": "object", "properties": {
                     "host": {"type": "string"},
                     "trust_score": {"type": "number"},
@@ -430,6 +436,19 @@ class SSHProvider(MCPToolProvider):
                 error=f"Unknown tool: {tool_name}",
             )
 
+        # Host/systemd mutation has one writer: the typed Ansible broker. Keep
+        # the legacy tool name discoverable for compatibility, but never turn
+        # it into a raw SSH command.
+        if tool_name in {"ssh_systemctl_restart", "ssh_reload_nginx"}:
+            return MCPToolResult(
+                success=False,
+                execution_id=execution_id,
+                error=(
+                    f"{tool_name}_disabled_use_"
+                    "host_ansible_controlled_executor"
+                ),
+            )
+
         host = _normalize_ssh_host(str(parameters.get("host", "")))
 
         # 守衛 2: 允許的 host
@@ -553,6 +572,11 @@ class SSHProvider(MCPToolProvider):
 
     def _build_command(self, tool_name: str, params: dict) -> str:
         # 所有接受用戶字串的工具，必須先通過 _validate_param() 白名單驗證
+        if tool_name in {"ssh_systemctl_restart", "ssh_reload_nginx"}:
+            raise ValueError(
+                f"{tool_name}_disabled_use_host_ansible_controlled_executor"
+            )
+
         if tool_name == "ssh_diagnose":
             # 2026-04-27 Claude Sonnet 4.6: 主機告警自動診斷 — 只讀，不修改任何狀態
             command = (
@@ -618,10 +642,6 @@ class SSHProvider(MCPToolProvider):
             service = _validate_param("service", params["service"])
             return f"cd {compose_dir} && docker compose restart {service}"
 
-        if tool_name == "ssh_systemctl_restart":
-            svc = _validate_param("service", params["service"])
-            return f"systemctl restart {svc}"
-
         if tool_name == "ssh_clear_docker_logs":
             name = _validate_param("container_name", params["container_name"])
             # 透過 docker inspect 取得 log 路徑，再截斷
@@ -635,9 +655,6 @@ class SSHProvider(MCPToolProvider):
             domain = _validate_param("domain", params["domain"])
             return f"/snap/bin/certbot renew --cert-name {domain} --non-interactive 2>&1"
 
-        if tool_name == "ssh_reload_nginx":
-            return "nginx -t 2>&1 && systemctl reload nginx && echo 'Nginx reloaded'"
-
         if tool_name == "ssh_docker_prune":
             # Disk-gated docker prune: only acts when the alerting condition still holds.
             # 2026-05-02 ogt + Claude Sonnet 4.6: ADR-068 飛輪 — disk full SOP
diff --git a/apps/api/src/services/approval_db.py b/apps/api/src/services/approval_db.py
index 3fa61956b..75402e956 100644
--- a/apps/api/src/services/approval_db.py
+++ b/apps/api/src/services/approval_db.py
@@ -955,6 +955,64 @@ class ApprovalDBService:
                 execution_kind=execution_kind,
             )
 
+    async def record_controlled_handoff_pending(
+        self,
+        approval_id: UUID,
+        *,
+        execution_kind: str,
+        receipt: dict[str, Any],
+    ) -> bool:
+        """Persist queue acceptance without claiming terminal execution."""
+
+        async with get_db_context() as db:
+            result = await db.execute(
+                select(ApprovalRecord)
+                .where(ApprovalRecord.id == str(approval_id))
+                .with_for_update()
+            )
+            record = result.scalar_one_or_none()
+            if record is None:
+                logger.warning(
+                    "approval_controlled_handoff_missing",
+                    id=str(approval_id),
+                )
+                return False
+
+            status_value = _record_value(record.status)
+            if status_value != ApprovalStatus.APPROVED.value:
+                logger.warning(
+                    "approval_controlled_handoff_status_rejected",
+                    id=str(approval_id),
+                    status=status_value,
+                )
+                return False
+
+            metadata = dict(record.extra_metadata or {})
+            metadata.update(
+                {
+                    "execution_kind": execution_kind,
+                    "execution_state": "queued",
+                    "repair_attempted": False,
+                    "repair_executed": False,
+                    "repair_verified": False,
+                    "incident_closure_allowed": False,
+                    "automation_run_id": receipt.get("automation_run_id"),
+                    "single_writer_executor": receipt.get(
+                        "single_writer_executor"
+                    ),
+                }
+            )
+            record.extra_metadata = metadata
+            await db.flush()
+            logger.info(
+                "approval_controlled_handoff_recorded",
+                id=str(approval_id),
+                execution_kind=execution_kind,
+                automation_run_id=receipt.get("automation_run_id"),
+                status_preserved=status_value,
+            )
+            return True
+
     async def update_incident_id(self, approval_id: UUID, incident_id: str) -> None:
         """
         2026-04-06 ogt: Phase 26 — 回寫 incident_id 到 approval_records
diff --git a/apps/api/src/services/approval_execution.py b/apps/api/src/services/approval_execution.py
index 43d471890..45cecdb8b 100644
--- a/apps/api/src/services/approval_execution.py
+++ b/apps/api/src/services/approval_execution.py
@@ -73,10 +73,8 @@ _SSH_GATEWAY_TOOL_SCOPES: dict[str, str] = {
     "ssh_diagnose": "read",
     "ssh_docker_restart": "write",
     "ssh_docker_compose_restart": "write",
-    "ssh_systemctl_restart": "write",
     "ssh_clear_docker_logs": "write",
     "ssh_renew_ssl": "write",
-    "ssh_reload_nginx": "write",
     "ssh_docker_prune": "admin",
 }
 
@@ -672,6 +670,82 @@ class ApprovalExecutionService:
                 )
                 attempt += 1
 
+        response = (
+            result.k8s_response
+            if isinstance(result.k8s_response, dict)
+            else {}
+        )
+        controlled_handoff = (
+            response.get("controlled_handoff")
+            if isinstance(response.get("controlled_handoff"), dict)
+            else {}
+        )
+        if (
+            response.get("controlled_execution_queued") is True
+            and controlled_handoff.get("queued") is True
+        ):
+            execution_kind = "host_ansible_check_mode_queued"
+            duration_ms = int((time.time() - _aol_started_ms) * 1000)
+            await service.record_controlled_handoff_pending(
+                approval.id,
+                execution_kind=execution_kind,
+                receipt=controlled_handoff,
+            )
+            await timeline.add_event(
+                event_type="exec",
+                status="pending",
+                title="⏳ 已排入 Host Ansible check-mode（尚未修復）",
+                description=(
+                    "Typed route accepted. Runtime apply and independent "
+                    "verification remain pending."
+                ),
+                actor="leWOOOgo",
+                actor_role="executor",
+                approval_id=str(approval.id),
+                incident_id=approval.incident_id,
+            )
+            receipt = {
+                "automation_run_id": controlled_handoff.get(
+                    "automation_run_id"
+                ),
+                "single_writer_executor": controlled_handoff.get(
+                    "single_writer_executor"
+                ),
+                "runtime_write_performed": False,
+                "repair_executed": False,
+                "repair_verified": False,
+                "incident_closure_allowed": False,
+            }
+            await self._log_aol_completed(
+                op_id=_aol_op_id,
+                status="dry_run",
+                duration_ms=duration_ms,
+                output={
+                    "execution_kind": execution_kind,
+                    **receipt,
+                },
+            )
+            await self._push_execution_result_to_alert(
+                approval,
+                success=False,
+                error=None,
+                execution_kind=execution_kind,
+                repair_executed=False,
+                repair_attempted=False,
+            )
+            logger.info(
+                "background_execution_host_ansible_queued",
+                approval_id=str(approval.id),
+                incident_id=approval.incident_id,
+                automation_run_id=controlled_handoff.get("automation_run_id"),
+                runtime_write_performed=False,
+                incident_resolved=False,
+            )
+            # Bool return preserves the historical "runtime execution
+            # completed" contract. Queue acceptance is intentionally False;
+            # durable approval metadata above records the pending handoff.
+            return False
+
         execution_kind = self._execution_kind_for_result(operation_type, result)
         repair_kind = self._is_repair_kind(execution_kind)
         repair_executed = bool(result.success and repair_kind)
@@ -1012,12 +1086,13 @@ class ApprovalExecutionService:
 
         2026-05-02 ogt + Claude Sonnet 4.6: 修補手動批准 SSH action 卡住的 bug
         根因：parse_operation_from_action 只懂 kubectl，approval_execution 走 K8s executor 拒收
-        修法：偵測 SSH_HOST 後改走 SSHProvider，行為與 decision_manager._ssh_execute 對齊
+        修法：唯讀與 container legacy action 仍走 MCP；Host/systemd mutation
+        一律交給 typed Ansible controlled executor。
 
         action 解析邏輯：
         - "docker prune" / "docker image prune" / "docker volume prune" → ssh_docker_prune
         - "docker restart <name>" → ssh_docker_restart
-        - "systemctl restart <svc>" → ssh_systemctl_restart
+        - "systemctl restart <svc>" → Host Ansible controlled queue
         - "ps aux" / "df -h" / "free -h" / "top" / "uptime" / 'echo' / 'ls -lah' → ssh_diagnose
         - 其他：回傳失敗，提示 LLM 改寫 action
         """
@@ -1029,6 +1104,69 @@ class ApprovalExecutionService:
         params: dict = {"host": host}
         tool_name: str | None = None
 
+        from src.services.auto_repair_service import (
+            _is_host_service_command,
+            _parse_bounded_systemd_mutation,
+        )
+
+        systemd_request = _parse_bounded_systemd_mutation(action)
+        if systemd_request is not None or _is_host_service_command(action):
+            if systemd_request is None or systemd_request.action != "restart":
+                duration_ms = int((time.time() - start) * 1000)
+                return ExecutionResult(
+                    success=False,
+                    message="host_ansible_controlled_executor blocked",
+                    operation_type=OperationType.SSH_HOST,
+                    target_resource=host,
+                    namespace="host",
+                    duration_ms=duration_ms,
+                    error=(
+                        "host_ansible_controlled_executor:"
+                        "unbounded_or_unallowlisted_systemd_mutation"
+                    ),
+                )
+
+            from src.services.host_ansible_controlled_executor import (
+                queue_host_ansible_controlled_action_for_incident_id,
+            )
+
+            handoff = await queue_host_ansible_controlled_action_for_incident_id(
+                incident_id=str(approval.incident_id or ""),
+                requested_action=(
+                    "reconcile host service "
+                    f"{systemd_request.service}"
+                ),
+                risk_level=getattr(approval, "risk_level", None),
+                source="manual_approval_host_systemd",
+                requested_host=systemd_request.requested_host or host,
+                approval_id=str(approval.id),
+            )
+            duration_ms = int((time.time() - start) * 1000)
+            queued = handoff.get("queued") is True
+            return ExecutionResult(
+                success=queued,
+                message=(
+                    "host_ansible_controlled_executor queued"
+                    if queued
+                    else "host_ansible_controlled_executor blocked"
+                ),
+                operation_type=OperationType.SSH_HOST,
+                target_resource=host,
+                namespace="host",
+                duration_ms=duration_ms,
+                k8s_response={
+                    "controlled_execution_queued": queued,
+                    "runtime_write_performed": False,
+                    "repair_executed": False,
+                    "repair_verified": False,
+                    "controlled_handoff": handoff,
+                },
+                error=None if queued else str(
+                    handoff.get("status")
+                    or "host_ansible_controlled_executor_blocked"
+                ),
+            )
+
         if "docker" in action_lower and "prune" in action_lower:
             tool_name = "ssh_docker_prune"
             params["trust_score"] = 0.85
@@ -1042,15 +1180,6 @@ class ApprovalExecutionService:
                 params["trust_score"] = 0.85
             else:
                 tool_name = None  # 沒抓到 container 名稱，降級
-        elif "systemctl restart" in action_lower:
-            tool_name = "ssh_systemctl_restart"
-            import re as _re
-            m = _re.search(r"systemctl\s+restart\s+([a-z0-9._-]+)", action_lower)
-            if m:
-                params["service"] = m.group(1)
-                params["trust_score"] = 0.85
-            else:
-                tool_name = None
         elif any(kw in action_lower for kw in ("ps aux", "df -h", "free -h", "top ", "uptime", "echo ", "ls -")):
             # 主機診斷類（合 ssh_diagnose 一鍵收集）
             tool_name = "ssh_diagnose"
@@ -1313,6 +1442,9 @@ class ApprovalExecutionService:
 
             outcome = await self._fetch_operator_outcome_for_result(approval)
             no_action = success and is_no_action_approval_action(approval.action)
+            controlled_handoff_queued = (
+                execution_kind == "host_ansible_check_mode_queued"
+            )
             text = self._format_execution_result_message(
                 approval=approval,
                 success=success,
@@ -1320,6 +1452,7 @@ class ApprovalExecutionService:
                 no_action=no_action,
                 km_info=km_info,
                 outcome=outcome,
+                execution_kind=execution_kind,
             )
             truth_chain_row = incident_truth_chain_button_row(
                 approval.incident_id or ""
@@ -1388,11 +1521,17 @@ class ApprovalExecutionService:
                     approval_id=str(approval.id),
                     actor="approval_execution",
                     action_detail=(
-                        "telegram_execution_result_sent"
+                        "telegram_controlled_handoff_queued_sent"
+                        if controlled_handoff_queued and delivery_succeeded
+                        else "telegram_execution_result_sent"
                         if delivery_succeeded
                         else "telegram_execution_result_no_send"
                     ),
-                    success=(success if delivery_succeeded else False),
+                    success=(
+                        delivery_succeeded
+                        if controlled_handoff_queued
+                        else success if delivery_succeeded else False
+                    ),
                     error_message=(
                         error
                         if delivery_succeeded
@@ -1407,7 +1546,10 @@ class ApprovalExecutionService:
                         "delivery": "reply" if orig_msg_id is not None else "standalone",
                         "delivery_status": delivery_status,
                         "provider_delivery_acknowledged": delivery_succeeded,
-                        "execution_success": success,
+                        "execution_success": (
+                            False if controlled_handoff_queued else success
+                        ),
+                        "queue_accepted": controlled_handoff_queued,
                     },
                 )
             except Exception as _log_e:
@@ -1478,10 +1620,14 @@ class ApprovalExecutionService:
         no_action: bool,
         km_info: str,
         outcome: dict[str, Any],
+        execution_kind: str | None = None,
     ) -> str:
         state = str(outcome.get("state") or "unknown")
         needs_human = bool(outcome.get("needs_human"))
-        if no_action:
+        if execution_kind == "host_ansible_check_mode_queued":
+            icon = "⏳"
+            title = "已排入 Ansible check-mode，尚未修復或驗證"
+        elif no_action:
             icon = "ℹ️"
             title = "已記錄觀察，未執行修復"
         elif not success:
diff --git a/apps/api/src/services/auto_repair_service.py b/apps/api/src/services/auto_repair_service.py
index 614cfc56f..d2906cb6c 100644
--- a/apps/api/src/services/auto_repair_service.py
+++ b/apps/api/src/services/auto_repair_service.py
@@ -23,9 +23,11 @@ Phase 8: 自動化層實作
 """
 
 import re
+import shlex
 from collections.abc import Callable
 from dataclasses import dataclass
 from typing import Any, Protocol
+from urllib.parse import unquote, urlsplit
 from uuid import NAMESPACE_URL, UUID, uuid5
 
 import structlog
@@ -88,6 +90,9 @@ class AutoRepairResult:
     playbook_id: str
     incident_id: str
     executed_steps: list[str]
+    status: str = "completed"
+    repair_executed: bool = False
+    repair_verified: bool = False
     error: str | None = None
     execution_time_ms: int = 0
 
@@ -125,6 +130,153 @@ _SSH_DIAGNOSTIC_KEYWORDS = (
     "uptime",
 )
 
+_SYSTEMD_ACTION_RE = re.compile(
+    r"^\s*(?:sudo\s+)?systemctl\s+"
+    r"(?P<action>restart|start|stop|reload)\s+"
+    r"(?P<service>[a-zA-Z0-9@_][a-zA-Z0-9@._-]*)\s*$",
+    re.IGNORECASE,
+)
+_SYSTEMD_COMMAND_RE = re.compile(r"\bsystemctl\b", re.IGNORECASE)
+_LEGACY_SERVICE_COMMAND_RE = re.compile(
+    r"(?:\b(?:service|rc-service)\s+[a-zA-Z0-9@._-]+\s+"
+    r"(?:restart|start|stop|reload)\b|"
+    r"/etc/init\.d/[a-zA-Z0-9@._-]+\s+"
+    r"(?:restart|start|stop|reload)\b)",
+    re.IGNORECASE,
+)
+_SSH_OPTIONS_WITH_VALUE = frozenset(
+    {
+        "-b",
+        "-c",
+        "-D",
+        "-E",
+        "-e",
+        "-F",
+        "-I",
+        "-i",
+        "-J",
+        "-L",
+        "-l",
+        "-m",
+        "-O",
+        "-o",
+        "-p",
+        "-Q",
+        "-R",
+        "-S",
+        "-W",
+        "-w",
+    }
+)
+
+
+@dataclass(frozen=True)
+class _SystemdMutationRequest:
+    action: str
+    service: str
+    requested_host: str | None = None
+
+
+def _parse_systemd_body(
+    body: str,
+    *,
+    requested_host: str | None = None,
+) -> _SystemdMutationRequest | None:
+    text = str(body or "").strip()
+    if not text or any(
+        token in text for token in (";", "&&", "||", "$(", "`", "\n")
+    ):
+        return None
+    match = _SYSTEMD_ACTION_RE.fullmatch(text)
+    if match is None:
+        return None
+    return _SystemdMutationRequest(
+        action=match.group("action").lower(),
+        service=match.group("service"),
+        requested_host=requested_host,
+    )
+
+
+def _parse_bounded_systemd_mutation(
+    command: str,
+) -> _SystemdMutationRequest | None:
+    """Parse one shell-free systemd mutation and retain its explicit host."""
+
+    text = str(command or "").strip()
+    if not text:
+        return None
+
+    parsed_uri = urlsplit(text)
+    if parsed_uri.scheme.lower() == "ssh":
+        if parsed_uri.query or parsed_uri.fragment or not parsed_uri.hostname:
+            return None
+        return _parse_systemd_body(
+            unquote(parsed_uri.path).lstrip("/"),
+            requested_host=parsed_uri.hostname,
+        )
+
+    try:
+        tokens = shlex.split(text, posix=True)
+    except ValueError:
+        return None
+    if not tokens:
+        return None
+    if tokens[0].lower() != "ssh":
+        return _parse_systemd_body(text)
+
+    index = 1
+    while index < len(tokens) and tokens[index].startswith("-"):
+        option = tokens[index]
+        index += 1
+        if option in _SSH_OPTIONS_WITH_VALUE:
+            index += 1
+    if index >= len(tokens) - 1:
+        return None
+    destination = tokens[index]
+    if destination.startswith("-"):
+        return None
+    requested_host = destination.rsplit("@", 1)[-1]
+    if requested_host == "{host}":
+        requested_host = None
+    return _parse_systemd_body(
+        " ".join(tokens[index + 1 :]),
+        requested_host=requested_host,
+    )
+
+
+def _shell_normalized_variants(command: str) -> set[str]:
+    """Return bounded shell-token variants to expose quote/backslash aliases."""
+
+    variants = {unquote(str(command or ""))}
+    for _ in range(2):
+        for candidate in tuple(variants):
+            try:
+                tokens = shlex.split(candidate, posix=True)
+            except ValueError:
+                continue
+            variants.add(" ".join(tokens))
+            variants.update(token for token in tokens if token)
+    return variants
+
+
+def _is_host_service_command(command: str) -> bool:
+    """Detect system service commands even when shell aliases hide literals."""
+
+    return any(
+        _SYSTEMD_COMMAND_RE.search(candidate) is not None
+        or _LEGACY_SERVICE_COMMAND_RE.search(candidate) is not None
+        for candidate in _shell_normalized_variants(command)
+    )
+
+
+def _incident_requested_host(incident: Incident) -> str | None:
+    labels = incident.signals[0].labels if incident.signals else {}
+    for key in ("host", "instance", "node"):
+        value = str(labels.get(key) or "").strip()
+        if value:
+            return value
+    return None
+
 _SSH_WRITE_KEYWORDS = (
     "docker restart",
     "docker start",
@@ -557,20 +709,65 @@ class AutoRepairService:
             steps_count=len(playbook.repair_steps),
         )
 
-        # ADR-039: 記錄全域修復計數（用於熔斷檢查）
-        await record_global_repair_action()
-
         try:
+            host_steps = [
+                step
+                for step in playbook.repair_steps
+                if step.action_type == ActionType.SSH_COMMAND
+            ]
+            if host_steps and len(playbook.repair_steps) != 1:
+                raise RuntimeError(
+                    "cross_domain_or_multi_step_host_playbook_rejected"
+                )
+
+            global_repair_action_recorded = False
             # 執行每個步驟
             for step in playbook.repair_steps:
                 # 2026-04-07 Claude Code: 統帥指令「直接全部跳成自動修復」
                 # 移除 step-level 風險門檻，所有步驟直接執行
 
+                # A Host/systemd step only queues the single-writer broker and
+                # therefore must not consume the runtime repair circuit budget.
+                # Other executable steps retain the original pre-dispatch
+                # repair counter behavior.
+                if (
+                    not global_repair_action_recorded
+                    and not (
+                        step.action_type == ActionType.SSH_COMMAND
+                        and _is_host_service_command(step.command)
+                    )
+                ):
+                    await record_global_repair_action()
+                    global_repair_action_recorded = True
+
                 # 執行步驟
                 step_result = await self._execute_step(incident, step)
                 executed_steps.append(
                     f"Step {step.step_number}: {step.command[:50]}... -> {step_result}"
                 )
+                if self._is_step_queued_result(step_result):
+                    execution_time = int(
+                        (time.perf_counter() - start_time) * 1000
+                    )
+                    logger.info(
+                        "auto_repair_controlled_executor_queued",
+                        incident_id=incident.incident_id,
+                        playbook_id=playbook.playbook_id,
+                        execution_time_ms=execution_time,
+                        runtime_write_performed=False,
+                        repair_executed=False,
+                        incident_resolved=False,
+                    )
+                    return AutoRepairResult(
+                        success=False,
+                        status="queued",
+                        repair_executed=False,
+                        repair_verified=False,
+                        playbook_id=playbook.playbook_id,
+                        incident_id=incident.incident_id,
+                        executed_steps=executed_steps,
+                        execution_time_ms=execution_time,
+                    )
                 if self._is_step_failure_result(step_result):
                     raise RuntimeError(f"Step {step.step_number} failed: {step_result}")
 
@@ -592,6 +789,9 @@ class AutoRepairService:
 
             repair_result = AutoRepairResult(
                 success=True,
+                status="completed",
+                repair_executed=True,
+                repair_verified=False,
                 playbook_id=playbook.playbook_id,
                 incident_id=incident.incident_id,
                 executed_steps=executed_steps,
@@ -784,6 +984,9 @@ class AutoRepairService:
 
             fail_result = AutoRepairResult(
                 success=False,
+                status="failed",
+                repair_executed=False,
+                repair_verified=False,
                 playbook_id=playbook.playbook_id,
                 incident_id=incident.incident_id,
                 executed_steps=executed_steps,
@@ -907,6 +1110,12 @@ class AutoRepairService:
         normalized = (step_result or "").strip().upper()
         return normalized.startswith("FAILED:") or normalized == "UNKNOWN_ACTION_TYPE"
 
+    @staticmethod
+    def _is_step_queued_result(step_result: str) -> bool:
+        """A broker handoff is pending work, not repair success or failure."""
+
+        return (step_result or "").strip().upper().startswith("QUEUED:")
+
     def _get_max_risk_level(self, playbook: Playbook) -> RiskLevel:
         """取得 Playbook 中最高的風險等級"""
         risk_order = {
@@ -1554,7 +1763,50 @@ class AutoRepairService:
 
         # 2026-04-06 Claude Code: Sprint 3 — repair_by_uri (URI scheme 路由)
         if step.action_type == ActionType.SSH_COMMAND:
-            upstream_success = self._upstream_repair_success_result(incident, step.command)
+            systemd_request = _parse_bounded_systemd_mutation(step.command)
+            if systemd_request is not None:
+                if systemd_request.action != "restart":
+                    return (
+                        "FAILED: host_ansible_controlled_executor:"
+                        f"systemd_action_not_allowlisted:{systemd_request.action}"
+                    )
+                from src.services.host_ansible_controlled_executor import (
+                    queue_host_ansible_controlled_action,
+                )
+
+                handoff = await queue_host_ansible_controlled_action(
+                    incident=incident,
+                    requested_action=(
+                        "reconcile host service "
+                        f"{systemd_request.service}"
+                    ),
+                    risk_level=step.risk_level,
+                    source="auto_repair_host_systemd",
+                    requested_host=(
+                        systemd_request.requested_host
+                        or _incident_requested_host(incident)
+                    ),
+                )
+                if handoff.get("queued") is True:
+                    return (
+                        "QUEUED: host_ansible_executor:"
+                        f"{handoff.get('automation_run_id') or 'pending'}"
+                    )
+                return (
+                    "FAILED: host_ansible_controlled_executor:"
+                    f"{handoff.get('status') or 'queue_failed'}"
+                )
+
+            if _is_host_service_command(step.command):
+                return (
+                    "FAILED: host_ansible_controlled_executor:"
+                    "unbounded_systemd_mutation_rejected"
+                )
+
+            upstream_success = self._upstream_repair_success_result(
+                incident,
+                step.command,
+            )
             if upstream_success is not None:
                 return upstream_success
 
@@ -1571,13 +1823,17 @@ class AutoRepairService:
                     approved=approved,
                 )
 
-            from src.services.host_repair_agent import HostRepairAgent
-            agent = HostRepairAgent()
-            result = await agent.repair_by_uri(step.command, approved=approved)
-            if result.success:
-                return f"SUCCESS: {result.output}"
-            else:
-                return f"FAILED: {result.error}"
+            logger.warning(
+                "auto_repair_legacy_direct_host_executor_blocked",
+                incident_id=incident.incident_id,
+                command_scheme=(
+                    (step.command or "").split("://", 1)[0][:32]
+                    if "://" in (step.command or "")
+                    else "legacy_shell"
+                ),
+                runtime_write_performed=False,
+            )
+            return "FAILED: legacy_direct_host_executor_disabled_use_typed_executor"
 
         return "UNKNOWN_ACTION_TYPE"
 
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 cc18e539e..3aeabacc3 100644
--- a/apps/api/src/services/awooop_ansible_check_mode_service.py
+++ b/apps/api/src/services/awooop_ansible_check_mode_service.py
@@ -1378,9 +1378,12 @@ async def _finalize_controlled_approval_projection(
                 "automation_run_id": _automation_run_id_for_claim(claim),
                 "single_writer_executor": "awoooi-ansible-executor-broker",
                 "execution_kind": "controlled_apply",
+                "execution_state": "completed" if success else "failed",
                 "repair_attempted": True,
                 "repair_executed": result.returncode == 0,
+                "repair_verified": result.post_verifier_passed is True,
                 "post_verifier_passed": result.post_verifier_passed is True,
+                "incident_closure_allowed": success,
                 "apply_op_id": apply_op_id,
             })
             approval.extra_metadata = metadata
diff --git a/apps/api/src/services/awooop_truth_chain_service.py b/apps/api/src/services/awooop_truth_chain_service.py
index e6d938116..b891819de 100644
--- a/apps/api/src/services/awooop_truth_chain_service.py
+++ b/apps/api/src/services/awooop_truth_chain_service.py
@@ -758,13 +758,36 @@ def _approval_suppresses_repair_execution(approvals: list[dict[str, Any]]) -> bo
         if not isinstance(metadata, dict):
             continue
         execution_kind = str(metadata.get("execution_kind") or "").lower()
-        if execution_kind in {"no_action", "diagnostic", "parse_failed", "unsupported_action"}:
+        if execution_kind in {
+            "no_action",
+            "diagnostic",
+            "parse_failed",
+            "unsupported_action",
+            "host_ansible_check_mode_queued",
+        }:
             return True
         if metadata.get("repair_executed") is False and not execution_kind:
             return True
     return False
 
 
+def _approval_has_pending_controlled_handoff(
+    approvals: list[dict[str, Any]],
+) -> bool:
+    for row in approvals:
+        metadata = row.get("extra_metadata")
+        if not isinstance(metadata, dict):
+            continue
+        if (
+            str(metadata.get("execution_kind") or "").lower()
+            == "host_ansible_check_mode_queued"
+            and str(metadata.get("execution_state") or "").lower() == "queued"
+            and metadata.get("repair_executed") is False
+        ):
+            return True
+    return False
+
+
 def _is_no_action_operation(row: dict[str, Any]) -> bool:
     """Return true for durable audit rows that represent observation, not repair."""
     if str(row.get("operation_type") or "") != "playbook_executed":
@@ -1046,6 +1069,9 @@ def build_incident_reconciliation(
     succeeded = sum(int(row.get("sensors_succeeded") or 0) for row in evidence_rows)
     repair_rows = auto_repair_executions or []
     approval_suppressed = _approval_suppresses_repair_execution(approvals)
+    controlled_handoff_pending = _approval_has_pending_controlled_handoff(
+        approvals
+    )
     effective_ops = [] if approval_suppressed else _effective_execution_ops(automation_ops)
     executed_ops = [
         row
@@ -1069,6 +1095,7 @@ def build_incident_reconciliation(
         latest_approval
         and not incident_closed
         and (approval_resolved or approval_status in {"APPROVED", "REJECTED"})
+        and not controlled_handoff_pending
     ):
         if effective_execution_count:
             add(
@@ -1083,7 +1110,11 @@ def build_incident_reconciliation(
                 "Approval reached a terminal state while the incident is still open.",
             )
 
-    if approval_status == "APPROVED" and not effective_execution_count:
+    if (
+        approval_status == "APPROVED"
+        and not effective_execution_count
+        and not controlled_handoff_pending
+    ):
         add(
             "approval_approved_without_execution_record",
             "high",
@@ -1155,6 +1186,7 @@ def build_incident_reconciliation(
             "auto_repair_execution_records": len(repair_rows),
             "successful_auto_repair_records": len(successful_repairs),
             "effective_execution_records": effective_execution_count,
+            "controlled_handoff_pending": controlled_handoff_pending,
             "latest_verification_result": latest_verification,
             "timeline_events": len(timeline_events),
         },
diff --git a/apps/api/src/services/callback_action_spec.yaml b/apps/api/src/services/callback_action_spec.yaml
index 836a2f65a..8088d399c 100644
--- a/apps/api/src/services/callback_action_spec.yaml
+++ b/apps/api/src/services/callback_action_spec.yaml
@@ -307,20 +307,20 @@ actions:
     description: "kubectl rollout undo（回到上一版）"
 
   host_restart_service:
-    label: "重啟服務"
+    label: "受控修復服務"
     emoji: "🔄"
     risk: high
     callback_format: nonce
     category: host_resource
     mcp:
-      provider: ssh
-      tool: ssh_systemctl_restart
+      provider: controlled
+      tool: host_ansible_reconcile
       params:
         host: "{labels.instance}"
         service: "{labels.service}"
     reply_format: text
     timeout_sec: 30
-    description: "systemctl restart 服務（主機層）"
+    description: "Typed Host route → Ansible check-mode → controlled apply → independent verifier"
 
   host_clear_log:
     label: "清 Log"
@@ -371,19 +371,20 @@ actions:
     description: "重啟 MinIO Docker 容器"
 
   reload_nginx:
-    label: "重載 Nginx"
+    label: "受控修復 Nginx"
     emoji: "🔄"
     risk: low
     callback_format: nonce
     category: network
     mcp:
-      provider: ssh
-      tool: ssh_reload_nginx
+      provider: controlled
+      tool: host_ansible_reconcile
       params:
         host: "{labels.instance}"
+        service: "nginx"
     reply_format: text
     timeout_sec: 10
-    description: "nginx -s reload（不停機）"
+    description: "Typed Host route → Ansible check-mode → controlled apply → independent verifier"
 
   renew_cert:
     label: "更新憑證"
diff --git a/apps/api/src/services/callback_dispatcher.py b/apps/api/src/services/callback_dispatcher.py
index b0f33b29c..366ac1186 100644
--- a/apps/api/src/services/callback_dispatcher.py
+++ b/apps/api/src/services/callback_dispatcher.py
@@ -278,6 +278,67 @@ async def dispatch_action(
                 user_id=user_id, result_text=result_text, duration_ms=duration,
             )
 
+        # Host/systemd writes never dispatch the legacy SSH MCP tool. The
+        # callback only records one incident-bound candidate for the existing
+        # single-writer Ansible check/apply/verifier worker.
+        if (
+            spec.mcp_provider == "controlled"
+            and spec.mcp_tool == "host_ansible_reconcile"
+        ):
+            from src.services.host_ansible_controlled_executor import (
+                queue_host_ansible_controlled_action_for_incident_id,
+            )
+
+            requested_host = str(resolved_params.get("host") or "")
+            requested_service = str(resolved_params.get("service") or "")
+            handoff = await queue_host_ansible_controlled_action_for_incident_id(
+                incident_id=incident_id,
+                requested_action=(
+                    f"reconcile host service {requested_service}"
+                    if requested_service and "{" not in requested_service
+                    else "reconcile typed host service"
+                ),
+                risk_level=spec.risk,
+                source="telegram_callback_host_systemd",
+                requested_host=(
+                    requested_host
+                    if requested_host and "{" not in requested_host
+                    else None
+                ),
+            )
+            duration = (time.perf_counter() - start) * 1000
+            if handoff.get("queued") is True:
+                run_id = str(handoff.get("automation_run_id") or "--")
+                return DispatchResult(
+                    success=True,
+                    action=action_name,
+                    incident_id=incident_id,
+                    user_id=user_id,
+                    result_text=(
+                        f"{spec.emoji} <b>{spec.label}</b> 已排入 Ansible check-mode\n"
+                        "├ 目前：尚未執行修復、尚未完成驗證\n"
+                        "├ Executor：<code>host_ansible_executor</code>\n"
+                        "├ Verifier：<code>independent pending</code>\n"
+                        f"└ Run：<code>{run_id}</code>"
+                    ),
+                    duration_ms=duration,
+                )
+            reason = str(
+                handoff.get("status") or "controlled_executor_queue_failed"
+            )
+            return DispatchResult(
+                success=False,
+                action=action_name,
+                incident_id=incident_id,
+                user_id=user_id,
+                result_text=(
+                    f"{spec.emoji} <b>{spec.label}</b> 已由 typed gate 阻擋\n"
+                    f"└ 原因：<code>{reason}</code>"
+                ),
+                error=f"host_ansible_controlled_executor:{reason}",
+                duration_ms=duration,
+            )
+
         # MCP registry dispatch
         from src.plugins.mcp.registry import get_provider
         from src.services.mcp_audit_context import with_mcp_audit_context
diff --git a/apps/api/src/services/host_ansible_controlled_executor.py b/apps/api/src/services/host_ansible_controlled_executor.py
new file mode 100644
index 000000000..fd5cc020a
--- /dev/null
+++ b/apps/api/src/services/host_ansible_controlled_executor.py
@@ -0,0 +1,355 @@
+"""Typed Host/systemd handoff to the single-writer Ansible executor.
+
+This module never opens SSH, runs ``systemctl``, or performs a runtime write.
+It validates the incident-bound typed route, then records one allowlisted
+candidate for the existing check-mode -> controlled apply -> verifier worker.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Awaitable, Callable, Mapping
+from typing import Any
+
+import structlog
+
+from src.services.controlled_alert_target_router import (
+    resolve_typed_incident_target,
+)
+
+logger = structlog.get_logger(__name__)
+
+HostCandidateEnqueuer = Callable[..., Awaitable[dict[str, Any]]]
+
+_RISK_ORDER = {"low": 0, "medium": 1, "high": 2, "critical": 3}
+_HOST_ALIASES = {
+    "99": "192.168.0.99",
+    "host_99": "192.168.0.99",
+    "110": "192.168.0.110",
+    "host_110": "192.168.0.110",
+    "111": "192.168.0.111",
+    "host_111": "192.168.0.111",
+    "112": "192.168.0.112",
+    "host_112": "192.168.0.112",
+    "120": "192.168.0.120",
+    "host_120": "192.168.0.120",
+    "121": "192.168.0.121",
+    "host_121": "192.168.0.121",
+    "188": "192.168.0.188",
+    "host_188": "192.168.0.188",
+}
+_HOST_ROUTE_KINDS = {"host_systemd", "host_launchagent"}
+_ACCEPTED_QUEUE_STATUSES = {
+    "controlled_check_mode_queued",
+    "existing_current_generation",
+}
+
+
+def _incident_payload(incident: Any) -> dict[str, Any]:
+    if isinstance(incident, Mapping):
+        return dict(incident)
+    signals: list[dict[str, Any]] = []
+    for signal in getattr(incident, "signals", None) or []:
+        signals.append(
+            {
+                "alert_name": getattr(signal, "alert_name", None),
+                "labels": dict(getattr(signal, "labels", None) or {}),
+                "annotations": dict(getattr(signal, "annotations", None) or {}),
+            }
+        )
+    severity = getattr(incident, "severity", None)
+    return {
+        "incident_id": getattr(incident, "incident_id", None),
+        "project_id": getattr(incident, "project_id", None),
+        "alertname": getattr(incident, "alertname", None),
+        "alert_category": getattr(incident, "alert_category", None),
+        "severity": getattr(severity, "value", severity),
+        "affected_services": list(
+            getattr(incident, "affected_services", None) or []
+        ),
+        "signals": signals,
+    }
+
+
+def _normalized_host(value: str | None) -> str:
+    token = str(value or "").strip().lower()
+    if token.startswith("http://") or token.startswith("https://"):
+        token = token.split("//", 1)[1]
+    token = token.split("/", 1)[0]
+    if token.count(":") == 1:
+        token = token.split(":", 1)[0]
+    return _HOST_ALIASES.get(token, token)
+
+
+def _effective_risk(requested: Any, route_risk: Any) -> str:
+    requested_value = str(getattr(requested, "value", requested) or "").lower()
+    route_value = str(getattr(route_risk, "value", route_risk) or "").lower()
+    values = [value for value in (requested_value, route_value) if value in _RISK_ORDER]
+    if not values:
+        return ""
+    return max(values, key=_RISK_ORDER.__getitem__)
+
+
+def _blocked_receipt(
+    *,
+    incident_id: str,
+    status: str,
+    route: Mapping[str, Any] | None = None,
+) -> dict[str, Any]:
+    return {
+        "schema_version": "host_ansible_controlled_handoff_v1",
+        "incident_id": incident_id,
+        "status": status,
+        "queued": False,
+        "runtime_write_performed": False,
+        "repair_executed": False,
+        "repair_verified": False,
+        "incident_closure_allowed": False,
+        "single_writer_executor": "awoooi-ansible-executor-broker",
+        "typed_route": dict(route or {}),
+        "active_blockers": [status],
+    }
+
+
+async def load_incident_for_host_ansible_handoff(
+    incident_id: str,
+    *,
+    project_id: str = "awoooi",
+) -> Any | None:
+    """Load current/durable incident truth without synthesizing a target."""
+
+    if not str(incident_id or "").strip():
+        return None
+    from src.services.incident_service import get_incident_service
+
+    service = get_incident_service()
+    incident = await service.get_from_working_memory(incident_id)
+    if incident is None:
+        incident = await service.get_from_episodic_memory(
+            incident_id,
+            project_id=project_id,
+        )
+    return incident
+
+
+async def queue_host_ansible_controlled_action(
+    *,
+    incident: Any,
+    requested_action: str,
+    risk_level: Any,
+    source: str,
+    requested_host: str | None = None,
+    approval_id: str | None = None,
+    automation_run_id: str | None = None,
+    enqueuer: HostCandidateEnqueuer | None = None,
+) -> dict[str, Any]:
+    """Queue one exact typed Host/systemd candidate; never execute directly."""
+
+    payload = _incident_payload(incident)
+    incident_id = str(payload.get("incident_id") or "")
+    if not incident_id:
+        return _blocked_receipt(
+            incident_id="",
+            status="incident_identity_missing",
+        )
+
+    route = resolve_typed_incident_target(payload)
+    route_summary = {
+        key: route.get(key)
+        for key in (
+            "schema_version",
+            "route_id",
+            "resolution_status",
+            "target_kind",
+            "target_resource",
+            "canonical_asset_id",
+            "host",
+            "executor",
+            "verifier",
+            "risk_class",
+            "allowed_catalog_ids",
+            "allowed_inventory_hosts",
+            "controlled_apply_allowed",
+            "cross_domain_fallback_allowed",
+            "drift_work_item_id",
+        )
+    }
+    if route.get("resolution_status") != "resolved":
+        return _blocked_receipt(
+            incident_id=incident_id,
+            status="asset_identity_unresolved",
+            route=route_summary,
+        )
+    if (
+        route.get("target_kind") not in _HOST_ROUTE_KINDS
+        or route.get("executor") != "host_ansible_executor"
+        or route.get("cross_domain_fallback_allowed") is not False
+    ):
+        return _blocked_receipt(
+            incident_id=incident_id,
+            status="typed_host_ansible_route_required",
+            route=route_summary,
+        )
+    if not route.get("verifier"):
+        return _blocked_receipt(
+            incident_id=incident_id,
+            status="independent_verifier_missing",
+            route=route_summary,
+        )
+    if not route.get("allowed_catalog_ids") or not route.get(
+        "allowed_inventory_hosts"
+    ):
+        return _blocked_receipt(
+            incident_id=incident_id,
+            status="allowlisted_ansible_catalog_missing",
+            route=route_summary,
+        )
+    expected_host = _normalized_host(str(route.get("host") or ""))
+    if requested_host and _normalized_host(requested_host) != expected_host:
+        return _blocked_receipt(
+            incident_id=incident_id,
+            status="requested_host_typed_route_mismatch",
+            route=route_summary,
+        )
+
+    effective_risk = _effective_risk(risk_level, route.get("risk_class"))
+    if effective_risk not in {"low", "medium", "high"}:
+        return _blocked_receipt(
+            incident_id=incident_id,
+            status="critical_break_glass_required"
+            if effective_risk == "critical"
+            else "risk_policy_not_resolved",
+            route=route_summary,
+        )
+
+    if enqueuer is None:
+        from src.jobs.awooop_ansible_candidate_backfill_job import (
+            enqueue_ai_decision_ansible_candidate,
+        )
+
+        enqueuer = enqueue_ai_decision_ansible_candidate
+
+    proposal_data = {
+        "source": source,
+        "risk_level": effective_risk,
+        "action": "enqueue_allowlisted_host_ansible_reconciliation",
+        "requested_action": str(requested_action or "")[:240],
+        "approval_id": str(approval_id or ""),
+        "typed_route_id": str(route.get("route_id") or ""),
+        "canonical_asset_id": str(route.get("canonical_asset_id") or ""),
+        "cross_domain_fallback_allowed": False,
+        "check_mode_required_before_apply": True,
+        "independent_verifier": str(route.get("verifier") or ""),
+    }
+    try:
+        handoff = await enqueuer(
+            incident=incident,
+            proposal_data=proposal_data,
+            project_id=str(payload.get("project_id") or "awoooi"),
+            automation_run_id=automation_run_id,
+        )
+    except Exception as exc:
+        logger.warning(
+            "host_ansible_controlled_handoff_failed",
+            incident_id=incident_id,
+            error_type=type(exc).__name__,
+            runtime_write_performed=False,
+        )
+        return _blocked_receipt(
+            incident_id=incident_id,
+            status="controlled_executor_queue_failed",
+            route=route_summary,
+        )
+
+    status = str(handoff.get("status") or "controlled_executor_queue_failed")
+    active_blockers = list(handoff.get("active_blockers") or [])
+    valid_queue_receipt = (
+        handoff.get("schema_version")
+        == "ai_decision_controlled_executor_handoff_v1"
+        and handoff.get("queued") is True
+        and status in _ACCEPTED_QUEUE_STATUSES
+        and handoff.get("side_effect_performed") is False
+        and handoff.get("single_writer_executor")
+        == "awoooi-ansible-executor-broker"
+        and bool(str(handoff.get("automation_run_id") or "").strip())
+        and bool(str(handoff.get("trace_id") or "").strip())
+        and bool(str(handoff.get("work_item_id") or "").strip())
+        and not active_blockers
+    )
+    if not valid_queue_receipt:
+        logger.warning(
+            "host_ansible_controlled_handoff_receipt_invalid",
+            incident_id=incident_id,
+            handoff_status=status,
+            queued=handoff.get("queued") is True,
+            runtime_write_performed=False,
+        )
+        return _blocked_receipt(
+            incident_id=incident_id,
+            status="controlled_executor_receipt_invalid",
+            route=route_summary,
+        )
+
+    return {
+        "schema_version": "host_ansible_controlled_handoff_v1",
+        "incident_id": incident_id,
+        "status": status,
+        "queued": True,
+        "runtime_write_performed": False,
+        "repair_executed": False,
+        "repair_verified": False,
+        "incident_closure_allowed": False,
+        "single_writer_executor": "awoooi-ansible-executor-broker",
+        "automation_run_id": handoff.get("automation_run_id"),
+        "trace_id": handoff.get("trace_id"),
+        "work_item_id": handoff.get("work_item_id"),
+        "typed_route": route_summary,
+        "active_blockers": active_blockers,
+    }
+
+
+async def queue_host_ansible_controlled_action_for_incident_id(
+    *,
+    incident_id: str,
+    requested_action: str,
+    risk_level: Any,
+    source: str,
+    requested_host: str | None = None,
+    approval_id: str | None = None,
+    automation_run_id: str | None = None,
+) -> dict[str, Any]:
+    """Load the incident, then perform the same fail-closed typed handoff."""
+
+    try:
+        incident = await load_incident_for_host_ansible_handoff(incident_id)
+    except Exception as exc:
+        logger.warning(
+            "host_ansible_incident_lookup_failed",
+            incident_id=incident_id,
+            error_type=type(exc).__name__,
+            runtime_write_performed=False,
+        )
+        incident = None
+    if incident is None:
+        return _blocked_receipt(
+            incident_id=incident_id,
+            status="incident_not_found",
+        )
+    loaded_id = str(
+        getattr(incident, "incident_id", None)
+        or (incident.get("incident_id") if isinstance(incident, Mapping) else "")
+        or ""
+    )
+    if loaded_id != incident_id:
+        return _blocked_receipt(
+            incident_id=incident_id,
+            status="incident_identity_mismatch",
+        )
+    return await queue_host_ansible_controlled_action(
+        incident=incident,
+        requested_action=requested_action,
+        risk_level=risk_level,
+        source=source,
+        requested_host=requested_host,
+        approval_id=approval_id,
+        automation_run_id=automation_run_id,
+    )
diff --git a/apps/api/tests/test_ansible_verified_closure.py b/apps/api/tests/test_ansible_verified_closure.py
index f3339624f..2ccfdb61d 100644
--- a/apps/api/tests/test_ansible_verified_closure.py
+++ b/apps/api/tests/test_ansible_verified_closure.py
@@ -4,6 +4,7 @@ import inspect
 import json
 from contextlib import asynccontextmanager
 from datetime import UTC, datetime
+from types import SimpleNamespace
 from unittest.mock import AsyncMock
 
 import pytest
@@ -92,6 +93,46 @@ def _verified_result() -> service.AnsibleRunResult:
     )
 
 
+@pytest.mark.asyncio
+async def test_terminal_apply_replaces_queued_approval_truth(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    approval = SimpleNamespace(
+        status=service.ApprovalStatus.APPROVED,
+        resolved_at=None,
+        rejection_reason=None,
+        extra_metadata={
+            "execution_kind": "host_ansible_check_mode_queued",
+            "execution_state": "queued",
+            "repair_executed": False,
+            "repair_verified": False,
+            "incident_closure_allowed": False,
+        },
+    )
+    db = _SequenceDB(_MappingResult(scalar=approval))
+
+    @asynccontextmanager
+    async def fake_get_db_context(_project_id):
+        yield db
+
+    monkeypatch.setattr(service, "get_db_context", fake_get_db_context)
+
+    written = await service._finalize_controlled_approval_projection(
+        _claim(),
+        _verified_result(),
+        apply_op_id="00000000-0000-0000-0000-000000000104",
+        project_id="awoooi",
+    )
+
+    assert written is True
+    assert approval.status == service.ApprovalStatus.EXECUTION_SUCCESS
+    assert approval.extra_metadata["execution_kind"] == "controlled_apply"
+    assert approval.extra_metadata["execution_state"] == "completed"
+    assert approval.extra_metadata["repair_executed"] is True
+    assert approval.extra_metadata["repair_verified"] is True
+    assert approval.extra_metadata["incident_closure_allowed"] is True
+
+
 def _controlled_result_payload() -> dict:
     return {
         "chat_id": "sre-chat",
diff --git a/apps/api/tests/test_approval_controlled_handoff_pending.py b/apps/api/tests/test_approval_controlled_handoff_pending.py
new file mode 100644
index 000000000..aeb98e195
--- /dev/null
+++ b/apps/api/tests/test_approval_controlled_handoff_pending.py
@@ -0,0 +1,63 @@
+from contextlib import asynccontextmanager
+from types import SimpleNamespace
+from uuid import uuid4
+
+import pytest
+
+from src.models.approval import ApprovalStatus
+from src.services import approval_db as approval_db_module
+from src.services.approval_db import ApprovalDBService
+
+
+@pytest.mark.asyncio
+async def test_pending_handoff_preserves_approved_status_and_records_truth(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    record = SimpleNamespace(
+        status=ApprovalStatus.APPROVED,
+        extra_metadata={"existing": "kept"},
+    )
+
+    class FakeResult:
+        def scalar_one_or_none(self):
+            return record
+
+    class FakeDB:
+        async def execute(self, _query):
+            return FakeResult()
+
+        async def flush(self):
+            return None
+
+    @asynccontextmanager
+    async def fake_db_context():
+        yield FakeDB()
+
+    monkeypatch.setattr(
+        approval_db_module,
+        "get_db_context",
+        fake_db_context,
+    )
+
+    recorded = await ApprovalDBService().record_controlled_handoff_pending(
+        uuid4(),
+        execution_kind="host_ansible_check_mode_queued",
+        receipt={
+            "automation_run_id": "run-host-188",
+            "single_writer_executor": "awoooi-ansible-executor-broker",
+        },
+    )
+
+    assert recorded is True
+    assert record.status == ApprovalStatus.APPROVED
+    assert record.extra_metadata == {
+        "existing": "kept",
+        "execution_kind": "host_ansible_check_mode_queued",
+        "execution_state": "queued",
+        "repair_attempted": False,
+        "repair_executed": False,
+        "repair_verified": False,
+        "incident_closure_allowed": False,
+        "automation_run_id": "run-host-188",
+        "single_writer_executor": "awoooi-ansible-executor-broker",
+    }
diff --git a/apps/api/tests/test_approval_execution_mcp_audit.py b/apps/api/tests/test_approval_execution_mcp_audit.py
index 207e0c5e4..62086d07d 100644
--- a/apps/api/tests/test_approval_execution_mcp_audit.py
+++ b/apps/api/tests/test_approval_execution_mcp_audit.py
@@ -1,7 +1,9 @@
 from __future__ import annotations
 
 from contextlib import asynccontextmanager
+from types import SimpleNamespace
 from typing import Any
+from unittest.mock import AsyncMock
 from uuid import UUID
 
 import pytest
@@ -10,6 +12,7 @@ from src.models.approval import ApprovalRequest, RiskLevel
 from src.plugins.mcp.gateway import GateApprovalError
 from src.plugins.mcp.interfaces import MCPToolResult
 from src.services.approval_execution import ApprovalExecutionService
+from src.services.executor import ExecutionResult, OperationType
 
 
 class FakeRedis:
@@ -20,6 +23,251 @@ class FakeRedis:
         self.set_calls.append((key, value, ex))
 
 
+@pytest.mark.asyncio
+async def test_systemctl_approval_queues_ansible_without_ssh_gateway(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    queue = AsyncMock(
+        return_value={
+            "status": "controlled_check_mode_queued",
+            "queued": True,
+            "automation_run_id": "run-systemd-approval",
+            "single_writer_executor": "awoooi-ansible-executor-broker",
+            "runtime_write_performed": False,
+            "repair_executed": False,
+            "repair_verified": False,
+        }
+    )
+    gateway = AsyncMock(
+        side_effect=AssertionError("systemctl must not use the SSH gateway")
+    )
+    monkeypatch.setattr(
+        "src.services.host_ansible_controlled_executor."
+        "queue_host_ansible_controlled_action_for_incident_id",
+        queue,
+    )
+    monkeypatch.setattr(
+        ApprovalExecutionService,
+        "_execute_ssh_tool_via_gateway",
+        gateway,
+    )
+    approval = ApprovalRequest(
+        action="ssh 192.168.0.110 'systemctl restart node_exporter'",
+        description="typed Host/systemd approval",
+        risk_level=RiskLevel.MEDIUM,
+        requested_by="test",
+        required_signatures=0,
+        incident_id="INC-NODE-EXPORTER-110",
+    )
+
+    result = await ApprovalExecutionService()._execute_ssh_host_action(
+        approval=approval,
+        host="192.168.0.110",
+    )
+
+    assert result.success is True
+    assert result.k8s_response is not None
+    assert result.k8s_response["controlled_execution_queued"] is True
+    assert result.k8s_response["runtime_write_performed"] is False
+    assert result.k8s_response["repair_executed"] is False
+    queue.assert_awaited_once()
+    gateway.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "action",
+    [
+        "ssh://root@192.168.0.110/systemctl%20start%20node_exporter",
+        "ssh 192.168.0.110 'systemctl restart node_exporter || true'",
+        (
+            "ssh 192.168.0.110 'docker restart exporter "
+            "|| systemctl restart node_exporter'"
+        ),
+    ],
+)
+async def test_unbounded_systemd_approval_never_reaches_any_executor(
+    monkeypatch: pytest.MonkeyPatch,
+    action: str,
+) -> None:
+    queue = AsyncMock(
+        side_effect=AssertionError("invalid systemd action must not be queued")
+    )
+    gateway = AsyncMock(
+        side_effect=AssertionError("invalid systemd action must not use SSH MCP")
+    )
+    monkeypatch.setattr(
+        "src.services.host_ansible_controlled_executor."
+        "queue_host_ansible_controlled_action_for_incident_id",
+        queue,
+    )
+    monkeypatch.setattr(
+        ApprovalExecutionService,
+        "_execute_ssh_tool_via_gateway",
+        gateway,
+    )
+    approval = ApprovalRequest(
+        action=action,
+        description="unbounded Host/systemd approval",
+        risk_level=RiskLevel.MEDIUM,
+        requested_by="test",
+        required_signatures=0,
+        incident_id="INC-NODE-EXPORTER-110",
+    )
+
+    result = await ApprovalExecutionService()._execute_ssh_host_action(
+        approval=approval,
+        host="192.168.0.110",
+    )
+
+    assert result.success is False
+    assert "unbounded_or_unallowlisted" in str(result.error)
+    queue.assert_not_awaited()
+    gateway.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_queued_host_approval_does_not_close_or_learn_as_repair(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    approval = ApprovalRequest(
+        action="ssh 192.168.0.110 'systemctl restart node_exporter'",
+        description="typed Host/systemd approval",
+        risk_level=RiskLevel.MEDIUM,
+        requested_by="operator",
+        required_signatures=0,
+        incident_id="INC-NODE-EXPORTER-110",
+    )
+    handoff = {
+        "status": "controlled_check_mode_queued",
+        "queued": True,
+        "automation_run_id": "run-systemd-approval",
+        "single_writer_executor": "awoooi-ansible-executor-broker",
+        "runtime_write_performed": False,
+        "repair_executed": False,
+        "repair_verified": False,
+    }
+    execute_ssh = AsyncMock(
+        return_value=ExecutionResult(
+            success=True,
+            message="host_ansible_controlled_executor queued",
+            operation_type=OperationType.SSH_HOST,
+            target_resource="192.168.0.110",
+            namespace="host",
+            duration_ms=1,
+            k8s_response={
+                "controlled_execution_queued": True,
+                "runtime_write_performed": False,
+                "controlled_handoff": handoff,
+            },
+        )
+    )
+    update_status = AsyncMock()
+    record_pending = AsyncMock(return_value=True)
+    timeline = SimpleNamespace(add_event=AsyncMock())
+    log_completed = AsyncMock()
+    alert_completed = AsyncMock()
+    push_result = AsyncMock()
+
+    monkeypatch.setattr(
+        "src.services.approval_execution.get_approval_service",
+        lambda: SimpleNamespace(
+            update_execution_status=update_status,
+            record_controlled_handoff_pending=record_pending,
+        ),
+    )
+    monkeypatch.setattr(
+        "src.services.approval_execution.get_timeline_service",
+        lambda: timeline,
+    )
+    monkeypatch.setattr(
+        "src.services.approval_execution.get_executor",
+        lambda: object(),
+    )
+    monkeypatch.setattr(
+        "src.services.resource_resolver.get_resource_resolver",
+        lambda: SimpleNamespace(
+            resolve=AsyncMock(
+                return_value=SimpleNamespace(
+                    success=False,
+                    resource_name=None,
+                    candidates=[],
+                )
+            )
+        ),
+    )
+    monkeypatch.setattr(
+        ApprovalExecutionService,
+        "_execute_ssh_host_action",
+        execute_ssh,
+    )
+    monkeypatch.setattr(
+        ApprovalExecutionService,
+        "_log_aol_started",
+        AsyncMock(return_value="aol-op-id"),
+    )
+    monkeypatch.setattr(
+        ApprovalExecutionService,
+        "_log_alert_execution_started",
+        AsyncMock(),
+    )
+    monkeypatch.setattr(
+        ApprovalExecutionService,
+        "_log_aol_completed",
+        log_completed,
+    )
+    monkeypatch.setattr(
+        ApprovalExecutionService,
+        "_log_alert_execution_completed",
+        alert_completed,
+    )
+    monkeypatch.setattr(
+        ApprovalExecutionService,
+        "_push_execution_result_to_alert",
+        push_result,
+    )
+    learning = AsyncMock(
+        side_effect=AssertionError("queued handoff must not trigger learning")
+    )
+    resolve = AsyncMock(
+        side_effect=AssertionError("queued handoff must not resolve incident")
+    )
+    monkeypatch.setattr(
+        ApprovalExecutionService,
+        "_trigger_learning",
+        learning,
+    )
+    monkeypatch.setattr(
+        "src.services.incident_service.get_incident_service",
+        lambda: SimpleNamespace(resolve_incident=resolve),
+    )
+
+    result = await ApprovalExecutionService().execute_approved_action(approval)
+
+    assert result is False
+    update_status.assert_not_awaited()
+    record_pending.assert_awaited_once_with(
+        approval.id,
+        execution_kind="host_ansible_check_mode_queued",
+        receipt=handoff,
+    )
+    assert timeline.add_event.await_args.kwargs["status"] == "pending"
+    assert "尚未修復" in timeline.add_event.await_args.kwargs["title"]
+    alert_completed.assert_not_awaited()
+    assert log_completed.await_args.kwargs["status"] == "dry_run"
+    assert log_completed.await_args.kwargs["output"]["incident_closure_allowed"] is False
+    push_result.assert_awaited_once_with(
+        approval,
+        success=False,
+        error=None,
+        execution_kind="host_ansible_check_mode_queued",
+        repair_executed=False,
+        repair_attempted=False,
+    )
+    learning.assert_not_awaited()
+    resolve.assert_not_awaited()
+
+
 @pytest.mark.asyncio
 async def test_ssh_approval_execution_projects_approval_into_gateway(
     monkeypatch: pytest.MonkeyPatch,
diff --git a/apps/api/tests/test_auto_repair_api.py b/apps/api/tests/test_auto_repair_api.py
new file mode 100644
index 000000000..600e45372
--- /dev/null
+++ b/apps/api/tests/test_auto_repair_api.py
@@ -0,0 +1,66 @@
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+
+from src.api.v1 import auto_repair as auto_repair_api
+
+
+@pytest.mark.asyncio
+async def test_execute_api_preserves_controlled_queue_truth(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    incident = object()
+    playbook = object()
+    incident_service = SimpleNamespace(
+        get_from_working_memory=AsyncMock(return_value=incident)
+    )
+    playbook_service = SimpleNamespace(
+        get_by_id=AsyncMock(return_value=playbook)
+    )
+    repair_service = SimpleNamespace(
+        execute_auto_repair=AsyncMock(
+            return_value=SimpleNamespace(
+                success=False,
+                status="queued",
+                repair_executed=False,
+                repair_verified=False,
+                incident_id="INC-HOST-QUEUED",
+                playbook_id="PB-HOST-ANSIBLE",
+                executed_steps=[
+                    "Step 1: systemctl restart node_exporter -> QUEUED"
+                ],
+                error=None,
+                execution_time_ms=7,
+            )
+        )
+    )
+    monkeypatch.setattr(
+        auto_repair_api,
+        "get_incident_service",
+        lambda: incident_service,
+    )
+    monkeypatch.setattr(
+        "src.services.playbook_service.get_playbook_service",
+        lambda: playbook_service,
+    )
+    monkeypatch.setattr(
+        auto_repair_api,
+        "get_auto_repair_service",
+        lambda: repair_service,
+    )
+
+    response = await auto_repair_api.execute_auto_repair(
+        auto_repair_api.ExecuteRequest(
+            incident_id="INC-HOST-QUEUED",
+            playbook_id="PB-HOST-ANSIBLE",
+            force=True,
+        ),
+        None,  # type: ignore[arg-type]
+    )
+
+    assert response.success is False
+    assert response.status == "queued"
+    assert response.repair_executed is False
+    assert response.repair_verified is False
+    assert response.error is None
diff --git a/apps/api/tests/test_auto_repair_service.py b/apps/api/tests/test_auto_repair_service.py
index 59c9cf2ef..5437d1a48 100644
--- a/apps/api/tests/test_auto_repair_service.py
+++ b/apps/api/tests/test_auto_repair_service.py
@@ -8,6 +8,8 @@ Auto Repair Service Tests - #8 自動升級決策
 建立者: Claude Code (#8 自動升級決策)
 """
 
+from unittest.mock import AsyncMock
+
 import pytest
 
 from src.models.incident import Incident, IncidentStatus, Severity, Signal
@@ -179,6 +181,362 @@ class TestAutoRepairService:
         assert decision.risk_level == RiskLevel.HIGH
         assert decision.blocked_by is None
 
+    @pytest.mark.asyncio
+    async def test_systemctl_step_queues_host_ansible_without_legacy_agent(
+        self,
+        service,
+        monkeypatch,
+    ):
+        incident = create_test_incident(
+            severity=Severity.P2,
+            alert_category="infrastructure",
+            alert_name="NodeExporterDown",
+        )
+        incident.affected_services = ["node-exporter-110"]
+        incident.signals[0].labels.update(
+            {
+                "alertname": "NodeExporterDown",
+                "host": "110",
+                "instance": "192.168.0.110",
+                "service": "node-exporter-110",
+            }
+        )
+        queue = AsyncMock(
+            return_value={
+                "status": "controlled_check_mode_queued",
+                "queued": True,
+                "automation_run_id": "run-node-exporter-110",
+                "runtime_write_performed": False,
+                "repair_executed": False,
+                "repair_verified": False,
+            }
+        )
+        monkeypatch.setattr(
+            "src.services.host_ansible_controlled_executor."
+            "queue_host_ansible_controlled_action",
+            queue,
+        )
+        step = RepairStep(
+            step_number=1,
+            action_type=ActionType.SSH_COMMAND,
+            command=(
+                "ssh {host} 'sudo systemctl restart node_exporter.service'"
+            ),
+            risk_level=RiskLevel.MEDIUM,
+            requires_approval=False,
+        )
+
+        result = await service._execute_step(incident, step)
+
+        assert result == "QUEUED: host_ansible_executor:run-node-exporter-110"
+        queue.assert_awaited_once()
+        assert queue.await_args.kwargs["requested_host"] == "110"
+
+    @pytest.mark.asyncio
+    async def test_systemctl_uri_queues_with_explicit_host(
+        self,
+        service,
+        monkeypatch,
+    ):
+        incident = create_test_incident(
+            alert_category="infrastructure",
+            alert_name="NodeExporterDown",
+        )
+        queue = AsyncMock(
+            return_value={
+                "status": "controlled_check_mode_queued",
+                "queued": True,
+                "automation_run_id": "run-node-exporter-188",
+            }
+        )
+        legacy = AsyncMock(
+            side_effect=AssertionError("systemd URI must not use direct SSH")
+        )
+        monkeypatch.setattr(
+            "src.services.host_ansible_controlled_executor."
+            "queue_host_ansible_controlled_action",
+            queue,
+        )
+        monkeypatch.setattr(
+            "src.services.host_repair_agent.HostRepairAgent.repair_by_uri",
+            legacy,
+        )
+        step = RepairStep(
+            step_number=1,
+            action_type=ActionType.SSH_COMMAND,
+            command=(
+                "ssh://ollama@192.168.0.188/"
+                "systemctl%20restart%20node_exporter.service"
+            ),
+            risk_level=RiskLevel.MEDIUM,
+        )
+
+        result = await service._execute_step(incident, step)
+
+        assert result == "QUEUED: host_ansible_executor:run-node-exporter-188"
+        assert queue.await_args.kwargs["requested_host"] == "192.168.0.188"
+        legacy.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        "command,reason",
+        [
+            (
+                "ssh://root@192.168.0.188/systemctl%20start%20node_exporter",
+                "systemd_action_not_allowlisted:start",
+            ),
+            (
+                "ssh://root@192.168.0.188/systemctl%20stop%20node_exporter",
+                "systemd_action_not_allowlisted:stop",
+            ),
+            (
+                "ssh://root@192.168.0.188/systemctl%20reload%20node_exporter",
+                "systemd_action_not_allowlisted:reload",
+            ),
+            (
+                "ssh://root@192.168.0.188/systemctl%20restart%20--no-block",
+                "unbounded_systemd_mutation_rejected",
+            ),
+            (
+                "ssh 192.168.0.188 'sudo -n systemctl restart node_exporter'",
+                "unbounded_systemd_mutation_rejected",
+            ),
+            (
+                "ssh 192.168.0.188 'systemctl restart node_exporter || true'",
+                "unbounded_systemd_mutation_rejected",
+            ),
+            (
+                "ssh 192.168.0.188 'systemctl restart node_exporter; true'",
+                "unbounded_systemd_mutation_rejected",
+            ),
+            (
+                "ssh 192.168.0.188 'systemctl restart $(hostname)'",
+                "unbounded_systemd_mutation_rejected",
+            ),
+            (
+                "ssh 192.168.0.188 'systemctl restart\nnode_exporter'",
+                "unbounded_systemd_mutation_rejected",
+            ),
+            (
+                "ssh 192.168.0.188 'systemctl try-restart node_exporter'",
+                "unbounded_systemd_mutation_rejected",
+            ),
+            (
+                "ssh 192.168.0.188 'systemctl enable node_exporter'",
+                "unbounded_systemd_mutation_rejected",
+            ),
+            (
+                "ssh://root@192.168.0.188/"
+                "sys%22%22temctl%20restart%20node_exporter",
+                "unbounded_systemd_mutation_rejected",
+            ),
+            (
+                "ssh://root@192.168.0.188/"
+                "syst%5Cemctl%20restart%20node_exporter",
+                "unbounded_systemd_mutation_rejected",
+            ),
+            (
+                "ssh://root@192.168.0.188/"
+                "service%20node_exporter%20restart",
+                "unbounded_systemd_mutation_rejected",
+            ),
+        ],
+    )
+    async def test_unallowlisted_systemd_mutations_never_reach_direct_ssh(
+        self,
+        service,
+        monkeypatch,
+        command,
+        reason,
+    ):
+        incident = create_test_incident(
+            alert_category="infrastructure",
+            alert_name="NodeExporterDown",
+        )
+        queue = AsyncMock()
+        legacy = AsyncMock(
+            side_effect=AssertionError("systemd mutation must fail closed")
+        )
+        monkeypatch.setattr(
+            "src.services.host_ansible_controlled_executor."
+            "queue_host_ansible_controlled_action",
+            queue,
+        )
+        monkeypatch.setattr(
+            "src.services.host_repair_agent.HostRepairAgent.repair_by_uri",
+            legacy,
+        )
+        step = RepairStep(
+            step_number=1,
+            action_type=ActionType.SSH_COMMAND,
+            command=command,
+            risk_level=RiskLevel.MEDIUM,
+        )
+
+        result = await service._execute_step(incident, step)
+
+        assert result == f"FAILED: host_ansible_controlled_executor:{reason}"
+        queue.assert_not_awaited()
+        legacy.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        "command",
+        [
+            "ssh://wooo@192.168.0.110/echo%20ready",
+            "ansible://192.168.0.188/untyped_legacy.yml",
+            "openclaw://docker-110/sentry",
+            "unsupported://192.168.0.110/restart",
+        ],
+    )
+    async def test_untyped_host_executor_fallback_is_disabled(
+        self,
+        service,
+        monkeypatch,
+        command,
+    ):
+        incident = create_test_incident(
+            alert_category="infrastructure",
+            alert_name="HostRepairRequested",
+        )
+        legacy = AsyncMock(
+            side_effect=AssertionError("untyped commands must not use direct host executor")
+        )
+        monkeypatch.setattr(
+            "src.services.host_repair_agent.HostRepairAgent.repair_by_uri",
+            legacy,
+        )
+        step = RepairStep(
+            step_number=1,
+            action_type=ActionType.SSH_COMMAND,
+            command=command,
+            risk_level=RiskLevel.MEDIUM,
+        )
+
+        result = await service._execute_step(incident, step)
+
+        assert result == (
+            "FAILED: legacy_direct_host_executor_disabled_use_typed_executor"
+        )
+        legacy.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_controlled_queue_is_not_recorded_as_repair_success(
+        self,
+        service,
+        mock_playbook_service,
+        monkeypatch,
+    ):
+        incident = create_test_incident(
+            severity=Severity.P2,
+            alert_category="infrastructure",
+            alert_name="NodeExporterDown",
+        )
+        playbook = Playbook(
+            playbook_id="PB-HOST-ANSIBLE-QUEUED",
+            name="Host Ansible controlled repair",
+            description="queue exact Host/systemd repair",
+            status=PlaybookStatus.APPROVED,
+            symptom_pattern=SymptomPattern(
+                alert_names=["NodeExporterDown"],
+                affected_services=["node-exporter-110"],
+            ),
+            repair_steps=[
+                RepairStep(
+                    step_number=1,
+                    action_type=ActionType.SSH_COMMAND,
+                    command="ssh {host} 'systemctl restart node_exporter'",
+                    risk_level=RiskLevel.MEDIUM,
+                )
+            ],
+        )
+        mock_playbook_service.add_playbook(playbook)
+        monkeypatch.setattr(
+            service,
+            "_execute_step",
+            AsyncMock(
+                return_value="QUEUED: host_ansible_executor:run-pending"
+            ),
+        )
+        repair_counter = AsyncMock()
+        monkeypatch.setattr(
+            "src.services.auto_repair_service.record_global_repair_action",
+            repair_counter,
+        )
+
+        result = await service.execute_auto_repair(incident, playbook)
+
+        assert result.success is False
+        assert result.status == "queued"
+        assert result.repair_executed is False
+        assert result.repair_verified is False
+        assert playbook.success_count == 0
+        assert playbook.failure_count == 0
+        repair_counter.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        "host_command",
+        [
+            "systemctl restart node_exporter",
+            "ssh {host} 'sys$@temctl restart node_exporter'",
+            "ssh {host} 'sys$*temctl restart node_exporter'",
+        ],
+    )
+    async def test_host_playbook_cannot_execute_another_domain_first(
+        self,
+        service,
+        mock_playbook_service,
+        monkeypatch,
+        host_command,
+    ):
+        incident = create_test_incident(
+            alert_category="infrastructure",
+            alert_name="NodeExporterDown",
+        )
+        playbook = Playbook(
+            playbook_id="PB-MIXED-DOMAIN-BLOCKED",
+            name="mixed domain playbook",
+            description="must fail before any runtime action",
+            status=PlaybookStatus.APPROVED,
+            symptom_pattern=SymptomPattern(
+                alert_names=["NodeExporterDown"],
+                affected_services=["node-exporter-110"],
+            ),
+            repair_steps=[
+                RepairStep(
+                    step_number=1,
+                    action_type=ActionType.KUBECTL,
+                    command="kubectl rollout restart deployment/api",
+                    risk_level=RiskLevel.MEDIUM,
+                ),
+                RepairStep(
+                    step_number=2,
+                    action_type=ActionType.SSH_COMMAND,
+                    command=host_command,
+                    risk_level=RiskLevel.MEDIUM,
+                ),
+            ],
+        )
+        mock_playbook_service.add_playbook(playbook)
+        execute_step = AsyncMock(
+            side_effect=AssertionError("mixed domain must fail before execution")
+        )
+        repair_counter = AsyncMock()
+        monkeypatch.setattr(service, "_execute_step", execute_step)
+        monkeypatch.setattr(
+            "src.services.auto_repair_service.record_global_repair_action",
+            repair_counter,
+        )
+
+        result = await service.execute_auto_repair(incident, playbook)
+
+        assert result.success is False
+        assert result.status == "failed"
+        assert "cross_domain_or_multi_step" in str(result.error)
+        execute_step.assert_not_awaited()
+        repair_counter.assert_not_awaited()
+
     @pytest.mark.asyncio
     async def test_evaluate_allows_p0_severity_with_approved_playbook(
         self,
diff --git a/apps/api/tests/test_awooop_truth_chain_service.py b/apps/api/tests/test_awooop_truth_chain_service.py
index 85762083e..5289284b5 100644
--- a/apps/api/tests/test_awooop_truth_chain_service.py
+++ b/apps/api/tests/test_awooop_truth_chain_service.py
@@ -859,6 +859,95 @@ def test_reconciliation_ignores_no_action_audit_rows_as_execution() -> None:
     assert "incident_open_after_successful_execution" not in codes
 
 
+def test_reconciliation_keeps_host_ansible_queue_pending_not_executed() -> None:
+    reconciliation = build_incident_reconciliation(
+        incident={"incident_id": "INC-HOST-QUEUED", "status": "INVESTIGATING"},
+        approvals=[
+            {
+                "id": "approval-host-queued",
+                "status": "APPROVED",
+                "action": "ssh 192.168.0.188 'systemctl restart node_exporter'",
+                "resolved_at": "2026-07-18T01:00:00+00:00",
+                "extra_metadata": {
+                    "execution_kind": "host_ansible_check_mode_queued",
+                    "execution_state": "queued",
+                    "repair_attempted": False,
+                    "repair_executed": False,
+                    "repair_verified": False,
+                    "automation_run_id": "run-host-queued",
+                },
+            }
+        ],
+        evidence_rows=[{"sensors_attempted": 3, "sensors_succeeded": 3}],
+        automation_ops=[
+            {
+                "operation_type": "playbook_executed",
+                "status": "dry_run",
+                "actor": "approval_execution",
+                "output_execution_kind": "host_ansible_check_mode_queued",
+                "output_repair_executed": False,
+            },
+            {
+                "operation_type": "ansible_candidate_matched",
+                "status": "dry_run",
+                "actor": "awoooi-ansible-executor-broker",
+            },
+        ],
+        auto_repair_executions=[],
+        timeline_events=[{"event_type": "exec", "status": "pending"}],
+    )
+
+    codes = {row["code"] for row in reconciliation["mismatches"]}
+    assert reconciliation["facts"]["effective_execution_records"] == 0
+    assert reconciliation["facts"]["controlled_handoff_pending"] is True
+    assert "incident_open_after_successful_execution" not in codes
+    assert "incident_open_after_approval_resolved" not in codes
+    assert "approval_approved_without_execution_record" not in codes
+
+
+def test_reconciliation_counts_terminal_host_ansible_apply_after_queue() -> None:
+    reconciliation = build_incident_reconciliation(
+        incident={"incident_id": "INC-HOST-APPLIED", "status": "INVESTIGATING"},
+        approvals=[
+            {
+                "id": "approval-host-applied",
+                "status": "EXECUTION_SUCCESS",
+                "action": "systemctl restart node_exporter",
+                "resolved_at": "2026-07-18T01:05:00+00:00",
+                "extra_metadata": {
+                    "execution_kind": "controlled_apply",
+                    "execution_state": "queued",
+                    "repair_attempted": True,
+                    "repair_executed": True,
+                    "post_verifier_passed": True,
+                    "automation_run_id": "run-host-applied",
+                },
+            }
+        ],
+        evidence_rows=[
+            {
+                "sensors_attempted": 3,
+                "sensors_succeeded": 3,
+                "verification_result": "success",
+            }
+        ],
+        automation_ops=[
+            {
+                "operation_type": "ansible_apply_executed",
+                "status": "success",
+                "actor": "ansible_controlled_apply_worker",
+            }
+        ],
+        auto_repair_executions=[],
+        timeline_events=[{"event_type": "exec", "status": "success"}],
+    )
+
+    codes = {row["code"] for row in reconciliation["mismatches"]}
+    assert reconciliation["facts"]["effective_execution_records"] == 1
+    assert reconciliation["facts"]["controlled_handoff_pending"] is False
+    assert "incident_open_after_successful_execution" in codes
+
+
 def test_automation_quality_ignores_no_action_audit_rows_as_execution() -> None:
     quality = build_automation_quality(
         incident={"incident_id": "INC-1", "status": "RESOLVED"},
diff --git a/apps/api/tests/test_callback_dispatcher.py b/apps/api/tests/test_callback_dispatcher.py
index 9e7bb2d00..c48dd941a 100644
--- a/apps/api/tests/test_callback_dispatcher.py
+++ b/apps/api/tests/test_callback_dispatcher.py
@@ -13,21 +13,22 @@ Phase 5 Sprint 5.0-5.1 Callback Dispatcher 單元測試
 🔴 遵循「禁止 Mock 測試鐵律」: 用真實 spec registry，不 mock。
 """
 
+from unittest.mock import AsyncMock
+
 import pytest
 
 from src.plugins.mcp.interfaces import MCPTool, MCPToolProvider, MCPToolResult
 from src.services import callback_dispatcher as callback_dispatcher_module
 from src.services.callback_dispatcher import (
+    _lookup_context,
+    _resolve_provider_name,
+    _resolve_template,
     dispatch_action,
     get_action_spec,
     list_actions_for_category,
     load_action_registry,
-    _lookup_context,
-    _resolve_provider_name,
-    _resolve_template,
 )
 
-
 # =============================================================================
 # Registry loading
 # =============================================================================
@@ -49,6 +50,25 @@ class TestRegistryLoading:
             assert spec.mcp_provider, f"{name} missing mcp_provider"
             assert spec.mcp_tool, f"{name} missing mcp_tool"
 
+    @pytest.mark.parametrize(
+        "action_name,service",
+        [
+            ("host_restart_service", "{labels.service}"),
+            ("reload_nginx", "nginx"),
+        ],
+    )
+    def test_systemd_callbacks_use_only_controlled_host_ansible_route(
+        self,
+        action_name,
+        service,
+    ):
+        spec = get_action_spec(action_name)
+
+        assert spec is not None
+        assert spec.mcp_provider == "controlled"
+        assert spec.mcp_tool == "host_ansible_reconcile"
+        assert spec.mcp_params["service"] == service
+
     def test_secops_requires_multi_sig(self):
         for sa in ("secops_isolate", "secops_block_ip", "secops_evict"):
             spec = get_action_spec(sa)
@@ -356,3 +376,50 @@ async def test_dispatch_action_injects_mcp_audit_context(monkeypatch):
     assert audit_context["flywheel_node"] == "operate"
     assert audit_context["agent_role"] == "telegram_callback_dispatcher"
     assert audit_context["operator_user_id"] == 12345
+
+
+@pytest.mark.asyncio
+async def test_host_restart_button_queues_ansible_without_ssh_provider(
+    monkeypatch,
+):
+    queue = AsyncMock(
+        return_value={
+            "status": "controlled_check_mode_queued",
+            "queued": True,
+            "automation_run_id": "run-host-111",
+            "runtime_write_performed": False,
+            "repair_executed": False,
+            "repair_verified": False,
+        }
+    )
+    provider_lookup = AsyncMock(
+        side_effect=AssertionError("Host/systemd callback must not load SSH MCP")
+    )
+    monkeypatch.setattr(
+        "src.services.host_ansible_controlled_executor."
+        "queue_host_ansible_controlled_action_for_incident_id",
+        queue,
+    )
+    monkeypatch.setattr("src.plugins.mcp.registry.get_provider", provider_lookup)
+
+    result = await dispatch_action(
+        action_name="host_restart_service",
+        incident_id="INC-HOST-111",
+        user_id=12345,
+        labels={
+            "instance": "192.168.0.111",
+            "service": "ollama-local",
+        },
+    )
+
+    assert result.success is True
+    assert "已排入 Ansible check-mode" in result.result_text
+    assert "尚未執行修復" in result.result_text
+    queue.assert_awaited_once_with(
+        incident_id="INC-HOST-111",
+        requested_action="reconcile host service ollama-local",
+        risk_level="high",
+        source="telegram_callback_host_systemd",
+        requested_host="192.168.0.111",
+    )
+    provider_lookup.assert_not_awaited()
diff --git a/apps/api/tests/test_host_ansible_controlled_executor.py b/apps/api/tests/test_host_ansible_controlled_executor.py
new file mode 100644
index 000000000..e9be4633d
--- /dev/null
+++ b/apps/api/tests/test_host_ansible_controlled_executor.py
@@ -0,0 +1,184 @@
+from __future__ import annotations
+
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+
+from src.services.host_ansible_controlled_executor import (
+    queue_host_ansible_controlled_action,
+)
+
+
+def _ollama111_incident() -> SimpleNamespace:
+    return SimpleNamespace(
+        incident_id="INC-OLLAMA111-001",
+        project_id="awoooi",
+        alertname="Ollama111Unavailable",
+        alert_category="ai_provider",
+        severity=SimpleNamespace(value="P2"),
+        affected_services=["ollama-local"],
+        signals=[
+            SimpleNamespace(
+                alert_name="Ollama111Unavailable",
+                labels={
+                    "alertname": "Ollama111Unavailable",
+                    "host": "111",
+                    "instance": "192.168.0.111",
+                    "service": "ollama-local",
+                },
+                annotations={},
+            )
+        ],
+    )
+
+
+@pytest.mark.asyncio
+async def test_host_action_queues_exact_typed_ansible_candidate() -> None:
+    enqueuer = AsyncMock(
+        return_value={
+            "schema_version": "ai_decision_controlled_executor_handoff_v1",
+            "status": "controlled_check_mode_queued",
+            "queued": True,
+            "side_effect_performed": False,
+            "single_writer_executor": "awoooi-ansible-executor-broker",
+            "automation_run_id": "run-ollama111",
+            "trace_id": "trace-ollama111",
+            "work_item_id": "AIA-SRE-008",
+            "active_blockers": [],
+        }
+    )
+
+    receipt = await queue_host_ansible_controlled_action(
+        incident=_ollama111_incident(),
+        requested_action="reconcile host service ollama-local",
+        risk_level="low",
+        source="test_host_systemd",
+        requested_host="host_111",
+        enqueuer=enqueuer,
+    )
+
+    assert receipt["queued"] is True
+    assert receipt["runtime_write_performed"] is False
+    assert receipt["repair_executed"] is False
+    assert receipt["repair_verified"] is False
+    assert receipt["incident_closure_allowed"] is False
+    assert receipt["typed_route"]["target_kind"] == "host_launchagent"
+    assert receipt["typed_route"]["executor"] == "host_ansible_executor"
+    assert receipt["typed_route"]["verifier"] == (
+        "ollama111_k3s_path_independent_verifier"
+    )
+    assert receipt["typed_route"]["allowed_catalog_ids"] == [
+        "ansible:111-ollama-fallback"
+    ]
+    enqueuer.assert_awaited_once()
+    proposal = enqueuer.await_args.kwargs["proposal_data"]
+    assert proposal["risk_level"] == "medium"
+    assert proposal["cross_domain_fallback_allowed"] is False
+    assert proposal["check_mode_required_before_apply"] is True
+
+
+@pytest.mark.asyncio
+async def test_host_action_blocks_requested_host_mismatch_without_queue() -> None:
+    enqueuer = AsyncMock()
+
+    receipt = await queue_host_ansible_controlled_action(
+        incident=_ollama111_incident(),
+        requested_action="reconcile host service ollama-local",
+        risk_level="medium",
+        source="test_host_systemd",
+        requested_host="192.168.0.110",
+        enqueuer=enqueuer,
+    )
+
+    assert receipt["queued"] is False
+    assert receipt["status"] == "requested_host_typed_route_mismatch"
+    assert receipt["runtime_write_performed"] is False
+    enqueuer.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "malformed_receipt",
+    [
+        {"queued": True},
+        {
+            "schema_version": "ai_decision_controlled_executor_handoff_v1",
+            "status": "controlled_executor_queue_failed",
+            "queued": True,
+            "side_effect_performed": False,
+            "single_writer_executor": "awoooi-ansible-executor-broker",
+            "automation_run_id": "run-1",
+            "trace_id": "trace-1",
+            "work_item_id": "AIA-SRE-008",
+            "active_blockers": [],
+        },
+        {
+            "schema_version": "ai_decision_controlled_executor_handoff_v1",
+            "status": "controlled_check_mode_queued",
+            "queued": True,
+            "side_effect_performed": True,
+            "single_writer_executor": "awoooi-ansible-executor-broker",
+            "automation_run_id": "run-1",
+            "trace_id": "trace-1",
+            "work_item_id": "AIA-SRE-008",
+            "active_blockers": [],
+        },
+    ],
+)
+async def test_malformed_broker_receipt_fails_closed(
+    malformed_receipt: dict,
+) -> None:
+    receipt = await queue_host_ansible_controlled_action(
+        incident=_ollama111_incident(),
+        requested_action="reconcile host service ollama-local",
+        risk_level="medium",
+        source="test_host_systemd",
+        requested_host="192.168.0.111",
+        enqueuer=AsyncMock(return_value=malformed_receipt),
+    )
+
+    assert receipt["queued"] is False
+    assert receipt["status"] == "controlled_executor_receipt_invalid"
+    assert receipt["runtime_write_performed"] is False
+    assert receipt["incident_closure_allowed"] is False
+
+
+@pytest.mark.asyncio
+async def test_unknown_host_asset_fails_closed_without_fallback() -> None:
+    enqueuer = AsyncMock()
+    incident = SimpleNamespace(
+        incident_id="INC-UNKNOWN-HOST",
+        project_id="awoooi",
+        alertname="HostServiceDown",
+        alert_category="host_resource",
+        severity=SimpleNamespace(value="P2"),
+        affected_services=["mystery-daemon"],
+        signals=[
+            SimpleNamespace(
+                alert_name="HostServiceDown",
+                labels={
+                    "alertname": "HostServiceDown",
+                    "instance": "192.168.0.199",
+                    "service": "mystery-daemon",
+                },
+                annotations={},
+            )
+        ],
+    )
+
+    receipt = await queue_host_ansible_controlled_action(
+        incident=incident,
+        requested_action="reconcile host service mystery-daemon",
+        risk_level="medium",
+        source="test_host_systemd",
+        requested_host="192.168.0.199",
+        enqueuer=enqueuer,
+    )
+
+    assert receipt["queued"] is False
+    assert receipt["status"] == "asset_identity_unresolved"
+    assert receipt["typed_route"]["drift_work_item_id"].startswith(
+        "AIA-ASSET-DRIFT-"
+    )
+    enqueuer.assert_not_awaited()
diff --git a/apps/api/tests/test_ssh_provider_tools.py b/apps/api/tests/test_ssh_provider_tools.py
index 1a181652f..c61f85257 100644
--- a/apps/api/tests/test_ssh_provider_tools.py
+++ b/apps/api/tests/test_ssh_provider_tools.py
@@ -94,6 +94,50 @@ async def test_ssh_execute_normalizes_host_before_allowed_check(monkeypatch):
     assert isinstance(captured["timeout"], int)
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "tool_name,parameters",
+    [
+        (
+            "ssh_systemctl_restart",
+            {"service": "node_exporter"},
+        ),
+        ("ssh_reload_nginx", {}),
+    ],
+)
+async def test_systemd_tools_are_compatibility_only_no_ssh(
+    monkeypatch,
+    tool_name,
+    parameters,
+):
+    provider = SSHProvider()
+    ssh_exec_called = False
+
+    async def fail_if_called(*_args, **_kwargs):
+        nonlocal ssh_exec_called
+        ssh_exec_called = True
+        raise AssertionError("direct systemctl SSH must remain disabled")
+
+    monkeypatch.setattr(provider, "_ssh_exec", fail_if_called)
+
+    result = await provider.execute(
+        tool_name,
+        {
+            "host": "192.168.0.110",
+            "trust_score": 1.0,
+            **parameters,
+        },
+    )
+
+    assert result.success is False
+    assert result.error == (
+        f"{tool_name}_disabled_use_host_ansible_controlled_executor"
+    )
+    assert ssh_exec_called is False
+    with pytest.raises(ValueError, match="host_ansible_controlled_executor"):
+        provider._build_command(tool_name, parameters)
+
+
 def test_ssh_container_status_accepts_legacy_container_name_alias():
     provider = SSHProvider()
 
diff --git a/apps/web/messages/en.json b/apps/web/messages/en.json
index 26afdec3f..539ad8ba5 100644
--- a/apps/web/messages/en.json
+++ b/apps/web/messages/en.json
@@ -2296,6 +2296,9 @@
     "execCount": "執行次數",
     "decisionReason": "決策原因",
     "execSuccess": "執行成功 ({ms}ms)",
+    "execQueued": "Queued for controlled execution; not executed or verified yet",
+    "execVerificationPending": "Repair executed; awaiting independent verification",
+    "execVerified": "Repair completed and independently verified ({ms}ms)",
     "execFailed": "執行失敗: {error}",
     "executing": "執行中...",
     "execute": "執行修復",
diff --git a/apps/web/messages/zh-TW.json b/apps/web/messages/zh-TW.json
index 8ca1303f3..60818f432 100644
--- a/apps/web/messages/zh-TW.json
+++ b/apps/web/messages/zh-TW.json
@@ -2296,6 +2296,9 @@
     "execCount": "執行次數",
     "decisionReason": "決策原因",
     "execSuccess": "執行成功 ({ms}ms)",
+    "execQueued": "已排入受控執行佇列；尚未執行、尚未驗證",
+    "execVerificationPending": "修復已執行；等待獨立驗證",
+    "execVerified": "修復完成並通過獨立驗證 ({ms}ms)",
     "execFailed": "執行失敗: {error}",
     "executing": "執行中...",
     "execute": "執行修復",
diff --git a/apps/web/src/components/panels/AutoRepairPanel.tsx b/apps/web/src/components/panels/AutoRepairPanel.tsx
index f4126fab7..b043deeb1 100644
--- a/apps/web/src/components/panels/AutoRepairPanel.tsx
+++ b/apps/web/src/components/panels/AutoRepairPanel.tsx
@@ -13,6 +13,7 @@
 import { useState, useEffect, useCallback, useRef } from 'react'
 import { useTranslations } from 'next-intl'
 import { useIncidents } from '@/hooks/useIncidents'
+import { classifyAutoRepairResult } from '@/lib/auto-repair-result'
 import { cn } from '@/lib/utils'
 import {
   Wrench, RefreshCw, AlertCircle,
@@ -56,6 +57,9 @@ interface EvaluateResponse {
 
 interface ExecuteResponse {
   success: boolean
+  status: string
+  repair_executed: boolean
+  repair_verified: boolean
   incident_id: string
   playbook_id: string
   executed_steps: string[]
@@ -94,6 +98,7 @@ function IncidentEvalRow({
   const [executing, setExecuting] = useState(false)
   const [result, setResult] = useState<ExecuteResponse | null>(null)
   const [expanded, setExpanded] = useState(false)
+  const resultState = result ? classifyAutoRepairResult(result) : null
 
   const fetchEval = useCallback(async () => {
     const base = getApiBaseUrl()
@@ -193,14 +198,33 @@ function IncidentEvalRow({
           {result && (
             <div className={cn(
               'rounded-lg border p-4',
-              result.success ? 'bg-[#f0faf2] border-[#8fc29a]' : 'bg-[#fff1f0] border-[#e2a29b]'
+              resultState === 'verified'
+                ? 'bg-[#f0faf2] border-[#8fc29a]'
+                : resultState === 'failed'
+                  ? 'bg-[#fff1f0] border-[#e2a29b]'
+                  : 'bg-[#fff7e8] border-[#d9b36f]'
             )}>
               <div className="mb-3 flex items-center gap-2">
-                {result.success
+                {resultState === 'verified'
                   ? <CheckCircle2 className="h-5 w-5 text-[#17602a]" />
-                  : <XCircle className="h-5 w-5 text-[#9f2f25]" />}
-                <span className={cn('font-mono text-sm font-bold', result.success ? 'text-[#17602a]' : 'text-[#9f2f25]')}>
-                  {result.success ? t('execSuccess', { ms: result.execution_time_ms }) : t('execFailed', { error: result.error ?? '' })}
+                  : resultState === 'failed'
+                    ? <XCircle className="h-5 w-5 text-[#9f2f25]" />
+                    : <Activity className="h-5 w-5 text-[#8a5a08]" />}
+                <span className={cn(
+                  'font-mono text-sm font-bold',
+                  resultState === 'verified'
+                    ? 'text-[#17602a]'
+                    : resultState === 'failed'
+                      ? 'text-[#9f2f25]'
+                      : 'text-[#8a5a08]'
+                )}>
+                  {resultState === 'queued'
+                    ? t('execQueued')
+                    : resultState === 'verification_pending'
+                      ? t('execVerificationPending')
+                      : resultState === 'verified'
+                        ? t('execVerified', { ms: result.execution_time_ms })
+                        : t('execFailed', { error: result.error ?? '' })}
                 </span>
               </div>
               {result.executed_steps.length > 0 && (
diff --git a/apps/web/src/lib/__tests__/auto-repair-result.test.ts b/apps/web/src/lib/__tests__/auto-repair-result.test.ts
new file mode 100644
index 000000000..b1b4167a9
--- /dev/null
+++ b/apps/web/src/lib/__tests__/auto-repair-result.test.ts
@@ -0,0 +1,41 @@
+import { describe, expect, it } from 'vitest'
+
+import { classifyAutoRepairResult } from '../auto-repair-result'
+
+describe('classifyAutoRepairResult', () => {
+  it('keeps an accepted queue neutral instead of reporting failure', () => {
+    expect(classifyAutoRepairResult({
+      success: false,
+      status: 'queued',
+      repair_executed: false,
+      repair_verified: false,
+    })).toBe('queued')
+  })
+
+  it('does not mark an unverified runtime write green', () => {
+    expect(classifyAutoRepairResult({
+      success: true,
+      status: 'completed',
+      repair_executed: true,
+      repair_verified: false,
+    })).toBe('verification_pending')
+  })
+
+  it('requires independent verification for verified state', () => {
+    expect(classifyAutoRepairResult({
+      success: true,
+      status: 'completed',
+      repair_executed: true,
+      repair_verified: true,
+    })).toBe('verified')
+  })
+
+  it('keeps failed and malformed receipts red', () => {
+    expect(classifyAutoRepairResult({
+      success: false,
+      status: 'failed',
+      repair_executed: false,
+      repair_verified: false,
+    })).toBe('failed')
+  })
+})
diff --git a/apps/web/src/lib/auto-repair-result.ts b/apps/web/src/lib/auto-repair-result.ts
new file mode 100644
index 000000000..c86744358
--- /dev/null
+++ b/apps/web/src/lib/auto-repair-result.ts
@@ -0,0 +1,27 @@
+export type AutoRepairResultState =
+  | 'queued'
+  | 'verification_pending'
+  | 'verified'
+  | 'failed'
+
+export interface AutoRepairResultTruth {
+  success: boolean
+  status: string
+  repair_executed: boolean
+  repair_verified: boolean
+}
+
+export function classifyAutoRepairResult(
+  result: AutoRepairResultTruth,
+): AutoRepairResultState {
+  if (result.status === 'queued') return 'queued'
+  if (
+    result.status === 'completed'
+    && result.repair_executed
+    && result.repair_verified
+  ) return 'verified'
+  if (result.status === 'completed' && result.repair_executed) {
+    return 'verification_pending'
+  }
+  return 'failed'
+}