feat(awooop): surface cicd rollout evidence

apps/api/src/services/platform_operator_service.py

@@ -85,6 +85,8 @@ _CALLBACK_REPLY_RAW_STATUS_BY_FILTER = {
     "failed": "callback_reply_failed",
 }
 _CALLBACK_REPLY_ACTION_RE = re.compile(r"^[a-z0-9_:-]{1,64}$", re.IGNORECASE)
+_CICD_STATUS_FILTERS = {"running", "success", "failed", "pending"}
+_CICD_STAGE_RE = re.compile(r"^[a-z0-9_:-]{1,64}$", re.IGNORECASE)
 _AI_ROUTE_STATUS_SCHEMA_VERSION = "awooop_ai_route_status_v1"
 _AI_ROUTE_WORKLOADS = set(get_args(OllamaWorkloadType))
 _SOURCE_CORRELATION_SCHEMA_VERSION = "source_provider_correlation_v1"
@@ -414,6 +416,84 @@ async def list_callback_replies(
     }
 
 
+async def list_cicd_events(
+    *,
+    project_id: str | None,
+    stage: str | None,
+    status_filter: str | None,
+    limit: int,
+) -> dict[str, Any]:
+    """列出 CI/CD notification evidence，來源是 alert_operation_log。"""
+    safe_limit = max(1, min(limit, 50))
+    normalized_stage = _validate_cicd_stage_filter(stage)
+    normalized_status = _validate_cicd_status_filter(status_filter)
+
+    # alert_operation_log 目前是 legacy/global evidence table，CI/CD notification
+    # 只屬於 AWOOOI production；非 awoooi project filter 回空集合，避免誤導多租戶 UI。
+    if project_id and project_id != "awoooi":
+        return {"items": [], "total": 0, "limit": safe_limit}
+
+    where_clauses = [
+        "event_type = 'ALERT_RECEIVED'",
+        "actor = 'alertmanager'",
+        """
+        COALESCE(
+            context #>> '{labels,alertname}',
+            context ->> 'alertname',
+            ''
+        ) LIKE 'CI_%'
+        """,
+    ]
+    params: dict[str, Any] = {"limit": safe_limit}
+    if normalized_stage:
+        where_clauses.append(
+            "LOWER(COALESCE(context #>> '{labels,stage}', '')) = :stage"
+        )
+        params["stage"] = normalized_stage
+    if normalized_status:
+        where_clauses.append(
+            "LOWER(COALESCE(context #>> '{labels,status}', '')) = :status"
+        )
+        params["status"] = normalized_status
+
+    where_sql = " AND ".join(where_clauses)
+    sql = text(f"""
+        SELECT
+            id,
+            action_detail,
+            success,
+            created_at,
+            context,
+            COALESCE(
+                context #>> '{{labels,alertname}}',
+                context ->> 'alertname',
+                ''
+            ) AS alertname,
+            context #>> '{{labels,stage}}' AS stage,
+            context #>> '{{labels,status}}' AS status,
+            context #>> '{{labels,severity}}' AS severity,
+            context #>> '{{labels,commit}}' AS commit_sha,
+            context #>> '{{labels,triggered_by}}' AS triggered_by,
+            context #>> '{{labels,duration_seconds}}' AS duration_seconds,
+            context #>> '{{annotations,summary}}' AS summary,
+            context #>> '{{annotations,description}}' AS description,
+            context #>> '{{annotations,workflow_url}}' AS workflow_url,
+            context ->> 'alert_id' AS alert_id,
+            context ->> 'source' AS source
+        FROM alert_operation_log
+        WHERE {where_sql}
+        ORDER BY created_at DESC, id DESC
+        LIMIT :limit
+    """)
+
+    async with get_db_context("awoooi") as db:
+        result = await db.execute(sql, params)
+        rows = list(result.mappings().all())
+
+    items = [_cicd_event_item_from_row(row, project_id=project_id or "awoooi") for row in rows]
+    return {"items": items, "total": len(items), "limit": safe_limit}
+
+
 async def get_ai_route_status(
     workload_type: str | None = None,
 ) -> dict[str, Any]:
@@ -766,6 +846,89 @@ def _outbound_timeline_metadata(
     return metadata
 
 
+def _validate_cicd_stage_filter(value: str | None) -> str | None:
+    """Normalize a CI/CD stage filter without allowing arbitrary SQL fragments."""
+    if value is None:
+        return None
+    stage = value.strip().lower()
+    if not stage:
+        return None
+    if not _CICD_STAGE_RE.fullmatch(stage):
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
+            detail="stage 格式錯誤，僅允許 a-z、0-9、底線、冒號與短橫線",
+        )
+    return stage
+
+
+def _validate_cicd_status_filter(value: str | None) -> str | None:
+    """Normalize and validate CI/CD status filter."""
+    if value is None:
+        return None
+    status_value = value.strip().lower()
+    if not status_value:
+        return None
+    if status_value not in _CICD_STATUS_FILTERS:
+        allowed = ", ".join(sorted(_CICD_STATUS_FILTERS))
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
+            detail=f"status 必須是: {allowed}",
+        )
+    return status_value
+
+
+def _cicd_duration_seconds(value: Any) -> int:
+    """Coerce Alertmanager duration_seconds label into a non-negative integer."""
+    try:
+        duration = int(str(value or "0"))
+    except (TypeError, ValueError):
+        return 0
+    return max(duration, 0)
+
+
+def _cicd_event_needs_attention(status_value: str | None, severity: str | None) -> bool:
+    """Return whether a CI/CD evidence row should be highlighted for operators."""
+    normalized_status = str(status_value or "").lower()
+    normalized_severity = str(severity or "").lower()
+    return normalized_status in {"failed", "pending"} or normalized_severity in {
+        "critical",
+        "warning",
+    }
+
+
+def _cicd_event_item_from_row(row: Mapping[str, Any], *, project_id: str) -> dict[str, Any]:
+    """Convert one alert_operation_log CI/CD row into an operator-facing item."""
+    context = _as_dict(row.get("context"))
+    labels = _as_dict(context.get("labels"))
+    annotations = _as_dict(context.get("annotations"))
+    status_value = str(row.get("status") or labels.get("status") or "").lower() or None
+    severity = str(row.get("severity") or labels.get("severity") or "").lower() or None
+    summary = row.get("summary") or annotations.get("summary")
+    description = row.get("description") or annotations.get("description")
+    workflow_url = row.get("workflow_url") or annotations.get("workflow_url")
+    return {
+        "id": str(row.get("id") or ""),
+        "project_id": project_id,
+        "alertname": str(row.get("alertname") or labels.get("alertname") or ""),
+        "stage": row.get("stage") or labels.get("stage"),
+        "status": status_value,
+        "severity": severity,
+        "commit_sha": row.get("commit_sha") or labels.get("commit"),
+        "triggered_by": row.get("triggered_by") or labels.get("triggered_by"),
+        "duration_seconds": _cicd_duration_seconds(
+            row.get("duration_seconds") or labels.get("duration_seconds")
+        ),
+        "summary": str(summary).strip() if summary else None,
+        "description": str(description).strip() if description else None,
+        "workflow_url": str(workflow_url).strip() if workflow_url else None,
+        "alert_id": row.get("alert_id") or context.get("alert_id"),
+        "source": row.get("source") or context.get("source"),
+        "action_detail": row.get("action_detail"),
+        "needs_attention": _cicd_event_needs_attention(status_value, severity),
+        "created_at": row.get("created_at"),
+    }
+
+
 def _run_callback_reply_summary(
     outbound_messages: list[AwoooPOutboundMessage],
 ) -> dict[str, Any]: