feat(adr100): surface replay execution gate

apps/api/src/services/adr100_remediation_service.py

@@ -65,6 +65,7 @@ class Adr100RemediationService:
         slo_service: Adr100SloStatusService | None = None,
         incident_repository: _IncidentRepository | None = None,
         auto_repair_service: AutoRepairService | None = None,
+        playbook_service: Any | None = None,
         verifier: PostExecutionVerifier | None = None,
         approval_service: Any | None = None,
         timeline_service: Any | None = None,
@@ -74,6 +75,7 @@ class Adr100RemediationService:
         self._slo_service = slo_service or get_adr100_slo_status_service()
         self._incident_repository = incident_repository or IncidentDBRepository()
         self._auto_repair_service = auto_repair_service or AutoRepairService()
+        self._playbook_service = playbook_service
         self._verifier = verifier or get_post_execution_verifier()
         self._approval_service = approval_service
         self._timeline_service = timeline_service
@@ -323,6 +325,7 @@ class Adr100RemediationService:
         })
 
         post_state = await self._collect_current_state(incident)
+        replay_gate = await self._build_replay_gate(item, incident)
         action_taken = f"dry_run_replay:{item.get('playbook_id') or 'unknown'}"
         result = _assess_recovery(None, post_state, action_taken)
 
@@ -335,12 +338,125 @@ class Adr100RemediationService:
             extra={
                 "diagnostic_command_preview": diagnostic_command,
                 "mcp_route": route,
+                "replay_gate": replay_gate,
                 "promql": _promql_for_incident(incident),
             },
         )
         payload["history"] = await self._record_dry_run_history(item, payload)
         return payload
 
+    async def _build_replay_gate(
+        self,
+        item: dict[str, Any],
+        incident: Incident,
+    ) -> dict[str, Any]:
+        playbook_id = str(item.get("playbook_id") or "")
+        base: dict[str, Any] = {
+            "schema_version": "adr100_replay_gate_v1",
+            "playbook_id": playbook_id or None,
+            "playbook_loaded": False,
+            "status": "blocked_missing_playbook_id",
+            "execution_authorized": False,
+            "repair_executed": False,
+            "writes_incident_state": False,
+            "writes_auto_repair_result": False,
+            "can_runtime_replay": False,
+            "mutating_step_count": 0,
+            "supported_write_route_count": 0,
+            "unsupported_step_count": 0,
+            "approval_required_count": 0,
+            "next_step": "attach_playbook_to_work_item",
+            "steps": [],
+        }
+        if not playbook_id:
+            return base
+
+        playbook_svc = self._playbook_service
+        if playbook_svc is None:
+            from src.services.playbook_service import get_playbook_service
+
+            playbook_svc = get_playbook_service()
+
+        playbook = await playbook_svc.get_by_id(playbook_id)
+        if playbook is None:
+            return {
+                **base,
+                "status": "blocked_playbook_not_found",
+                "next_step": "reload_playbook_catalog_or_author_supported_executor",
+            }
+
+        steps: list[dict[str, Any]] = []
+        mutating_count = 0
+        supported_write_count = 0
+        unsupported_count = 0
+        approval_required_count = 0
+
+        for step in playbook.repair_steps:
+            command = str(getattr(step, "command", "") or "")
+            action_type = _step_action_type(step)
+            write_route = None
+            if action_type == "ssh_command":
+                write_route = self._auto_repair_service.preview_write_ssh_mcp_route(
+                    incident,
+                    command,
+                )
+            read_route = self._auto_repair_service.preview_read_only_ssh_mcp_route(
+                incident,
+                command,
+            ) if action_type == "ssh_command" else None
+            is_mutating = bool(write_route) or _step_looks_mutating(step)
+            requires_approval = bool(getattr(step, "requires_approval", False))
+            if is_mutating:
+                mutating_count += 1
+                if write_route:
+                    supported_write_count += 1
+                else:
+                    unsupported_count += 1
+                if requires_approval:
+                    approval_required_count += 1
+
+            steps.append({
+                "step_number": getattr(step, "step_number", None),
+                "action_type": action_type,
+                "risk_level": _step_risk_level(step),
+                "requires_approval": requires_approval,
+                "is_mutating": is_mutating,
+                "write_route": write_route,
+                "read_route": read_route,
+                "supported": bool(write_route) or not is_mutating,
+                "command_preview": _compact_command(command),
+            })
+
+        if mutating_count == 0:
+            status = "blocked_observe_only_playbook"
+            next_step = "author_mutating_repair_step"
+            can_runtime_replay = False
+        elif unsupported_count > 0:
+            status = "blocked_unsupported_write_route"
+            next_step = "author_supported_executor_step"
+            can_runtime_replay = False
+        elif approval_required_count > 0:
+            status = "approval_required"
+            next_step = "request_runtime_replay_approval"
+            can_runtime_replay = False
+        else:
+            status = "runtime_replay_ready"
+            next_step = "queue_runtime_replay_with_gate5_projection"
+            can_runtime_replay = True
+
+        return {
+            **base,
+            "playbook_loaded": True,
+            "status": status,
+            "can_runtime_replay": can_runtime_replay,
+            "mutating_step_count": mutating_count,
+            "supported_write_route_count": supported_write_count,
+            "unsupported_step_count": unsupported_count,
+            "approval_required_count": approval_required_count,
+            "next_step": next_step,
+            "steps": steps,
+        }
+
     async def _dry_run_ticket_proposal(
         self,
         item: dict[str, Any],
@@ -842,6 +958,48 @@ def _summarize_post_state(post_state: dict[str, Any]) -> dict[str, Any]:
     }
 
 
+def _step_action_type(step: Any) -> str:
+    action_type = getattr(step, "action_type", None)
+    return str(getattr(action_type, "value", action_type) or "unknown")
+
+
+def _step_risk_level(step: Any) -> str:
+    risk_level = getattr(step, "risk_level", None)
+    return str(getattr(risk_level, "value", risk_level) or "unknown").lower()
+
+
+def _step_looks_mutating(step: Any) -> bool:
+    command = str(getattr(step, "command", "") or "").strip().lower()
+    action_type = _step_action_type(step)
+    if not command or action_type == "manual":
+        return False
+    if action_type == "script":
+        return True
+    if action_type == "kubectl":
+        return not command.startswith((
+            "kubectl get ",
+            "kubectl describe ",
+            "kubectl logs ",
+            "kubectl top ",
+            "kubectl explain ",
+        ))
+    if action_type == "ssh_command":
+        return any(token in command for token in (
+            "docker restart",
+            "docker start",
+            "docker stop",
+            "systemctl restart",
+            "systemctl start",
+            "certbot renew",
+        ))
+    return False
+
+
+def _compact_command(command: str) -> str:
+    compact = " ".join(str(command or "").split())
+    return compact[:240]
+
+
 def _history_context(item: dict[str, Any], payload: dict[str, Any]) -> dict[str, Any]:
     return {
         "schema_version": "adr100_remediation_dry_run_history_v1",
@@ -862,6 +1020,7 @@ def _history_context(item: dict[str, Any], payload: dict[str, Any]) -> dict[str,
         "verification_result_preview": payload.get("verification_result_preview"),
         "post_state_summary": payload.get("post_state_summary"),
         "mcp_route": payload.get("mcp_route"),
+        "replay_gate": payload.get("replay_gate"),
         "checks": payload.get("checks"),
     }
 
@@ -938,6 +1097,7 @@ def _history_item(record: Any, context: dict[str, Any]) -> dict[str, Any]:
     route = context.get("mcp_route") or {}
     post_state = context.get("post_state_summary") or {}
     approval = context.get("approval") or {}
+    replay_gate = context.get("replay_gate") or {}
     return {
         "id": str(getattr(record, "id", "")),
         "incident_id": getattr(record, "incident_id", None),
@@ -971,6 +1131,9 @@ def _history_item(record: Any, context: dict[str, Any]) -> dict[str, Any]:
         "deduplicated": context.get("deduplicated"),
         "fingerprint": context.get("fingerprint"),
         "ticket_preview": context.get("ticket_preview"),
+        "replay_gate": replay_gate or None,
+        "replay_gate_status": replay_gate.get("status"),
+        "replay_gate_next_step": replay_gate.get("next_step"),
         "plan": context.get("plan"),
         "checks": context.get("checks") or [],
     }
@@ -993,6 +1156,8 @@ def _summarize_history_by_work_item(items: list[dict[str, Any]]) -> list[dict[st
                 "latest_agent_id": item.get("agent_id"),
                 "latest_tool_name": item.get("tool_name"),
                 "required_scope": item.get("required_scope"),
+                "latest_replay_gate_status": item.get("replay_gate_status"),
+                "latest_replay_gate_next_step": item.get("replay_gate_next_step"),
             }
         summary[key]["count"] += 1
     return list(summary.values())

apps/api/src/services/auto_repair_service.py

@@ -1237,6 +1237,30 @@ class AutoRepairService:
             "flywheel_node": "execute",
         }
 
+    def preview_write_ssh_mcp_route(
+        self,
+        incident: Incident,
+        command: str,
+    ) -> dict[str, Any] | None:
+        """Preview whether a legacy SSH repair can use the write MCP Gateway.
+
+        This mirrors the executor's narrow Docker restart routing without
+        projecting a Gate 5 approval or mutating any runtime state.  It lets
+        ADR-100 Work Items show whether ``ready_for_replay`` is actually ready
+        for a governed write executor.
+        """
+
+        route = self._route_legacy_ssh_write_command_to_mcp(incident, command)
+        if route is None:
+            return None
+        return {
+            "tool_name": route.tool_name,
+            "params": route.params,
+            "agent_id": "auto_repair_executor",
+            "required_scope": route.required_scope,
+            "flywheel_node": "execute",
+        }
+
     def _resolve_ssh_host_for_incident(self, incident: Incident, command: str) -> str:
         """Resolve ``{host}``, short host labels, and exporter instance ports."""