feat(awooop): expose repair candidate promotion contract

apps/api/src/api/v1/telegram.py

@@ -180,6 +180,19 @@ def _build_no_action_manual_handoff_payload(approval) -> dict:
         or _safe_str(metadata.get("repair_candidate_status"))
         or "repair_candidate_missing"
     )
+    promotion_contract = _safe_dict(
+        package.get("candidate_promotion_contract")
+        or metadata.get("repair_candidate_promotion_contract")
+    )
+    promotion_summary = _safe_str(metadata.get("repair_candidate_promotion_summary"))
+    if not promotion_summary and promotion_contract:
+        promotion_summary = (
+            f"route={promotion_contract.get('route_id') or '--'}; "
+            f"promotion={promotion_contract.get('ready_count') or 0}/"
+            f"{promotion_contract.get('total_count') or 0}; "
+            f"blocked={promotion_contract.get('blocked_count') or 0}; "
+            f"runtime=false"
+        )
 
     return {
         "message": "ApprovedForOwnerReviewHandoff" if draft_ready else "ApprovedForManualHandoff",
@@ -203,6 +216,8 @@ def _build_no_action_manual_handoff_payload(approval) -> dict:
         "work_item_id": work_item_id,
         "work_item_href": work_item_href,
         "repair_candidate_blocker": blocker,
+        "repair_candidate_promotion_summary": promotion_summary,
+        "repair_candidate_promotion_contract": promotion_contract,
         "required_fields": _safe_str_list(package.get("required_fields")),
         "blocked_operations": _safe_str_list(package.get("blocked_operations")),
         "required_writebacks": _safe_str_list(package.get("required_writebacks")),

apps/api/src/api/v1/webhooks.py

@@ -599,6 +599,7 @@ async def _push_to_telegram_background(
     repair_candidate_blocker_summary: str = "",
     repair_candidate_next_step: str = "",
     repair_candidate_required_fields: list[str] | None = None,
+    repair_candidate_promotion_summary: str = "",
     repair_candidate_work_item_href: str = "",
     repair_candidate_work_item_id: str = "",
 ) -> None:
@@ -697,6 +698,7 @@ async def _push_to_telegram_background(
             repair_candidate_blocker_summary=repair_candidate_blocker_summary,
             repair_candidate_next_step=repair_candidate_next_step,
             repair_candidate_required_fields=repair_candidate_required_fields,
+            repair_candidate_promotion_summary=repair_candidate_promotion_summary,
             repair_candidate_work_item_href=repair_candidate_work_item_href,
             repair_candidate_work_item_id=repair_candidate_work_item_id,
         )
@@ -2537,6 +2539,9 @@ async def _process_new_alert_background(
                     if isinstance(_approval_metadata_cs4.get("repair_candidate_draft_package"), dict)
                     else []
                 ),
+                repair_candidate_promotion_summary=str(
+                    _approval_metadata_cs4.get("repair_candidate_promotion_summary") or ""
+                ),
                 repair_candidate_work_item_href=str(
                     (
                         _approval_metadata_cs4.get("repair_candidate_draft_package", {})

apps/api/src/services/repair_candidate_service.py

@@ -375,6 +375,12 @@ class RepairCandidateService:
                 work_item["next_action"] = "owner_review_repair_candidate_draft"
                 work_item["owner_review_required"] = True
                 work_item["runtime_execution_authorized"] = False
+        promotion_contract = draft_package.get("candidate_promotion_contract")
+        if isinstance(promotion_contract, dict):
+            metadata["repair_candidate_promotion_contract"] = promotion_contract
+            metadata["repair_candidate_promotion_summary"] = (
+                self._promotion_summary_for_operator(promotion_contract)
+            )
         metadata["playbook_draft_required"] = True
         metadata["repair_candidate_prefilled_draft_summary"] = self._draft_summary_for_operator(
             draft_package.get("playbook_draft_template") or {}
@@ -576,6 +582,8 @@ class RepairCandidateService:
         incident_id = getattr(incident, "incident_id", "") if incident else ""
         project_id = str(getattr(incident, "project_id", "awoooi") or "awoooi")
         work_item: dict[str, Any] | None = None
+        work_item_id = ""
+        work_item_url_value = ""
         if incident_id:
             work_item_id = f"repair-candidate-draft:{project_id}:{incident_id}:{lane}"
             work_item_query = urlencode({
@@ -587,6 +595,11 @@ class RepairCandidateService:
                 "project_id": project_id,
                 "incident_id": incident_id,
             })
+            work_item_url_value = work_item_url(
+                work_item_id,
+                incident_id=incident_id,
+                project_id=project_id,
+            )
             work_item = {
                 "schema_version": "awooop_repair_candidate_draft_work_item_v1",
                 "work_item_id": work_item_id,
@@ -609,11 +622,7 @@ class RepairCandidateService:
                 "blocked_operations": blocked_operations,
                 "target_href": f"/awooop/runs?{run_query}",
                 "work_item_href": f"/awooop/work-items?{work_item_query}",
-                "work_item_url": work_item_url(
-                    work_item_id,
-                    incident_id=incident_id,
-                    project_id=project_id,
-                ),
+                "work_item_url": work_item_url_value,
                 "decision_effect": "none",
                 "safety_level": "read_only_work_item_projection",
                 "writes_incident_state": False,
@@ -621,6 +630,19 @@ class RepairCandidateService:
                 "writes_runtime_state": False,
             }
 
+        promotion_contract = self._build_candidate_promotion_contract(
+            coverage_gap=coverage_gap,
+            playbook_draft_template=playbook_draft_template,
+            lane=lane,
+            blockers=blockers,
+            incident_id=incident_id,
+            project_id=project_id,
+            work_item_id=work_item_id,
+            work_item_url_value=work_item_url_value,
+        )
+        if work_item is not None:
+            work_item["candidate_promotion_contract"] = promotion_contract
+
         package = {
             "schema_version": "repair_candidate_draft_package_v1",
             "status": "draft_required",
@@ -634,12 +656,222 @@ class RepairCandidateService:
             "required_writebacks": required_writebacks,
             "coverage_gap": coverage_gap,
             "playbook_draft_template": playbook_draft_template,
+            "candidate_promotion_contract": promotion_contract,
             "blocked_operations": blocked_operations,
         }
         if work_item:
             package["awooop_work_item"] = work_item
         return package
 
+    def _build_candidate_promotion_contract(
+        self,
+        *,
+        coverage_gap: dict[str, Any],
+        playbook_draft_template: dict[str, Any],
+        lane: str,
+        blockers: list[str],
+        incident_id: str,
+        project_id: str,
+        work_item_id: str,
+        work_item_url_value: str,
+    ) -> dict[str, Any]:
+        """Describe exactly what is needed to promote a draft into an apply gate.
+
+        This contract is still fail-closed. It gives AwoooP / Telegram a
+        machine-readable checklist so a draft does not collapse back into
+        generic "manual review" text.
+        """
+
+        route = str(playbook_draft_template.get("suggested_route") or "").strip()
+        repair_template = str(
+            playbook_draft_template.get("repair_command_template") or ""
+        ).strip()
+        rollback_template = str(
+            playbook_draft_template.get("rollback_command_template") or ""
+        ).strip()
+        verifier_plan = playbook_draft_template.get("verifier_plan_template")
+        if not isinstance(verifier_plan, list):
+            verifier_plan = []
+        evidence_refs = playbook_draft_template.get("mcp_evidence_refs")
+        if not isinstance(evidence_refs, list):
+            evidence_refs = []
+        alert_selector = playbook_draft_template.get("alert_selector")
+        if not isinstance(alert_selector, dict):
+            alert_selector = {}
+
+        field_rows = [
+            self._promotion_contract_field(
+                field="target_selector",
+                label="Alert / target selector",
+                status="ready" if alert_selector.get("target_resource") else "blocked",
+                source="coverage_gap",
+                value=coverage_gap.get("coverage_key"),
+            ),
+            self._promotion_contract_field(
+                field="mcp_evidence_refs",
+                label="MCP evidence refs",
+                status=(
+                    "ready"
+                    if coverage_gap.get("mcp_evidence_ready") and evidence_refs
+                    else "blocked"
+                ),
+                source="mcp_evidence",
+                value=evidence_refs[:8],
+            ),
+            self._promotion_contract_field(
+                field="route_id",
+                label="Safe route after owner review",
+                status="ready" if route.endswith("_after_owner_review") else "blocked",
+                source="playbook_draft_template",
+                value=route or "--",
+            ),
+            self._promotion_contract_field(
+                field="repair_command_template",
+                label="Repair command template",
+                status="ready"
+                if repair_template
+                and "<" not in repair_template
+                and not repair_template.startswith("owner_supplied")
+                else "blocked",
+                source="playbook_draft_template",
+                value=repair_template or "--",
+            ),
+            self._promotion_contract_field(
+                field="rollback_command_template",
+                label="Rollback command template",
+                status="ready"
+                if rollback_template
+                and "<" not in rollback_template
+                and not rollback_template.startswith("owner_supplied")
+                else "blocked",
+                source="playbook_draft_template",
+                value=rollback_template or "--",
+            ),
+            self._promotion_contract_field(
+                field="verifier_plan",
+                label="Post-apply verifier plan",
+                status="ready" if verifier_plan else "blocked",
+                source="playbook_draft_template",
+                value=verifier_plan[:8],
+            ),
+            self._promotion_contract_field(
+                field="owner_review",
+                label="Owner review release",
+                status="blocked",
+                source="owner_response",
+                value="required_before_apply_gate",
+            ),
+            self._promotion_contract_field(
+                field="maintenance_window",
+                label="Maintenance window",
+                status="blocked",
+                source="owner_response",
+                value="required_before_runtime_write",
+            ),
+            self._promotion_contract_field(
+                field="blast_radius",
+                label="Blast radius",
+                status="blocked",
+                source="owner_response",
+                value="required_before_runtime_write",
+            ),
+            self._promotion_contract_field(
+                field="km_writeback_owner",
+                label="KM writeback owner",
+                status="blocked",
+                source="owner_response",
+                value="required_before_closure",
+            ),
+            self._promotion_contract_field(
+                field="playbook_trust_owner",
+                label="PlayBook trust owner",
+                status="blocked",
+                source="owner_response",
+                value="required_before_trust_raise",
+            ),
+        ]
+        ready_fields = [row["field"] for row in field_rows if row["status"] == "ready"]
+        blocked_fields = [row["field"] for row in field_rows if row["status"] != "ready"]
+        status = (
+            "owner_review_ready_runtime_blocked"
+            if {
+                "target_selector",
+                "mcp_evidence_refs",
+                "route_id",
+                "repair_command_template",
+            } <= set(ready_fields)
+            else "blocked_missing_candidate_inputs"
+        )
+        return {
+            "schema_version": "repair_candidate_promotion_contract_v1",
+            "status": status,
+            "lane": lane,
+            "incident_id": incident_id or None,
+            "project_id": project_id,
+            "source_work_item_id": work_item_id or None,
+            "source_work_item_url": work_item_url_value or None,
+            "route_id": route or "--",
+            "repair_command_template": repair_template or "--",
+            "rollback_command_template": rollback_template or "--",
+            "verifier_plan_template": list(verifier_plan),
+            "ready_count": len(ready_fields),
+            "total_count": len(field_rows),
+            "blocked_count": len(blocked_fields),
+            "ready_fields": ready_fields,
+            "blocked_fields": blocked_fields,
+            "fields": field_rows,
+            "blockers": list(dict.fromkeys(blockers)),
+            "runtime_write_allowed": False,
+            "runtime_execution_authorized": False,
+            "approval_required_before_execution": True,
+            "owner_review_required": True,
+            "forbidden_until_promoted": [
+                "auto_execute",
+                "systemctl_restart",
+                "ssh_write",
+                "ansible_apply",
+                "telegram_success_message",
+                "km_writeback",
+                "playbook_trust_writeback",
+            ],
+            "next_steps": [
+                "owner_review_release",
+                "fill_maintenance_window_and_blast_radius",
+                "approve_post_apply_verifier",
+                "assign_km_and_playbook_trust_writeback_owner",
+                "rerun_repair_candidate_gate_after_owner_release",
+            ],
+        }
+
+    def _promotion_contract_field(
+        self,
+        *,
+        field: str,
+        label: str,
+        status: str,
+        source: str,
+        value: Any,
+    ) -> dict[str, Any]:
+        return {
+            "field": field,
+            "label": label,
+            "status": status,
+            "source": source,
+            "value": value,
+            "runtime_execution_authorized": False,
+        }
+
+    def _promotion_summary_for_operator(self, contract: dict[str, Any]) -> str:
+        route = str(contract.get("route_id") or "--")
+        ready = int(contract.get("ready_count") or 0)
+        total = int(contract.get("total_count") or 0)
+        blocked = int(contract.get("blocked_count") or 0)
+        status = str(contract.get("status") or "unknown")
+        return (
+            f"route={route}; promotion={ready}/{total}; "
+            f"blocked={blocked}; status={status}; runtime=false"
+        )
+
     def _draft_summary_for_operator(self, template: dict[str, Any]) -> str:
         route = str(template.get("suggested_route") or "--")
         repair = str(template.get("repair_command_template") or "--")

apps/api/src/services/telegram_gateway.py

@@ -1047,6 +1047,7 @@ def _format_manual_handoff_package_lines(
     repair_candidate_blocker_summary: str = "",
     repair_candidate_next_step: str = "",
     repair_candidate_required_fields: list[str] | None = None,
+    repair_candidate_promotion_summary: str = "",
     repair_candidate_work_item_href: str = "",
     repair_candidate_work_item_id: str = "",
     compact: bool = False,
@@ -1111,6 +1112,12 @@ def _format_manual_handoff_package_lines(
             f"├ 下一步：{html.escape(str(repair_candidate_next_step)[:360])}",
         )
         insert_at += 1
+    if repair_candidate_promotion_summary:
+        lines.insert(
+            insert_at,
+            f"├ 候選升級合約：<code>{html.escape(str(repair_candidate_promotion_summary)[:260])}</code>",
+        )
+        insert_at += 1
     if required_fields:
         field_text = ", ".join(required_fields[:7])
         lines.insert(
@@ -3015,6 +3022,7 @@ class TelegramMessage:
     repair_candidate_blocker_summary: str = ""  # 修復候選阻擋原因摘要
     repair_candidate_next_step: str = ""  # 修復候選阻擋後的下一步
     repair_candidate_required_fields: list[str] | None = None  # PlayBook 草案必填欄位
+    repair_candidate_promotion_summary: str = ""  # 候選升級合約摘要
     repair_candidate_work_item_href: str = ""  # AwoooP 修復候選草案連結
     repair_candidate_work_item_id: str = ""  # AwoooP 修復候選草案 ID
 
@@ -3203,6 +3211,7 @@ class TelegramMessage:
             repair_candidate_blocker_summary=self.repair_candidate_blocker_summary,
             repair_candidate_next_step=self.repair_candidate_next_step,
             repair_candidate_required_fields=self.repair_candidate_required_fields,
+            repair_candidate_promotion_summary=self.repair_candidate_promotion_summary,
             repair_candidate_work_item_href=self.repair_candidate_work_item_href,
             repair_candidate_work_item_id=self.repair_candidate_work_item_id,
         )
@@ -5317,6 +5326,7 @@ class TelegramGateway:
         repair_candidate_blocker_summary: str = "",
         repair_candidate_next_step: str = "",
         repair_candidate_required_fields: list[str] | None = None,
+        repair_candidate_promotion_summary: str = "",
         repair_candidate_work_item_href: str = "",
         repair_candidate_work_item_id: str = "",
     ) -> dict:
@@ -5428,6 +5438,7 @@ class TelegramGateway:
             repair_candidate_blocker_summary=repair_candidate_blocker_summary,
             repair_candidate_next_step=repair_candidate_next_step,
             repair_candidate_required_fields=repair_candidate_required_fields,
+            repair_candidate_promotion_summary=repair_candidate_promotion_summary,
             repair_candidate_work_item_href=repair_candidate_work_item_href,
             repair_candidate_work_item_id=repair_candidate_work_item_id,
         )

apps/api/tests/test_repair_candidate_service.py

@@ -347,11 +347,28 @@ async def test_candidate_blocked_observe_only_prompts_repair_playbook_draft() ->
     assert draft_template["repair_command_template"] == "systemctl restart node-exporter-188"
     assert draft_template["template_is_executable"] is False
     assert draft_template["runtime_execution_authorized"] is False
+    promotion_contract = draft_package["candidate_promotion_contract"]
+    assert promotion_contract["schema_version"] == "repair_candidate_promotion_contract_v1"
+    assert promotion_contract["status"] == "owner_review_ready_runtime_blocked"
+    assert promotion_contract["route_id"] == "host_service_route_after_owner_review"
+    assert promotion_contract["repair_command_template"] == "systemctl restart node-exporter-188"
+    assert promotion_contract["ready_count"] == 6
+    assert promotion_contract["total_count"] == 11
+    assert promotion_contract["blocked_count"] == 5
+    assert "owner_review" in promotion_contract["blocked_fields"]
+    assert "maintenance_window" in promotion_contract["blocked_fields"]
+    assert promotion_contract["runtime_execution_authorized"] is False
+    assert promotion_contract["runtime_write_allowed"] is False
+    assert result.metadata["repair_candidate_promotion_contract"] == promotion_contract
+    assert "promotion=6/11" in result.metadata["repair_candidate_promotion_summary"]
     work_item = draft_package["awooop_work_item"]
     assert work_item["status"] == "owner_review_ready"
     assert work_item["next_action"] == "owner_review_repair_candidate_draft"
     assert work_item["owner_review_required"] is True
     assert work_item["runtime_execution_authorized"] is False
+    assert work_item["candidate_promotion_contract"]["route_id"] == (
+        "host_service_route_after_owner_review"
+    )
     assert work_item["target_resource"] == "node-exporter-188"
     assert work_item["lane"] == "promote_diagnostic_to_repair_playbook"
     assert work_item["safety_level"] == "read_only_work_item_projection"

apps/api/tests/test_telegram_ai_automation_block.py

@@ -123,6 +123,10 @@ def test_repair_candidate_draft_ready_card_exposes_owner_review_handoff() -> Non
             "verifier_plan",
             "owner_review",
         ],
+        repair_candidate_promotion_summary=(
+            "route=host_service_route_after_owner_review; promotion=6/11; "
+            "blocked=5; status=owner_review_ready_runtime_blocked; runtime=false"
+        ),
         repair_candidate_work_item_href=(
             "https://awoooi.wooo.work/zh-TW/awooop/work-items?"
             "project_id=awoooi&incident_id=INC-20260625-977E5F"
@@ -140,6 +144,9 @@ def test_repair_candidate_draft_ready_card_exposes_owner_review_handoff() -> Non
     assert "Owner review 修復候選草案" in body
     assert "PlayBook 只有觀察或診斷步驟" in body
     assert "把診斷命令保留為 MCP evidence collector" in body
+    assert "候選升級合約" in body
+    assert "route=host_service_route_after_owner_review" in body
+    assert "promotion=6/11" in body
     assert "AwoooP 修復候選草案" in body
     assert "自動化資產總帳" in body
     assert "按鈕：<b>Work Item</b> 開啟 owner review" in body

apps/api/tests/test_telegram_webhook_execution_handoff.py

@@ -239,6 +239,16 @@ async def test_telegram_approval_exposes_owner_review_handoff_for_draft_ready(mo
             "repair_candidate_draft_package": {
                 "status": "owner_review_ready",
                 "next_step": "owner_review_repair_candidate_draft",
+                "candidate_promotion_contract": {
+                    "schema_version": "repair_candidate_promotion_contract_v1",
+                    "status": "owner_review_ready_runtime_blocked",
+                    "route_id": "host_service_route_after_owner_review",
+                    "ready_count": 6,
+                    "total_count": 11,
+                    "blocked_count": 5,
+                    "runtime_execution_authorized": False,
+                    "runtime_write_allowed": False,
+                },
                 "required_fields": [
                     "alertname",
                     "target_selector",
@@ -309,6 +319,14 @@ async def test_telegram_approval_exposes_owner_review_handoff_for_draft_ready(mo
     assert result["repair_candidate_draft_ready"] is True
     assert result["owner_review_required"] is True
     assert result["next_action"] == "owner_review_repair_candidate_draft"
+    assert result["repair_candidate_promotion_summary"] == (
+        "route=host_service_route_after_owner_review; promotion=6/11; "
+        "blocked=5; runtime=false"
+    )
+    assert result["repair_candidate_promotion_contract"]["status"] == (
+        "owner_review_ready_runtime_blocked"
+    )
+    assert result["repair_candidate_promotion_contract"]["runtime_execution_authorized"] is False
     assert result["work_item_id"] == (
         "repair-candidate-draft:awoooi:INC-20260625-977E5F:"
         "promote_diagnostic_to_repair_playbook"