1613 lines
64 KiB
Python
1613 lines
64 KiB
Python
"""
|
||
Agent Teams API - Phase 9.5 多專家協作系統
|
||
==========================================
|
||
|
||
Endpoints:
|
||
- POST /api/v1/agents/analyze - 觸發 Agent Teams 分析
|
||
- GET /api/v1/agents/status/{task_id} - 查詢分析狀態
|
||
- GET /api/v1/agents/result/{task_id} - 取得分析結果
|
||
- GET /api/v1/agents/stream/{task_id} - SSE 串流進度
|
||
|
||
Phase 9.4-9.5 核心功能:
|
||
1. ConsensusEngine 整合多專家意見
|
||
2. BackgroundTasks 執行長時間分析
|
||
3. Redis Working Memory 儲存結果
|
||
4. SSE 推送即時進度
|
||
|
||
Phase 17 技術債修復 (2026-03-26):
|
||
- Router 層不再直接存取 Redis
|
||
- 所有業務邏輯封裝至 AgentService
|
||
- 符合 leWOOOgo 積木化原則
|
||
|
||
統帥鐵律:
|
||
- 所有分析任務必須可追蹤 (task_id)
|
||
- 超過 60 秒的分析必須用 BackgroundTasks
|
||
- 結果必須存入 Redis (7 天 TTL)
|
||
"""
|
||
|
||
import asyncio
|
||
import json
|
||
from typing import Any
|
||
|
||
from fastapi import APIRouter, BackgroundTasks, HTTPException, status
|
||
from fastapi.responses import StreamingResponse
|
||
from pydantic import BaseModel, Field
|
||
|
||
from src.core.logging import get_logger
|
||
from src.core.sse import get_publisher
|
||
from src.services.agent_market_governance_snapshot import (
|
||
load_latest_agent_market_governance_snapshot,
|
||
)
|
||
from src.services.agent_service import (
|
||
AgentService,
|
||
TaskState,
|
||
get_agent_service,
|
||
)
|
||
from src.services.ai_agent_automation_backlog_snapshot import (
|
||
load_latest_ai_agent_automation_backlog_snapshot,
|
||
)
|
||
from src.services.ai_agent_automation_inventory_snapshot import (
|
||
load_latest_ai_agent_automation_inventory_snapshot,
|
||
)
|
||
from src.services.ai_agent_communication_learning_contract import (
|
||
load_latest_ai_agent_communication_learning_contract,
|
||
)
|
||
from src.services.ai_agent_deployment_layout import (
|
||
load_latest_ai_agent_deployment_layout,
|
||
)
|
||
from src.services.ai_agent_gitea_pr_draft_lane import (
|
||
load_latest_ai_agent_gitea_pr_draft_lane,
|
||
)
|
||
from src.services.ai_agent_host_stateful_version_inventory import (
|
||
load_latest_ai_agent_host_stateful_version_inventory,
|
||
)
|
||
from src.services.ai_agent_interaction_learning_proof import (
|
||
load_latest_ai_agent_interaction_learning_proof,
|
||
)
|
||
from src.services.ai_agent_learning_writeback_approval_package import (
|
||
load_latest_ai_agent_learning_writeback_approval_package,
|
||
)
|
||
from src.services.ai_agent_live_read_model_gate import (
|
||
load_latest_ai_agent_live_read_model_gate,
|
||
)
|
||
from src.services.ai_agent_owner_approved_fixture_dry_run import (
|
||
load_latest_ai_agent_owner_approved_fixture_dry_run,
|
||
)
|
||
from src.services.ai_agent_owner_approved_learning_dry_run import (
|
||
load_latest_ai_agent_owner_approved_learning_dry_run,
|
||
)
|
||
from src.services.ai_agent_post_write_verifier_package import (
|
||
load_latest_ai_agent_post_write_verifier_package,
|
||
)
|
||
from src.services.ai_agent_proactive_operations_contract import (
|
||
load_latest_ai_agent_proactive_operations_contract,
|
||
)
|
||
from src.services.ai_agent_redis_dry_run_gate import (
|
||
load_latest_ai_agent_redis_dry_run_gate,
|
||
)
|
||
from src.services.ai_agent_report_automation_review import (
|
||
load_latest_ai_agent_report_automation_review,
|
||
)
|
||
from src.services.ai_agent_report_truth_actionability_review import (
|
||
load_latest_ai_agent_report_truth_actionability_review,
|
||
)
|
||
from src.services.ai_agent_runtime_write_gate_review import (
|
||
load_latest_ai_agent_runtime_write_gate_review,
|
||
)
|
||
from src.services.ai_agent_runtime_verifier_evidence_review import (
|
||
load_latest_ai_agent_runtime_verifier_evidence_review,
|
||
)
|
||
from src.services.ai_agent_telegram_action_required_digest_policy import (
|
||
load_latest_ai_agent_telegram_action_required_digest_policy,
|
||
)
|
||
from src.services.ai_agent_telegram_receipt_approval_package import (
|
||
load_latest_ai_agent_telegram_receipt_approval_package,
|
||
)
|
||
from src.services.ai_agent_tool_adoption_approval_package import (
|
||
load_latest_ai_agent_tool_adoption_approval_package,
|
||
)
|
||
from src.services.ai_agent_version_freshness_snapshot import (
|
||
load_latest_ai_agent_version_freshness_snapshot,
|
||
)
|
||
from src.services.ai_provider_route_matrix import (
|
||
load_latest_ai_provider_route_matrix,
|
||
)
|
||
from src.services.backup_dr_readiness_matrix import (
|
||
load_latest_backup_dr_readiness_matrix,
|
||
)
|
||
from src.services.backup_dr_target_inventory import (
|
||
load_latest_backup_dr_target_inventory,
|
||
)
|
||
from src.services.backup_notification_policy import (
|
||
load_latest_backup_notification_policy,
|
||
)
|
||
from src.services.backup_restore_drill_approval_package_template import (
|
||
load_latest_backup_restore_drill_approval_package_template,
|
||
)
|
||
from src.services.dependency_drift_check_plan import (
|
||
load_latest_dependency_drift_check_plan,
|
||
)
|
||
from src.services.dependency_risk_policy import (
|
||
load_latest_dependency_risk_policy,
|
||
)
|
||
from src.services.dependency_upgrade_approval_package_template import (
|
||
load_latest_dependency_upgrade_approval_package_template,
|
||
)
|
||
from src.services.docker_build_surface_inventory import (
|
||
load_latest_docker_build_surface_inventory,
|
||
)
|
||
from src.services.gitea_workflow_runner_health import (
|
||
load_latest_gitea_workflow_runner_health,
|
||
)
|
||
from src.services.javascript_package_inventory import (
|
||
load_latest_javascript_package_inventory,
|
||
)
|
||
from src.services.observability_contract_matrix import (
|
||
load_latest_observability_contract_matrix,
|
||
)
|
||
from src.services.offsite_escrow_readiness_status import (
|
||
load_latest_offsite_escrow_readiness_status,
|
||
)
|
||
from src.services.package_supply_chain_inventory import (
|
||
load_latest_package_supply_chain_inventory,
|
||
)
|
||
from src.services.runtime_surface_inventory import (
|
||
load_latest_runtime_surface_inventory,
|
||
)
|
||
from src.services.service_health_failure_notification_policy import (
|
||
load_latest_service_health_failure_notification_policy,
|
||
)
|
||
from src.services.service_health_gap_matrix import (
|
||
load_latest_service_health_gap_matrix,
|
||
)
|
||
|
||
router = APIRouter(prefix="/agents", tags=["Agent Teams"])
|
||
logger = get_logger("awoooi.agents")
|
||
|
||
|
||
# =============================================================================
|
||
# Request/Response Models
|
||
# =============================================================================
|
||
|
||
class AnalyzeRequest(BaseModel):
|
||
"""分析請求"""
|
||
incident_id: str | None = Field(
|
||
None,
|
||
description="現有 Incident ID (二選一)"
|
||
)
|
||
# 或直接提供 Incident 資訊
|
||
severity: str | None = Field(
|
||
None,
|
||
description="事件嚴重度 (P0/P1/P2/P3)"
|
||
)
|
||
affected_services: list[str] | None = Field(
|
||
None,
|
||
description="受影響服務列表"
|
||
)
|
||
alert_names: list[str] | None = Field(
|
||
None,
|
||
description="告警名稱列表"
|
||
)
|
||
context: dict[str, Any] | None = Field(
|
||
None,
|
||
description="額外上下文"
|
||
)
|
||
|
||
|
||
class AnalyzeResponse(BaseModel):
|
||
"""分析回應"""
|
||
task_id: str
|
||
status: str
|
||
message: str
|
||
estimated_seconds: int = 30
|
||
|
||
|
||
class TaskStatusResponse(BaseModel):
|
||
"""任務狀態回應"""
|
||
task_id: str
|
||
state: str
|
||
progress: int # 0-100
|
||
current_step: str | None = None
|
||
agents_completed: int = 0
|
||
total_agents: int = 4
|
||
started_at: str | None = None
|
||
completed_at: str | None = None
|
||
error: str | None = None
|
||
|
||
|
||
class TaskResultResponse(BaseModel):
|
||
"""任務結果回應"""
|
||
task_id: str
|
||
state: str
|
||
consensus_id: str | None = None
|
||
incident_id: str | None = None
|
||
consensus_score: float | None = None
|
||
recommended_action: str | None = None
|
||
recommended_kubectl: str | None = None
|
||
risk_level: str | None = None
|
||
final_reasoning: str | None = None
|
||
opinions: list[dict[str, Any]] | None = None
|
||
dissenting_opinions: list[str] | None = None
|
||
created_at: str | None = None
|
||
|
||
|
||
# =============================================================================
|
||
# Background Task Wrapper
|
||
# =============================================================================
|
||
|
||
async def _run_analysis_task(
|
||
service: AgentService,
|
||
task_id: str,
|
||
incident_id: str,
|
||
) -> None:
|
||
"""
|
||
背景任務包裝器
|
||
|
||
從 AgentService 取得 Incident 並執行分析
|
||
"""
|
||
incident = await service.get_incident(incident_id)
|
||
if incident is None:
|
||
logger.error("background_task_incident_not_found", incident_id=incident_id)
|
||
return
|
||
|
||
await service.run_analysis(task_id, incident)
|
||
|
||
|
||
# =============================================================================
|
||
# API Endpoints
|
||
# =============================================================================
|
||
|
||
@router.post(
|
||
"/analyze",
|
||
response_model=AnalyzeResponse,
|
||
summary="觸發 Agent Teams 分析",
|
||
description="""
|
||
觸發多專家協作分析。
|
||
|
||
可提供:
|
||
- 現有 Incident ID (從 Redis 讀取)
|
||
- 或直接提供事件資訊 (severity, affected_services, alert_names)
|
||
|
||
分析在背景執行，使用 task_id 追蹤進度。
|
||
|
||
專家團隊:
|
||
- SRE Agent: 系統穩定性分析
|
||
- Security Agent: 資安風險評估
|
||
- Cost Agent: 成本效益分析
|
||
- Performance Agent: 效能優化建議
|
||
""",
|
||
)
|
||
async def analyze(
|
||
request: AnalyzeRequest,
|
||
background_tasks: BackgroundTasks,
|
||
) -> AnalyzeResponse:
|
||
"""
|
||
觸發 Agent Teams 分析
|
||
|
||
返回 task_id 用於追蹤進度
|
||
"""
|
||
service = get_agent_service()
|
||
|
||
# 取得或建立 Incident
|
||
if request.incident_id:
|
||
# 從 Redis 讀取現有 Incident
|
||
incident = await service.get_incident(request.incident_id)
|
||
|
||
if incident is None:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail=f"Incident not found: {request.incident_id}",
|
||
)
|
||
|
||
elif request.severity and request.affected_services:
|
||
# 建立臨時 Incident
|
||
incident = service.create_adhoc_incident(
|
||
severity=request.severity,
|
||
affected_services=request.affected_services,
|
||
alert_names=request.alert_names,
|
||
)
|
||
|
||
else:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_400_BAD_REQUEST,
|
||
detail="Must provide either incident_id or (severity + affected_services)",
|
||
)
|
||
|
||
# 建立任務
|
||
task_id = await service.create_analysis_task(incident, trigger="manual")
|
||
|
||
# 加入背景任務
|
||
background_tasks.add_task(
|
||
service.run_analysis,
|
||
task_id,
|
||
incident,
|
||
)
|
||
|
||
logger.info(
|
||
"agent_analysis_started",
|
||
task_id=task_id,
|
||
incident_id=incident.incident_id,
|
||
severity=incident.severity.value,
|
||
)
|
||
|
||
return AnalyzeResponse(
|
||
task_id=task_id,
|
||
status="pending",
|
||
message="Agent Teams 分析已啟動",
|
||
estimated_seconds=30,
|
||
)
|
||
|
||
|
||
@router.get(
|
||
"/status/{task_id}",
|
||
response_model=TaskStatusResponse,
|
||
summary="查詢分析狀態",
|
||
description="查詢 Agent Teams 分析任務的目前狀態與進度。",
|
||
)
|
||
async def get_status(task_id: str) -> TaskStatusResponse:
|
||
"""
|
||
查詢任務狀態
|
||
|
||
返回進度百分比與目前步驟
|
||
"""
|
||
service = get_agent_service()
|
||
task_data = await service.get_task_status(task_id)
|
||
|
||
if task_data is None:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail=f"Task not found: {task_id}",
|
||
)
|
||
|
||
return TaskStatusResponse(
|
||
task_id=task_id,
|
||
state=task_data.get("state", "unknown"),
|
||
progress=task_data.get("progress", 0),
|
||
current_step=task_data.get("current_step"),
|
||
agents_completed=task_data.get("agents_completed", 0),
|
||
total_agents=task_data.get("total_agents", 4),
|
||
started_at=task_data.get("started_at"),
|
||
completed_at=task_data.get("completed_at"),
|
||
error=task_data.get("error"),
|
||
)
|
||
|
||
|
||
@router.get(
|
||
"/result/{task_id}",
|
||
response_model=TaskResultResponse,
|
||
summary="取得分析結果",
|
||
description="取得 Agent Teams 分析的完整結果，包含所有專家意見與共識決策。",
|
||
)
|
||
async def get_result(task_id: str) -> TaskResultResponse:
|
||
"""
|
||
取得分析結果
|
||
|
||
只有 COMPLETED 狀態才有完整結果
|
||
"""
|
||
service = get_agent_service()
|
||
task_data = await service.get_task_result(task_id)
|
||
|
||
if task_data is None:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail=f"Task not found: {task_id}",
|
||
)
|
||
|
||
return TaskResultResponse(
|
||
task_id=task_id,
|
||
state=task_data.get("state", "unknown"),
|
||
consensus_id=task_data.get("consensus_id"),
|
||
incident_id=task_data.get("incident_id"),
|
||
consensus_score=task_data.get("consensus_score"),
|
||
recommended_action=task_data.get("recommended_action"),
|
||
recommended_kubectl=task_data.get("recommended_kubectl"),
|
||
risk_level=task_data.get("risk_level"),
|
||
final_reasoning=task_data.get("final_reasoning"),
|
||
opinions=task_data.get("opinions"),
|
||
dissenting_opinions=task_data.get("dissenting_opinions"),
|
||
created_at=task_data.get("completed_at"),
|
||
)
|
||
|
||
|
||
@router.get(
|
||
"/stream/{task_id}",
|
||
summary="SSE 串流進度",
|
||
description="透過 Server-Sent Events 即時接收分析進度更新。",
|
||
)
|
||
async def stream_progress(task_id: str) -> StreamingResponse:
|
||
"""
|
||
SSE 串流分析進度
|
||
|
||
客戶端可訂閱此端點接收即時更新
|
||
"""
|
||
service = get_agent_service()
|
||
|
||
# 驗證任務存在
|
||
task_data = await service.get_task_status(task_id)
|
||
if task_data is None:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail=f"Task not found: {task_id}",
|
||
)
|
||
|
||
async def generate():
|
||
"""SSE 串流生成器"""
|
||
publisher = await get_publisher()
|
||
client = await publisher.subscribe(
|
||
topics=[f"agent_task:{task_id}"],
|
||
metadata={"task_id": task_id},
|
||
)
|
||
|
||
try:
|
||
# 發送初始狀態
|
||
current_data = await service.get_task_status(task_id)
|
||
if current_data:
|
||
payload = json.dumps(
|
||
{"type": "status", **current_data},
|
||
ensure_ascii=False,
|
||
)
|
||
yield f"data: {payload}\n\n"
|
||
|
||
# 串流後續更新
|
||
async for event_str in publisher.stream(client):
|
||
yield event_str
|
||
|
||
# 檢查是否完成或失敗
|
||
current_data = await service.get_task_status(task_id)
|
||
if current_data:
|
||
final_states = [TaskState.COMPLETED.value, TaskState.FAILED.value]
|
||
if current_data.get("state") in final_states:
|
||
break
|
||
|
||
except asyncio.CancelledError:
|
||
logger.info("agent_stream_cancelled", task_id=task_id)
|
||
raise
|
||
finally:
|
||
await publisher.unsubscribe(client.id)
|
||
|
||
return StreamingResponse(
|
||
generate(),
|
||
media_type="text/event-stream",
|
||
headers={
|
||
"Cache-Control": "no-cache",
|
||
"Connection": "keep-alive",
|
||
"X-Accel-Buffering": "no",
|
||
},
|
||
)
|
||
|
||
|
||
@router.get(
|
||
"/market-governance-snapshot",
|
||
response_model=dict[str, Any],
|
||
summary="取得 AI Agent 市場治理快照",
|
||
description=(
|
||
"讀取最新已提交的 Agent market governance snapshot；"
|
||
"此 endpoint 不呼叫外部來源、不批准 SDK/API/replay/shadow/canary/production change。"
|
||
),
|
||
)
|
||
async def get_market_governance_snapshot() -> dict[str, Any]:
|
||
"""Return the latest read-only Agent market governance snapshot."""
|
||
try:
|
||
return await asyncio.to_thread(load_latest_agent_market_governance_snapshot)
|
||
except FileNotFoundError as exc:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail=str(exc),
|
||
) from exc
|
||
except (json.JSONDecodeError, ValueError) as exc:
|
||
logger.error("agent_market_governance_snapshot_invalid", error=str(exc))
|
||
raise HTTPException(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail="Agent market governance snapshot is invalid",
|
||
) from exc
|
||
|
||
|
||
@router.get(
|
||
"/automation-inventory-snapshot",
|
||
response_model=dict[str, Any],
|
||
summary="取得 AI Agent 自動化盤點快照",
|
||
description=(
|
||
"讀取最新已提交的 AI Agent 自動化盤點快照；"
|
||
"此端點不呼叫外部來源、不碰 DB/Redis、不批准 SDK/API/shadow/canary/生產變更。"
|
||
),
|
||
)
|
||
async def get_automation_inventory_snapshot() -> dict[str, Any]:
|
||
"""Return the latest read-only AI Agent automation inventory snapshot."""
|
||
try:
|
||
return await asyncio.to_thread(load_latest_ai_agent_automation_inventory_snapshot)
|
||
except FileNotFoundError as exc:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail=str(exc),
|
||
) from exc
|
||
except (json.JSONDecodeError, ValueError) as exc:
|
||
logger.error("ai_agent_automation_inventory_snapshot_invalid", error=str(exc))
|
||
raise HTTPException(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail="AI Agent automation inventory snapshot is invalid",
|
||
) from exc
|
||
|
||
|
||
@router.get(
|
||
"/automation-backlog-snapshot",
|
||
response_model=dict[str, Any],
|
||
summary="取得 AI Agent 自動化待辦快照",
|
||
description=(
|
||
"讀取最新已提交的 AI Agent 自動化待辦快照；"
|
||
"此端點不呼叫外部來源、不碰 DB/Redis、不批准 SDK/API/shadow/canary/生產變更。"
|
||
),
|
||
)
|
||
async def get_automation_backlog_snapshot() -> dict[str, Any]:
|
||
"""Return the latest read-only AI Agent automation backlog snapshot."""
|
||
try:
|
||
return await asyncio.to_thread(load_latest_ai_agent_automation_backlog_snapshot)
|
||
except FileNotFoundError as exc:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail=str(exc),
|
||
) from exc
|
||
except (json.JSONDecodeError, ValueError) as exc:
|
||
logger.error("ai_agent_automation_backlog_snapshot_invalid", error=str(exc))
|
||
raise HTTPException(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail="AI Agent automation backlog snapshot is invalid",
|
||
) from exc
|
||
|
||
|
||
@router.get(
|
||
"/agent-deployment-layout",
|
||
response_model=dict[str, Any],
|
||
summary="取得 AI Agent 佈建布局快照",
|
||
description=(
|
||
"讀取最新已提交的 OpenClaw / Hermes / NemoTron 佈建布局快照；"
|
||
"此端點不部署 Agent、不呼叫外部模型、不送 Telegram、不碰 DB/Redis、不讀 Secret payload、"
|
||
"不批准 SDK/API/shadow/canary/生產路由或主機變更。"
|
||
),
|
||
)
|
||
async def get_agent_deployment_layout() -> dict[str, Any]:
|
||
"""Return the latest read-only AI Agent deployment layout snapshot."""
|
||
try:
|
||
return await asyncio.to_thread(load_latest_ai_agent_deployment_layout)
|
||
except FileNotFoundError as exc:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail=str(exc),
|
||
) from exc
|
||
except (json.JSONDecodeError, ValueError) as exc:
|
||
logger.error("ai_agent_deployment_layout_invalid", error=str(exc))
|
||
raise HTTPException(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail="AI Agent 佈建布局快照無效",
|
||
) from exc
|
||
|
||
|
||
@router.get(
|
||
"/agent-communication-learning-contract",
|
||
response_model=dict[str, Any],
|
||
summary="取得 AI Agent 主動溝通與學習契約",
|
||
description=(
|
||
"讀取最新已提交的 OpenClaw / Hermes / NemoTron 主動溝通、學習、記錄、MCP 與 RAG 契約；"
|
||
"此端點不啟動 worker、不建立 DB migration、不送 Telegram、不安裝 SDK、不呼叫付費服務、"
|
||
"不修改生產路由或主機。"
|
||
),
|
||
)
|
||
async def get_agent_communication_learning_contract() -> dict[str, Any]:
|
||
"""Return the latest read-only AI Agent communication learning contract."""
|
||
try:
|
||
return await asyncio.to_thread(load_latest_ai_agent_communication_learning_contract)
|
||
except FileNotFoundError as exc:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail=str(exc),
|
||
) from exc
|
||
except (json.JSONDecodeError, ValueError) as exc:
|
||
logger.error("ai_agent_communication_learning_contract_invalid", error=str(exc))
|
||
raise HTTPException(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail="AI Agent 主動溝通與學習契約無效",
|
||
) from exc
|
||
|
||
|
||
@router.get(
|
||
"/agent-interaction-learning-proof",
|
||
response_model=dict[str, Any],
|
||
summary="取得 AI Agent 互動與學習證據面",
|
||
description=(
|
||
"讀取最新已提交的 OpenClaw / Hermes / NemoTron 互動、接手、學習、成長與 Telegram 收據證據面；"
|
||
"此端點不啟動 worker、不讀寫 Redis consumer group、不建立 DB migration、不送 Telegram、"
|
||
"不回傳內部協作逐字稿、提示詞、私有推理或機密值。"
|
||
),
|
||
)
|
||
async def get_agent_interaction_learning_proof() -> dict[str, Any]:
|
||
"""Return the latest read-only AI Agent interaction and learning proof surface."""
|
||
try:
|
||
return await asyncio.to_thread(load_latest_ai_agent_interaction_learning_proof)
|
||
except FileNotFoundError as exc:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail=str(exc),
|
||
) from exc
|
||
except (json.JSONDecodeError, ValueError) as exc:
|
||
logger.error("ai_agent_interaction_learning_proof_invalid", error=str(exc))
|
||
raise HTTPException(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail="AI Agent 互動與學習證據面無效",
|
||
) from exc
|
||
|
||
|
||
@router.get(
|
||
"/agent-live-read-model-gate",
|
||
response_model=dict[str, Any],
|
||
summary="取得 AI Agent live read model gate",
|
||
description=(
|
||
"讀取最新已提交的 AgentSession / Redis Streams live read model gate；"
|
||
"此端點不連 DB、不讀寫 Redis、不啟動 worker、不建立 DB migration、不送 Telegram、"
|
||
"不回傳內部協作逐字稿、Agent 原始輸出、提示詞、私有推理或機密值。"
|
||
),
|
||
)
|
||
async def get_agent_live_read_model_gate() -> dict[str, Any]:
|
||
"""Return the latest read-only AI Agent live read model gate snapshot."""
|
||
try:
|
||
return await asyncio.to_thread(load_latest_ai_agent_live_read_model_gate)
|
||
except FileNotFoundError as exc:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail=str(exc),
|
||
) from exc
|
||
except (json.JSONDecodeError, ValueError) as exc:
|
||
logger.error("ai_agent_live_read_model_gate_invalid", error=str(exc))
|
||
raise HTTPException(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail="AI Agent live read model gate 無效",
|
||
) from exc
|
||
|
||
|
||
@router.get(
|
||
"/agent-redis-dry-run-gate",
|
||
response_model=dict[str, Any],
|
||
summary="取得 AI Agent Redis dry-run gate",
|
||
description=(
|
||
"讀取最新已提交的 Redis Streams consumer group dry-run、handoff envelope、"
|
||
"ack / dead-letter / replay gate；此端點不連 Redis、不建立 consumer group、不 XADD、"
|
||
"不 XREADGROUP、不 ACK、不寫 dead-letter、不 replay、不送 Telegram、不做 learning writeback、"
|
||
"不回傳未核准內部細節。"
|
||
),
|
||
)
|
||
async def get_agent_redis_dry_run_gate() -> dict[str, Any]:
|
||
"""Return the latest read-only AI Agent Redis dry-run gate snapshot."""
|
||
try:
|
||
return await asyncio.to_thread(load_latest_ai_agent_redis_dry_run_gate)
|
||
except FileNotFoundError as exc:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail=str(exc),
|
||
) from exc
|
||
except (json.JSONDecodeError, ValueError) as exc:
|
||
logger.error("ai_agent_redis_dry_run_gate_invalid", error=str(exc))
|
||
raise HTTPException(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail="AI Agent Redis dry-run gate 無效",
|
||
) from exc
|
||
|
||
|
||
@router.get(
|
||
"/agent-learning-writeback-approval-package",
|
||
response_model=dict[str, Any],
|
||
summary="取得 AI Agent learning writeback approval package",
|
||
description=(
|
||
"讀取最新已提交的 KM / PlayBook trust / timeline learning / replay score 回寫批准包；"
|
||
"此端點不寫 KM、不更新 PlayBook trust、不寫 timeline、不寫 replay score、不送 Telegram、"
|
||
"不啟動 runtime worker、不回傳未核准內部細節。"
|
||
),
|
||
)
|
||
async def get_agent_learning_writeback_approval_package() -> dict[str, Any]:
|
||
"""Return the latest read-only AI Agent learning writeback approval package."""
|
||
try:
|
||
return await asyncio.to_thread(load_latest_ai_agent_learning_writeback_approval_package)
|
||
except FileNotFoundError as exc:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail=str(exc),
|
||
) from exc
|
||
except (json.JSONDecodeError, ValueError) as exc:
|
||
logger.error("ai_agent_learning_writeback_approval_package_invalid", error=str(exc))
|
||
raise HTTPException(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail="AI Agent learning writeback approval package 無效",
|
||
) from exc
|
||
|
||
|
||
@router.get(
|
||
"/agent-telegram-receipt-approval-package",
|
||
response_model=dict[str, Any],
|
||
summary="取得 AI Agent Telegram receipt approval package",
|
||
description=(
|
||
"讀取最新已提交的 Telegram receipt / queue / delivery / ack / failure / retry 批准包；"
|
||
"此端點不寫 Gateway queue、不呼叫 Telegram Bot API、不改 receiver route、不發送通知、"
|
||
"不啟動 runtime worker、不回傳 Telegram token、raw chat id 或未脫敏 payload。"
|
||
),
|
||
)
|
||
async def get_agent_telegram_receipt_approval_package() -> dict[str, Any]:
|
||
"""Return the latest read-only AI Agent Telegram receipt approval package."""
|
||
try:
|
||
return await asyncio.to_thread(load_latest_ai_agent_telegram_receipt_approval_package)
|
||
except FileNotFoundError as exc:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail=str(exc),
|
||
) from exc
|
||
except (json.JSONDecodeError, ValueError) as exc:
|
||
logger.error("ai_agent_telegram_receipt_approval_package_invalid", error=str(exc))
|
||
raise HTTPException(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail="AI Agent Telegram receipt approval package 無效",
|
||
) from exc
|
||
|
||
|
||
@router.get(
|
||
"/agent-owner-approved-learning-dry-run",
|
||
response_model=dict[str, Any],
|
||
summary="取得 AI Agent owner-approved learning dry-run contract",
|
||
description=(
|
||
"讀取最新已提交的 owner-approved learning writeback dry-run 契約；"
|
||
"此端點只回傳 dry-run preview、人工操作選項、驗證與 rollback 契約，"
|
||
"不寫 KM、不更新 PlayBook trust、不寫 timeline、不寫 replay score、不發 Telegram、"
|
||
"不啟動 runtime worker、不回傳未脫敏 payload。"
|
||
),
|
||
)
|
||
async def get_agent_owner_approved_learning_dry_run() -> dict[str, Any]:
|
||
"""Return the latest read-only AI Agent owner-approved learning dry-run contract."""
|
||
try:
|
||
return await asyncio.to_thread(load_latest_ai_agent_owner_approved_learning_dry_run)
|
||
except FileNotFoundError as exc:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail=str(exc),
|
||
) from exc
|
||
except (json.JSONDecodeError, ValueError) as exc:
|
||
logger.error("ai_agent_owner_approved_learning_dry_run_invalid", error=str(exc))
|
||
raise HTTPException(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail="AI Agent owner-approved learning dry-run 無效",
|
||
) from exc
|
||
|
||
|
||
@router.get(
|
||
"/agent-runtime-write-gate-review",
|
||
response_model=dict[str, Any],
|
||
summary="取得 AI Agent runtime write gate review",
|
||
description=(
|
||
"讀取最新已提交的 runtime write gate review 契約；此端點只回傳雙重批准、"
|
||
"dry-run hash、post-write verifier 與 redaction 欄位檢查，"
|
||
"不寫 KM、不更新 PlayBook trust、不寫 timeline、不寫 replay score、不發 Telegram、"
|
||
"不啟動 runtime worker、不回傳未脫敏 payload。"
|
||
),
|
||
)
|
||
async def get_agent_runtime_write_gate_review() -> dict[str, Any]:
|
||
"""Return the latest read-only AI Agent runtime write gate review."""
|
||
try:
|
||
return await asyncio.to_thread(load_latest_ai_agent_runtime_write_gate_review)
|
||
except FileNotFoundError as exc:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail=str(exc),
|
||
) from exc
|
||
except (json.JSONDecodeError, ValueError) as exc:
|
||
logger.error("ai_agent_runtime_write_gate_review_invalid", error=str(exc))
|
||
raise HTTPException(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail="AI Agent runtime write gate review 無效",
|
||
) from exc
|
||
|
||
|
||
@router.get(
|
||
"/agent-post-write-verifier-package",
|
||
response_model=dict[str, Any],
|
||
summary="取得 AI Agent post-write verifier package",
|
||
description=(
|
||
"讀取最新已提交的 post-write verifier implementation package；此端點只回傳 verifier package、"
|
||
"rollback lane、failure lane 與人工操作選項，"
|
||
"不寫 KM、不更新 PlayBook trust、不寫 timeline、不寫 replay score、不發 Telegram、"
|
||
"不啟動 runtime worker、不讀 canonical target、不回傳未脫敏 payload。"
|
||
),
|
||
)
|
||
async def get_agent_post_write_verifier_package() -> dict[str, Any]:
|
||
"""Return the latest read-only AI Agent post-write verifier package."""
|
||
try:
|
||
return await asyncio.to_thread(load_latest_ai_agent_post_write_verifier_package)
|
||
except FileNotFoundError as exc:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail=str(exc),
|
||
) from exc
|
||
except (json.JSONDecodeError, ValueError) as exc:
|
||
logger.error("ai_agent_post_write_verifier_package_invalid", error=str(exc))
|
||
raise HTTPException(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail="AI Agent post-write verifier package 無效",
|
||
) from exc
|
||
|
||
|
||
@router.get(
|
||
"/agent-runtime-verifier-evidence-review",
|
||
response_model=dict[str, Any],
|
||
summary="取得 AI Agent runtime verifier evidence review",
|
||
description=(
|
||
"讀取最新已提交的 runtime verifier evidence implementation review；此端點只回傳 "
|
||
"evidence checks、implementation review lanes、redaction policy 與人工操作選項，"
|
||
"不實作或執行 verifier、不讀 canonical target、不寫 rollback work item、不發 Telegram、"
|
||
"不寫 KM / PlayBook trust / timeline / replay score、不啟動 runtime worker。"
|
||
),
|
||
)
|
||
async def get_agent_runtime_verifier_evidence_review() -> dict[str, Any]:
|
||
"""Return the latest read-only AI Agent runtime verifier evidence review."""
|
||
try:
|
||
return await asyncio.to_thread(load_latest_ai_agent_runtime_verifier_evidence_review)
|
||
except FileNotFoundError as exc:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail=str(exc),
|
||
) from exc
|
||
except (json.JSONDecodeError, ValueError) as exc:
|
||
logger.error("ai_agent_runtime_verifier_evidence_review_invalid", error=str(exc))
|
||
raise HTTPException(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail="AI Agent runtime verifier evidence review 無效",
|
||
) from exc
|
||
|
||
|
||
@router.get(
|
||
"/agent-report-truth-actionability-review",
|
||
response_model=dict[str, Any],
|
||
summary="取得 AI Agent 報表真相與告警可處置性審查",
|
||
description=(
|
||
"讀取最新已提交的日報 / 週報 / 月報真相與告警可處置性審查；此端點只回傳 "
|
||
"zero-signal findings、cadence contracts、actionability lanes 與人工操作選項，"
|
||
"不發 Telegram、不修改 CronJob、不改 Prometheus / Alertmanager、不建立 work item、"
|
||
"不寫 KM / PlayBook trust、不啟動 runtime worker。"
|
||
),
|
||
)
|
||
async def get_agent_report_truth_actionability_review() -> dict[str, Any]:
|
||
"""Return the latest read-only AI Agent report truth actionability review."""
|
||
try:
|
||
return await asyncio.to_thread(load_latest_ai_agent_report_truth_actionability_review)
|
||
except FileNotFoundError as exc:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail=str(exc),
|
||
) from exc
|
||
except (json.JSONDecodeError, ValueError) as exc:
|
||
logger.error("ai_agent_report_truth_actionability_review_invalid", error=str(exc))
|
||
raise HTTPException(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail="AI Agent 報表真相與告警可處置性審查無效",
|
||
) from exc
|
||
|
||
|
||
@router.get(
|
||
"/agent-report-automation-review",
|
||
response_model=dict[str, Any],
|
||
summary="取得 AI Agent 日週月報與風險自動化 review",
|
||
description=(
|
||
"讀取最新已提交的 AI Agent 日報、週報、月報、Agent 工作量、圖表化報告、"
|
||
"AI 分析建議與高/中/低風險自動化政策；此端點不排程實發、不送 Telegram、"
|
||
"不啟動中低風險自動執行器、不執行生產優化、不讀 secret、不回傳內部工作視窗對話。"
|
||
),
|
||
)
|
||
async def get_agent_report_automation_review() -> dict[str, Any]:
|
||
"""Return the latest read-only AI Agent report automation review."""
|
||
try:
|
||
return await asyncio.to_thread(load_latest_ai_agent_report_automation_review)
|
||
except FileNotFoundError as exc:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail=str(exc),
|
||
) from exc
|
||
except (json.JSONDecodeError, ValueError) as exc:
|
||
logger.error("ai_agent_report_automation_review_invalid", error=str(exc))
|
||
raise HTTPException(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail="AI Agent 日週月報與風險自動化 review 無效",
|
||
) from exc
|
||
|
||
|
||
@router.get(
|
||
"/agent-owner-approved-fixture-dry-run",
|
||
response_model=dict[str, Any],
|
||
summary="取得 AI Agent owner-approved fixture dry-run 批准包",
|
||
description=(
|
||
"讀取最新已提交的 owner-approved fixture dry-run 批准包；此端點只回傳 fixture-only dry-run 證據，"
|
||
"不寫 KM、不更新 PlayBook trust、不寫 timeline / replay score、不寫 Gateway queue、不呼叫 Telegram Bot API、"
|
||
"不啟動 runtime worker、不開 Redis consumer group、不執行 DB migration、不觸發 workflow、"
|
||
"不執行主機或 cluster 指令、不使用 secrets 或付費 API、不回傳未核准內部細節。"
|
||
),
|
||
)
|
||
async def get_agent_owner_approved_fixture_dry_run() -> dict[str, Any]:
|
||
"""Return the latest read-only AI Agent owner-approved fixture dry-run package."""
|
||
try:
|
||
return await asyncio.to_thread(load_latest_ai_agent_owner_approved_fixture_dry_run)
|
||
except FileNotFoundError as exc:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail=str(exc),
|
||
) from exc
|
||
except (json.JSONDecodeError, ValueError) as exc:
|
||
logger.error("ai_agent_owner_approved_fixture_dry_run_invalid", error=str(exc))
|
||
raise HTTPException(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail="AI Agent owner-approved fixture dry-run 批准包無效",
|
||
) from exc
|
||
|
||
|
||
@router.get(
|
||
"/agent-proactive-operations-contract",
|
||
response_model=dict[str, Any],
|
||
summary="取得 AI Agent 主動營運委派與版本生命週期契約",
|
||
description=(
|
||
"讀取最新已提交的 AI Agent 主動營運、版本生命週期、可委派能力、MCP、RAG 與 Telegram 邊界契約；"
|
||
"此端點不啟用排程、不升級套件、不更新主機、不 pull image、不 auto merge、不送 Telegram、"
|
||
"不呼叫付費服務、不修改生產路由。"
|
||
),
|
||
)
|
||
async def get_agent_proactive_operations_contract() -> dict[str, Any]:
|
||
"""Return the latest read-only AI Agent proactive operations contract."""
|
||
try:
|
||
return await asyncio.to_thread(load_latest_ai_agent_proactive_operations_contract)
|
||
except FileNotFoundError as exc:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail=str(exc),
|
||
) from exc
|
||
except (json.JSONDecodeError, ValueError) as exc:
|
||
logger.error("ai_agent_proactive_operations_contract_invalid", error=str(exc))
|
||
raise HTTPException(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail="AI Agent 主動營運委派與版本生命週期契約無效",
|
||
) from exc
|
||
|
||
|
||
@router.get(
|
||
"/agent-version-freshness-snapshot",
|
||
response_model=dict[str, Any],
|
||
summary="取得 AI Agent repo-only 版本新鮮度快照",
|
||
description=(
|
||
"讀取最新已提交的 AI Agent repo-only 版本新鮮度快照；此端點不啟用每日排程、"
|
||
"不查外部 registry/CVE、不安裝或升級套件、不寫 lockfile、不 build/pull image、"
|
||
"不 probe 主機、不建立 PR、不發 Telegram、不呼叫付費服務、不修改生產路由。"
|
||
),
|
||
)
|
||
async def get_agent_version_freshness_snapshot() -> dict[str, Any]:
|
||
"""Return the latest read-only AI Agent version freshness snapshot."""
|
||
try:
|
||
return await asyncio.to_thread(load_latest_ai_agent_version_freshness_snapshot)
|
||
except FileNotFoundError as exc:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail=str(exc),
|
||
) from exc
|
||
except (json.JSONDecodeError, ValueError) as exc:
|
||
logger.error("ai_agent_version_freshness_snapshot_invalid", error=str(exc))
|
||
raise HTTPException(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail="AI Agent 版本新鮮度快照無效",
|
||
) from exc
|
||
|
||
|
||
@router.get(
|
||
"/agent-tool-adoption-approval-package",
|
||
response_model=dict[str, Any],
|
||
summary="取得 AI Agent 工具採用批准包",
|
||
description=(
|
||
"讀取最新已提交的 Renovate / OSV-Scanner / Trivy / Syft / Grype 工具採用批准包；"
|
||
"此端點不安裝工具、不寫 CI workflow、不下載漏洞資料庫、不查外部 registry、不升級套件、"
|
||
"不寫 lockfile、不 build/pull image、不建立 Gitea PR、不 auto merge、不發 Telegram、"
|
||
"不呼叫付費服務、不修改生產路由。"
|
||
),
|
||
)
|
||
async def get_agent_tool_adoption_approval_package() -> dict[str, Any]:
|
||
"""Return the latest read-only AI Agent tool adoption approval package."""
|
||
try:
|
||
return await asyncio.to_thread(load_latest_ai_agent_tool_adoption_approval_package)
|
||
except FileNotFoundError as exc:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail=str(exc),
|
||
) from exc
|
||
except (json.JSONDecodeError, ValueError) as exc:
|
||
logger.error("ai_agent_tool_adoption_approval_package_invalid", error=str(exc))
|
||
raise HTTPException(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail="AI Agent 工具採用批准包無效",
|
||
) from exc
|
||
|
||
|
||
@router.get(
|
||
"/agent-telegram-action-required-digest-policy",
|
||
response_model=dict[str, Any],
|
||
summary="取得 AI Agent Telegram action-required digest policy",
|
||
description=(
|
||
"讀取最新已提交的 AI Agent Telegram action-required digest policy；"
|
||
"此端點只回傳 critical / action-required / failure-only digest 規則與 redaction 邊界，"
|
||
"不送 Telegram、不寫 Telegram Gateway queue、不改 Alertmanager route / receiver、"
|
||
"不寫 AwoooP event、不觸發 workflow、不查外部掃描、不執行 runtime、不讀取 secret、"
|
||
"不回傳內部協作逐字稿。"
|
||
),
|
||
)
|
||
async def get_agent_telegram_action_required_digest_policy() -> dict[str, Any]:
|
||
"""Return the latest read-only AI Agent Telegram action-required digest policy."""
|
||
try:
|
||
return await asyncio.to_thread(load_latest_ai_agent_telegram_action_required_digest_policy)
|
||
except FileNotFoundError as exc:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail=str(exc),
|
||
) from exc
|
||
except (json.JSONDecodeError, ValueError) as exc:
|
||
logger.error("ai_agent_telegram_action_required_digest_policy_invalid", error=str(exc))
|
||
raise HTTPException(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail="AI Agent Telegram action-required digest policy 無效",
|
||
) from exc
|
||
|
||
|
||
@router.get(
|
||
"/agent-gitea-pr-draft-lane",
|
||
response_model=dict[str, Any],
|
||
summary="取得 AI Agent Gitea PR 草案 lane",
|
||
description=(
|
||
"讀取最新已提交的 AI Agent Gitea PR 草案 lane；"
|
||
"此端點只回傳 grouping、automerge=false、測試證據、rollback、owner response 與 redaction 邊界，"
|
||
"不 push branch、不建立或更新 Gitea PR、不留言、不 auto merge、不觸發 workflow、不改 CI、"
|
||
"不寫 lockfile、不升級套件、不 build/pull image、不改 production route、不發 Telegram、"
|
||
"不讀取 secret、不回傳內部協作逐字稿。"
|
||
),
|
||
)
|
||
async def get_agent_gitea_pr_draft_lane() -> dict[str, Any]:
|
||
"""Return the latest read-only AI Agent Gitea PR draft lane policy."""
|
||
try:
|
||
return await asyncio.to_thread(load_latest_ai_agent_gitea_pr_draft_lane)
|
||
except FileNotFoundError as exc:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail=str(exc),
|
||
) from exc
|
||
except (json.JSONDecodeError, ValueError) as exc:
|
||
logger.error("ai_agent_gitea_pr_draft_lane_invalid", error=str(exc))
|
||
raise HTTPException(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail="AI Agent Gitea PR 草案 lane 無效",
|
||
) from exc
|
||
|
||
|
||
@router.get(
|
||
"/agent-host-stateful-version-inventory",
|
||
response_model=dict[str, Any],
|
||
summary="取得 AI Agent host / K3s / stateful 版本只讀盤點",
|
||
description=(
|
||
"讀取最新已提交的 AI Agent host OS / K3s / stateful services 版本只讀盤點與 "
|
||
"maintenance window 批准包；此端點不 SSH、不執行 host command、不執行 kubectl、"
|
||
"不 apt upgrade、不升級 kernel/K3s、不 drain node、不 reboot、不 restart stateful service、"
|
||
"不做 DB migration、不刪備份、不 restore、不 pull image、不安裝套件、不查外部版本來源、"
|
||
"不 active scan、不發 Telegram、不讀取 secret、不回傳內部協作逐字稿。"
|
||
),
|
||
)
|
||
async def get_agent_host_stateful_version_inventory() -> dict[str, Any]:
|
||
"""Return the latest read-only host / K3s / stateful version inventory."""
|
||
try:
|
||
return await asyncio.to_thread(load_latest_ai_agent_host_stateful_version_inventory)
|
||
except FileNotFoundError as exc:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail=str(exc),
|
||
) from exc
|
||
except (json.JSONDecodeError, ValueError) as exc:
|
||
logger.error("ai_agent_host_stateful_version_inventory_invalid", error=str(exc))
|
||
raise HTTPException(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail="AI Agent host / K3s / stateful 版本只讀盤點無效",
|
||
) from exc
|
||
|
||
|
||
@router.get(
|
||
"/runtime-surface-inventory",
|
||
response_model=dict[str, Any],
|
||
summary="取得 Runtime surface 只讀盤點",
|
||
description=(
|
||
"讀取最新已提交的 API / Web / Worker / K8s runtime surface 盤點；"
|
||
"此端點不呼叫 live cluster、不碰 DB/Redis、不讀 Secret payload、"
|
||
"不執行 rollout/restart/scale/delete、不 patch K8s、不改 workflow、不改生產路由。"
|
||
),
|
||
)
|
||
async def get_runtime_surface_inventory() -> dict[str, Any]:
|
||
"""Return the latest read-only runtime surface inventory."""
|
||
try:
|
||
return await asyncio.to_thread(load_latest_runtime_surface_inventory)
|
||
except FileNotFoundError as exc:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail=str(exc),
|
||
) from exc
|
||
except (json.JSONDecodeError, ValueError) as exc:
|
||
logger.error("runtime_surface_inventory_invalid", error=str(exc))
|
||
raise HTTPException(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail="Runtime surface 盤點快照無效",
|
||
) from exc
|
||
|
||
|
||
@router.get(
|
||
"/gitea-workflow-runner-health",
|
||
response_model=dict[str, Any],
|
||
summary="取得 Gitea 工作流程與 runner 健康合約",
|
||
description=(
|
||
"讀取最新已提交的 Gitea workflow / runner health contract；"
|
||
"此端點不呼叫 Gitea API、不修改 workflow、不重啟 runner、不停止 container、"
|
||
"不讀 Secret payload、不送 Telegram 測試通知、不觸發 deploy 或 migration。"
|
||
),
|
||
)
|
||
async def get_gitea_workflow_runner_health() -> dict[str, Any]:
|
||
"""Return the latest read-only Gitea workflow / runner health contract."""
|
||
try:
|
||
return await asyncio.to_thread(load_latest_gitea_workflow_runner_health)
|
||
except FileNotFoundError as exc:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail=str(exc),
|
||
) from exc
|
||
except (json.JSONDecodeError, ValueError) as exc:
|
||
logger.error("gitea_workflow_runner_health_invalid", error=str(exc))
|
||
raise HTTPException(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail="Gitea 工作流程與 runner 健康合約快照無效",
|
||
) from exc
|
||
|
||
|
||
@router.get(
|
||
"/observability-contract-matrix",
|
||
response_model=dict[str, Any],
|
||
summary="取得監控合約與降噪機會矩陣",
|
||
description=(
|
||
"讀取最新已提交的 Prometheus / Alertmanager / Grafana / SigNoz / ClickHouse / Sentry "
|
||
"只讀 observability matrix；此端點不修改 alert rules、不呼叫 silence API、"
|
||
"不建立 Grafana dashboard、不改 SigNoz / Sentry 設定、不讀 Secret payload、"
|
||
"不送 Telegram 測試通知、不觸發 monitoring deploy。"
|
||
),
|
||
)
|
||
async def get_observability_contract_matrix() -> dict[str, Any]:
|
||
"""Return the latest read-only observability contract matrix."""
|
||
try:
|
||
return await asyncio.to_thread(load_latest_observability_contract_matrix)
|
||
except FileNotFoundError as exc:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail=str(exc),
|
||
) from exc
|
||
except (json.JSONDecodeError, ValueError) as exc:
|
||
logger.error("observability_contract_matrix_invalid", error=str(exc))
|
||
raise HTTPException(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail="監控合約與降噪機會矩陣快照無效",
|
||
) from exc
|
||
|
||
|
||
@router.get(
|
||
"/ai-provider-route-matrix",
|
||
response_model=dict[str, Any],
|
||
summary="取得 AI Provider 路由只讀矩陣",
|
||
description=(
|
||
"讀取最新已提交的 AI Router / Ollama / OpenClaw / Nemotron / Gemini provider route matrix；"
|
||
"此端點不切換 provider、不呼叫 Gemini / NVIDIA / Claude、不改 USE_AI_ROUTER、"
|
||
"不修改 fallback order、不讀 Secret payload、不進 shadow / canary、"
|
||
"不觸發 workflow / deploy / reload / runtime execution。"
|
||
),
|
||
)
|
||
async def get_ai_provider_route_matrix() -> dict[str, Any]:
|
||
"""Return the latest read-only AI provider route matrix."""
|
||
try:
|
||
return await asyncio.to_thread(load_latest_ai_provider_route_matrix)
|
||
except FileNotFoundError as exc:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail=str(exc),
|
||
) from exc
|
||
except (json.JSONDecodeError, ValueError) as exc:
|
||
logger.error("ai_provider_route_matrix_invalid", error=str(exc))
|
||
raise HTTPException(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail="AI Provider 路由只讀矩陣快照無效",
|
||
) from exc
|
||
|
||
|
||
@router.get(
|
||
"/service-health-gap-matrix",
|
||
response_model=dict[str, Any],
|
||
summary="取得服務健康缺口與過期端點矩陣",
|
||
description=(
|
||
"讀取最新已提交的 service health gap matrix；此端點不做 live probe、"
|
||
"不重啟服務、不修改 endpoint / ConfigMap、不讀 Secret/Redis/DB payload、"
|
||
"不發通知、不觸發 workflow/deploy/reload/runtime execution。"
|
||
),
|
||
)
|
||
async def get_service_health_gap_matrix() -> dict[str, Any]:
|
||
"""Return the latest read-only service health gap matrix."""
|
||
try:
|
||
return await asyncio.to_thread(load_latest_service_health_gap_matrix)
|
||
except FileNotFoundError as exc:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail=str(exc),
|
||
) from exc
|
||
except (json.JSONDecodeError, ValueError) as exc:
|
||
logger.error("service_health_gap_matrix_invalid", error=str(exc))
|
||
raise HTTPException(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail="服務健康缺口矩陣快照無效",
|
||
) from exc
|
||
|
||
|
||
@router.get(
|
||
"/backup-dr-target-inventory",
|
||
response_model=dict[str, Any],
|
||
summary="取得 Backup / DR 目標盤點",
|
||
description=(
|
||
"讀取最新已提交的 Backup / DR 目標盤點；"
|
||
"此端點不呼叫外部來源、不執行備份/restore/offsite sync、"
|
||
"不寫 credential marker、不改排程、不批准任何破壞性操作。"
|
||
),
|
||
)
|
||
async def get_backup_dr_target_inventory() -> dict[str, Any]:
|
||
"""Return the latest read-only Backup / DR target inventory."""
|
||
try:
|
||
return await asyncio.to_thread(load_latest_backup_dr_target_inventory)
|
||
except FileNotFoundError as exc:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail=str(exc),
|
||
) from exc
|
||
except (json.JSONDecodeError, ValueError) as exc:
|
||
logger.error("backup_dr_target_inventory_invalid", error=str(exc))
|
||
raise HTTPException(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail="Backup / DR target inventory is invalid",
|
||
) from exc
|
||
|
||
|
||
@router.get(
|
||
"/backup-dr-readiness-matrix",
|
||
response_model=dict[str, Any],
|
||
summary="取得 Backup / DR 準備度矩陣",
|
||
description=(
|
||
"讀取最新已提交的 Backup / DR 準備度矩陣；"
|
||
"此端點不呼叫外部來源、不執行備份/restore/offsite sync、"
|
||
"不寫 credential marker、不改排程、不批准任何破壞性操作。"
|
||
),
|
||
)
|
||
async def get_backup_dr_readiness_matrix() -> dict[str, Any]:
|
||
"""Return the latest read-only Backup / DR readiness matrix."""
|
||
try:
|
||
return await asyncio.to_thread(load_latest_backup_dr_readiness_matrix)
|
||
except FileNotFoundError as exc:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail=str(exc),
|
||
) from exc
|
||
except (json.JSONDecodeError, ValueError) as exc:
|
||
logger.error("backup_dr_readiness_matrix_invalid", error=str(exc))
|
||
raise HTTPException(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail="Backup / DR readiness matrix is invalid",
|
||
) from exc
|
||
|
||
|
||
@router.get(
|
||
"/backup-notification-policy",
|
||
response_model=dict[str, Any],
|
||
summary="取得備份通知政策",
|
||
description=(
|
||
"讀取最新已提交的備份通知政策；此端點只回傳 success-noise suppression、"
|
||
"failure/action-required 升級與每日摘要合約，不送通知、不執行備份/restore/offsite sync、"
|
||
"不寫 credential marker、不改排程、不寫 workflow、不發 Telegram 測試訊息。"
|
||
),
|
||
)
|
||
async def get_backup_notification_policy() -> dict[str, Any]:
|
||
"""Return the latest read-only backup notification policy."""
|
||
try:
|
||
return await asyncio.to_thread(load_latest_backup_notification_policy)
|
||
except FileNotFoundError as exc:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail=str(exc),
|
||
) from exc
|
||
except (json.JSONDecodeError, ValueError) as exc:
|
||
logger.error("backup_notification_policy_invalid", error=str(exc))
|
||
raise HTTPException(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail="備份通知政策快照無效",
|
||
) from exc
|
||
|
||
|
||
@router.get(
|
||
"/service-health-failure-notification-policy",
|
||
response_model=dict[str, Any],
|
||
summary="取得服務健康失敗限定通知合約",
|
||
description=(
|
||
"讀取最新已提交的 service health failure-only Telegram / AwoooP 通知合約；"
|
||
"此端點只回傳成功降噪、action-required 與 failure escalation 規則，"
|
||
"不送通知、不做 live probe、不重啟服務、不改 endpoint、不觸發 workflow / runtime execution、"
|
||
"不讀取 secret payload、不回傳內部協作逐字稿或提示詞。"
|
||
),
|
||
)
|
||
async def get_service_health_failure_notification_policy() -> dict[str, Any]:
|
||
"""Return the latest read-only service health failure-only notification policy."""
|
||
try:
|
||
return await asyncio.to_thread(load_latest_service_health_failure_notification_policy)
|
||
except FileNotFoundError as exc:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail=str(exc),
|
||
) from exc
|
||
except (json.JSONDecodeError, ValueError) as exc:
|
||
logger.error("service_health_failure_notification_policy_invalid", error=str(exc))
|
||
raise HTTPException(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail="服務健康失敗限定通知合約快照無效",
|
||
) from exc
|
||
|
||
|
||
@router.get(
|
||
"/backup-restore-drill-approval-package-template",
|
||
response_model=dict[str, Any],
|
||
summary="取得 Backup / DR 復原演練批准包模板",
|
||
description=(
|
||
"讀取最新已提交的 Backup / DR restore drill、credential escrow review、"
|
||
"K8s resource recovery、observability recovery 與 route reconstruction 批准包模板；"
|
||
"此端點只回傳 read-only template，不執行 backup、restore、offsite sync、"
|
||
"不寫 credential marker、不改排程、不寫 workflow、不送 Telegram 測試通知、"
|
||
"不輸出 secret 明文、不做破壞性 prune、不呼叫付費 API、不建立 shadow/canary、不改生產路由。"
|
||
),
|
||
)
|
||
async def get_backup_restore_drill_approval_package_template() -> dict[str, Any]:
|
||
"""Return the latest read-only Backup / DR restore drill approval package template."""
|
||
try:
|
||
return await asyncio.to_thread(load_latest_backup_restore_drill_approval_package_template)
|
||
except FileNotFoundError as exc:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail=str(exc),
|
||
) from exc
|
||
except (json.JSONDecodeError, ValueError) as exc:
|
||
logger.error("backup_restore_drill_approval_package_template_invalid", error=str(exc))
|
||
raise HTTPException(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail="Backup / DR 復原演練批准包模板快照無效",
|
||
) from exc
|
||
|
||
|
||
@router.get(
|
||
"/offsite-escrow-readiness-status",
|
||
response_model=dict[str, Any],
|
||
summary="取得異地 / Escrow 準備度狀態",
|
||
description=(
|
||
"讀取最新已提交的異地備份、credential escrow 與 K8s resource offsite readiness 狀態；"
|
||
"此端點只回傳 read-only status，不執行 backup、restore、offsite sync、"
|
||
"不寫 credential marker、不讀 credential、不輸出 secret 明文、不改排程、不寫 workflow、"
|
||
"不送 Telegram 測試通知、不做破壞性 prune、不改生產路由。"
|
||
),
|
||
)
|
||
async def get_offsite_escrow_readiness_status() -> dict[str, Any]:
|
||
"""Return the latest read-only offsite / escrow readiness status."""
|
||
try:
|
||
return await asyncio.to_thread(load_latest_offsite_escrow_readiness_status)
|
||
except FileNotFoundError as exc:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail=str(exc),
|
||
) from exc
|
||
except (json.JSONDecodeError, ValueError) as exc:
|
||
logger.error("offsite_escrow_readiness_status_invalid", error=str(exc))
|
||
raise HTTPException(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail="異地 / Escrow 準備度狀態快照無效",
|
||
) from exc
|
||
|
||
|
||
@router.get(
|
||
"/package-supply-chain-inventory",
|
||
response_model=dict[str, Any],
|
||
summary="取得套件 / 供應鏈盤點",
|
||
description=(
|
||
"讀取最新已提交的套件 / 供應鏈盤點；"
|
||
"此端點不呼叫外部來源、不安裝依賴、不升級套件、"
|
||
"不寫 lockfile、不查外部 CVE、不重建 image、不改生產路由。"
|
||
),
|
||
)
|
||
async def get_package_supply_chain_inventory() -> dict[str, Any]:
|
||
"""Return the latest read-only package supply-chain inventory."""
|
||
try:
|
||
return await asyncio.to_thread(load_latest_package_supply_chain_inventory)
|
||
except FileNotFoundError as exc:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail=str(exc),
|
||
) from exc
|
||
except (json.JSONDecodeError, ValueError) as exc:
|
||
logger.error("package_supply_chain_inventory_invalid", error=str(exc))
|
||
raise HTTPException(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail="套件 / 供應鏈盤點快照無效",
|
||
) from exc
|
||
|
||
|
||
@router.get(
|
||
"/javascript-package-inventory",
|
||
response_model=dict[str, Any],
|
||
summary="取得 JavaScript 套件盤點",
|
||
description=(
|
||
"讀取最新已提交的 JavaScript / pnpm 套件盤點；"
|
||
"此端點不呼叫外部來源、不安裝套件、不升級套件、"
|
||
"不寫 lockfile、不執行 npm audit、不改生產路由。"
|
||
),
|
||
)
|
||
async def get_javascript_package_inventory() -> dict[str, Any]:
|
||
"""Return the latest read-only JavaScript package inventory."""
|
||
try:
|
||
return await asyncio.to_thread(load_latest_javascript_package_inventory)
|
||
except FileNotFoundError as exc:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail=str(exc),
|
||
) from exc
|
||
except (json.JSONDecodeError, ValueError) as exc:
|
||
logger.error("javascript_package_inventory_invalid", error=str(exc))
|
||
raise HTTPException(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail="JavaScript 套件盤點快照無效",
|
||
) from exc
|
||
|
||
|
||
@router.get(
|
||
"/docker-build-surface-inventory",
|
||
response_model=dict[str, Any],
|
||
summary="取得 Docker build surface 盤點",
|
||
description=(
|
||
"讀取最新已提交的 Docker base image 與 build surface 盤點；"
|
||
"此端點不執行 docker build、不 pull image、不推 registry、"
|
||
"不查外部 CVE、不安裝套件、不改生產路由。"
|
||
),
|
||
)
|
||
async def get_docker_build_surface_inventory() -> dict[str, Any]:
|
||
"""Return the latest read-only Docker build surface inventory."""
|
||
try:
|
||
return await asyncio.to_thread(load_latest_docker_build_surface_inventory)
|
||
except FileNotFoundError as exc:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail=str(exc),
|
||
) from exc
|
||
except (json.JSONDecodeError, ValueError) as exc:
|
||
logger.error("docker_build_surface_inventory_invalid", error=str(exc))
|
||
raise HTTPException(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail="Docker build surface 盤點快照無效",
|
||
) from exc
|
||
|
||
|
||
@router.get(
|
||
"/dependency-risk-policy",
|
||
response_model=dict[str, Any],
|
||
summary="取得依賴風險政策",
|
||
description=(
|
||
"讀取最新已提交的 CVE / license / drift 嚴重度政策；"
|
||
"此端點不呼叫外部 CVE 或 license 來源、不安裝套件、不升級套件、"
|
||
"不寫 lockfile、不執行 docker build、不 pull image、不推 registry、"
|
||
"不呼叫付費 API、不建立 shadow/canary、不改生產路由。"
|
||
),
|
||
)
|
||
async def get_dependency_risk_policy() -> dict[str, Any]:
|
||
"""Return the latest read-only dependency risk policy."""
|
||
try:
|
||
return await asyncio.to_thread(load_latest_dependency_risk_policy)
|
||
except FileNotFoundError as exc:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail=str(exc),
|
||
) from exc
|
||
except (json.JSONDecodeError, ValueError) as exc:
|
||
logger.error("dependency_risk_policy_invalid", error=str(exc))
|
||
raise HTTPException(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail="依賴風險政策快照無效",
|
||
) from exc
|
||
|
||
|
||
@router.get(
|
||
"/dependency-drift-check-plan",
|
||
response_model=dict[str, Any],
|
||
summary="取得依賴漂移檢查設計",
|
||
description=(
|
||
"讀取最新已提交的定期依賴漂移、外部資料來源與 AI Agent 市場觀察設計；"
|
||
"此端點只回傳 read-only plan，不啟用排程、不寫 workflow、不呼叫外部 CVE / license / registry / 市場來源、"
|
||
"不安裝 SDK、不呼叫付費 API、不安裝或升級套件、不寫 lockfile、"
|
||
"不執行 docker build、不 pull image、不推 registry、不建立 shadow/canary、不改生產路由。"
|
||
),
|
||
)
|
||
async def get_dependency_drift_check_plan() -> dict[str, Any]:
|
||
"""Return the latest read-only dependency drift check plan."""
|
||
try:
|
||
return await asyncio.to_thread(load_latest_dependency_drift_check_plan)
|
||
except FileNotFoundError as exc:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail=str(exc),
|
||
) from exc
|
||
except (json.JSONDecodeError, ValueError) as exc:
|
||
logger.error("dependency_drift_check_plan_invalid", error=str(exc))
|
||
raise HTTPException(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail="依賴漂移檢查設計快照無效",
|
||
) from exc
|
||
|
||
|
||
@router.get(
|
||
"/dependency-upgrade-approval-package-template",
|
||
response_model=dict[str, Any],
|
||
summary="取得依賴升級批准包模板",
|
||
description=(
|
||
"讀取最新已提交的依賴升級、digest pin、publish boundary 與外部來源啟用批准包模板；"
|
||
"此端點只回傳 read-only template，不安裝或升級套件、不寫 manifest 或 lockfile、"
|
||
"不修改 Dockerfile、不執行 docker build、不 pull image、不推 registry、不 publish package、"
|
||
"不安裝 SDK、不呼叫付費 API、不建立 shadow/canary、不改生產路由。"
|
||
),
|
||
)
|
||
async def get_dependency_upgrade_approval_package_template() -> dict[str, Any]:
|
||
"""Return the latest read-only dependency upgrade approval package template."""
|
||
try:
|
||
return await asyncio.to_thread(load_latest_dependency_upgrade_approval_package_template)
|
||
except FileNotFoundError as exc:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_404_NOT_FOUND,
|
||
detail=str(exc),
|
||
) from exc
|
||
except (json.JSONDecodeError, ValueError) as exc:
|
||
logger.error("dependency_upgrade_approval_package_template_invalid", error=str(exc))
|
||
raise HTTPException(
|
||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
detail="依賴升級批准包模板快照無效",
|
||
) from exc
|
||
|
||
|
||
# =============================================================================
|
||
# Integration with Incident Flow
|
||
# =============================================================================
|
||
|
||
async def trigger_agent_analysis_for_incident(
|
||
incident_id: str,
|
||
background_tasks: BackgroundTasks,
|
||
) -> str | None:
|
||
"""
|
||
整合點: 當 Incident 需要複雜決策時自動觸發 Agent Teams
|
||
|
||
這個函數可被 incident_engine 或 webhooks 調用
|
||
|
||
Returns:
|
||
task_id if triggered, None if skipped
|
||
"""
|
||
service = get_agent_service()
|
||
|
||
task_id, incident = await service.trigger_for_incident(incident_id)
|
||
|
||
if task_id is None:
|
||
return None
|
||
|
||
if incident is None:
|
||
return None
|
||
|
||
# 加入背景任務
|
||
background_tasks.add_task(
|
||
service.run_analysis,
|
||
task_id,
|
||
incident,
|
||
)
|
||
|
||
return task_id
|