diff --git a/apps/api/src/api/v1/telegram.py b/apps/api/src/api/v1/telegram.py
index 4d84573e..00e87bda 100644
--- a/apps/api/src/api/v1/telegram.py
+++ b/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")),
diff --git a/apps/api/src/api/v1/webhooks.py b/apps/api/src/api/v1/webhooks.py
index 67fb623e..b8d4e55e 100644
--- a/apps/api/src/api/v1/webhooks.py
+++ b/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", {})
diff --git a/apps/api/src/services/repair_candidate_service.py b/apps/api/src/services/repair_candidate_service.py
index 770b7305..049dbdc4 100644
--- a/apps/api/src/services/repair_candidate_service.py
+++ b/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 "--")
diff --git a/apps/api/src/services/telegram_gateway.py b/apps/api/src/services/telegram_gateway.py
index 743df2ce..c0aa3197 100644
--- a/apps/api/src/services/telegram_gateway.py
+++ b/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"├ 候選升級合約：{html.escape(str(repair_candidate_promotion_summary)[:260])}",
+ )
+ 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,
)
diff --git a/apps/api/tests/test_repair_candidate_service.py b/apps/api/tests/test_repair_candidate_service.py
index c3d10e8c..b7cab6ff 100644
--- a/apps/api/tests/test_repair_candidate_service.py
+++ b/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"
diff --git a/apps/api/tests/test_telegram_ai_automation_block.py b/apps/api/tests/test_telegram_ai_automation_block.py
index a06aa545..98215c95 100644
--- a/apps/api/tests/test_telegram_ai_automation_block.py
+++ b/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 "按鈕：Work Item 開啟 owner review" in body
diff --git a/apps/api/tests/test_telegram_webhook_execution_handoff.py b/apps/api/tests/test_telegram_webhook_execution_handoff.py
index 22b1f168..21f99aec 100644
--- a/apps/api/tests/test_telegram_webhook_execution_handoff.py
+++ b/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"