feat(adr100): surface playbook ticket remediation

apps/api/src/services/adr100_remediation_service.py

@@ -31,9 +31,11 @@ from src.services.post_execution_verifier import (
 
 logger = structlog.get_logger(__name__)
 
-RemediationMode = Literal["auto", "reverify", "replay"]
+RemediationMode = Literal["auto", "reverify", "replay", "ticket"]
 
 _READY_STATUSES = {"ready_for_replay", "ready_for_reverify"}
+_TICKET_STATUSES = {"needs_playbook_ticket"}
+_TICKET_ACTIONS = {"create_playbook_ticket", "promote_diagnostic_to_repair_playbook"}
 
 
 class RemediationNotFoundError(LookupError):
@@ -108,6 +110,8 @@ class Adr100RemediationService:
             payload["history"] = await self._record_dry_run_history(item, payload)
             return payload
 
+        if selected_mode == "ticket":
+            return await self._dry_run_ticket_proposal(item, incident, checks)
         if selected_mode == "replay":
             return await self._dry_run_replay(item, incident, checks)
         return await self._dry_run_reverify(item, incident, checks)
@@ -255,6 +259,35 @@ class Adr100RemediationService:
         payload["history"] = await self._record_dry_run_history(item, payload)
         return payload
 
+    async def _dry_run_ticket_proposal(
+        self,
+        item: dict[str, Any],
+        incident: Incident,
+        checks: list[dict[str, Any]],
+    ) -> dict[str, Any]:
+        ticket_preview = _ticket_preview_for_item(item, incident)
+        checks.append({
+            "name": "external_ticket_not_created",
+            "passed": True,
+            "detail": "dry_run_records_internal_history_only",
+        })
+
+        payload = _dry_run_result_payload(
+            item=item,
+            mode="ticket",
+            checks=checks,
+            post_state={},
+            verification_result_preview="ticket_proposal",
+            extra={
+                "ticket_preview": ticket_preview,
+                "writes_ticket": False,
+                "creates_external_ticket": False,
+                "plan": _plan_for_item(item, "ticket"),
+            },
+        )
+        payload["history"] = await self._record_dry_run_history(item, payload)
+        return payload
+
     async def _collect_current_state(self, incident: Incident) -> dict[str, Any]:
         try:
             return await asyncio.wait_for(
@@ -351,9 +384,15 @@ class Adr100RemediationService:
         return history
 
 
-def _select_mode(item: dict[str, Any], requested: RemediationMode) -> Literal["reverify", "replay"]:
+def _select_mode(item: dict[str, Any], requested: RemediationMode) -> Literal["reverify", "replay", "ticket"]:
     if requested in ("reverify", "replay"):
         return requested
+    if requested == "ticket":
+        return "ticket"
+    if item.get("remediation_status") in _TICKET_STATUSES:
+        return "ticket"
+    if item.get("remediation_action") in _TICKET_ACTIONS:
+        return "ticket"
     if item.get("remediation_status") == "ready_for_reverify":
         return "reverify"
     if item.get("remediation_action") == "reverify_with_promql_template":
@@ -367,14 +406,15 @@ def _base_checks(item: dict[str, Any]) -> list[dict[str, Any]]:
     return [
         {
             "name": "queue_item_ready",
-            "passed": status in _READY_STATUSES,
+            "passed": status in _READY_STATUSES or status in _TICKET_STATUSES,
             "detail": status,
         },
         {
-            "name": "read_only_guardrail",
+            "name": "read_or_record_only_guardrail",
             "passed": action in {
                 "replay_with_supported_executor",
                 "reverify_with_promql_template",
+                *_TICKET_ACTIONS,
             },
             "detail": action,
         },
@@ -394,6 +434,14 @@ def _plan_for_item(item: dict[str, Any], mode: str) -> dict[str, Any]:
             "required_scope": "read",
             "writes": [],
         }
+    if mode == "ticket":
+        return {
+            "step": "create_playbook_authoring_ticket_proposal",
+            "agent_id": "openclaw_playbook_planner",
+            "required_scope": "record_only",
+            "writes": ["alert_operation_log", "timeline"],
+            "target_action": item.get("remediation_action"),
+        }
     return {
         "step": "validate_supported_executor_route_then_collect_current_state",
         "agent_id": "auto_repair_executor",
@@ -419,6 +467,8 @@ def _dry_run_blocked_payload(
         "safety_level": "read_only",
         "writes_incident_state": False,
         "writes_auto_repair_result": False,
+        "writes_ticket": False,
+        "creates_external_ticket": False,
         "checks": checks,
         "verification_result_preview": "blocked",
         "post_state_summary": {},
@@ -445,6 +495,8 @@ def _dry_run_result_payload(
         "safety_level": "read_only",
         "writes_incident_state": False,
         "writes_auto_repair_result": False,
+        "writes_ticket": extra.get("writes_ticket", False),
+        "creates_external_ticket": extra.get("creates_external_ticket", False),
         "checks": checks,
         "verification_result_preview": verification_result_preview,
         "post_state_summary": _summarize_post_state(post_state),
@@ -474,6 +526,10 @@ def _history_context(item: dict[str, Any], payload: dict[str, Any]) -> dict[str,
         "safety_level": payload.get("safety_level"),
         "writes_incident_state": payload.get("writes_incident_state"),
         "writes_auto_repair_result": payload.get("writes_auto_repair_result"),
+        "writes_ticket": payload.get("writes_ticket"),
+        "creates_external_ticket": payload.get("creates_external_ticket"),
+        "ticket_preview": payload.get("ticket_preview"),
+        "plan": payload.get("plan"),
         "verification_result_preview": payload.get("verification_result_preview"),
         "post_state_summary": payload.get("post_state_summary"),
         "mcp_route": payload.get("mcp_route"),
@@ -537,6 +593,10 @@ def _history_item(record: Any, context: dict[str, Any]) -> dict[str, Any]:
         "required_scope": route.get("required_scope"),
         "writes_incident_state": context.get("writes_incident_state"),
         "writes_auto_repair_result": context.get("writes_auto_repair_result"),
+        "writes_ticket": context.get("writes_ticket"),
+        "creates_external_ticket": context.get("creates_external_ticket"),
+        "ticket_preview": context.get("ticket_preview"),
+        "plan": context.get("plan"),
         "checks": context.get("checks") or [],
     }
 
@@ -572,6 +632,44 @@ def _diagnostic_command_for_incident(incident: Incident) -> str:
     return f"ssh {host} 'uptime; docker stats --no-stream'"
 
 
+def _ticket_preview_for_item(item: dict[str, Any], incident: Incident) -> dict[str, Any]:
+    labels = _labels_for_incident(incident)
+    alertname = str(item.get("alertname") or labels.get("alertname") or "unknown_alert")
+    incident_id = str(item.get("incident_id") or incident.incident_id)
+    playbook_id = str(item.get("playbook_id") or "unknown_playbook")
+    host = str(labels.get("host") or labels.get("instance") or "unknown_host")
+    container = str(labels.get("container_name") or labels.get("container") or "")
+    target = f"host={host}" + (f" container={container}" if container else "")
+    title = f"[ADR-100] Promote diagnostic PlayBook to repair: {alertname}"
+    body = (
+        f"Incident: {incident_id}\n"
+        f"Auto repair: {item.get('auto_repair_id') or 'unknown'}\n"
+        f"PlayBook: {playbook_id}\n"
+        f"Target: {target}\n"
+        f"Failure class: {item.get('failure_class') or 'observe_only_playbook'}\n"
+        "Required change: add a gated mutating repair step such as docker restart, "
+        "Ansible check-mode/apply, or another approved executor action, then keep "
+        "post-execution verification tied to the same target.\n"
+        "Guardrail: do not mark the old diagnostic-only run as verified_success."
+    )
+    return {
+        "would_create": True,
+        "external_ticket_created": False,
+        "title": title,
+        "labels": [
+            "adr100",
+            "playbook-authoring",
+            "observe-only-playbook",
+            "needs-owner-review",
+        ],
+        "body_preview": body[:1000],
+        "owner": item.get("remediation_owner") or "solver_or_operator",
+        "next_step": "author_mutating_repair_step",
+        "playbook_id": playbook_id,
+        "target": target,
+    }
+
+
 def _promql_for_incident(incident: Incident) -> str:
     labels = _labels_for_incident(incident)
     alertname = ""