From cb7c775d12772212b13ed7bcac5738a8cf48043b Mon Sep 17 00:00:00 2001 From: ogt Date: Thu, 16 Jul 2026 23:19:52 +0800 Subject: [PATCH] fix(telegram): fail closed unusable callbacks --- agent99-control-plane.ps1 | 112 +++++- apps/api/src/api/v1/telegram.py | 36 +- apps/api/src/api/v1/telegram_webhook.py | 25 +- apps/api/src/models/agent99_completion.py | 13 + .../services/agent99_telegram_lifecycle.py | 160 ++++++-- apps/api/src/services/security_interceptor.py | 8 +- .../src/services/telegram_button_registry.py | 222 +++++++++++ apps/api/src/services/telegram_gateway.py | 206 +++++++++- .../test_agent99_telegram_lifecycle_api.py | 97 +++++ .../tests/test_telegram_button_consistency.py | 14 +- ...test_telegram_callback_ingress_contract.py | 366 ++++++++++++++++++ 11 files changed, 1178 insertions(+), 81 deletions(-) create mode 100644 apps/api/src/services/telegram_button_registry.py create mode 100644 apps/api/tests/test_telegram_callback_ingress_contract.py diff --git a/agent99-control-plane.ps1 b/agent99-control-plane.ps1 index 53afecfff..3226cbdc5 100644 --- a/agent99-control-plane.ps1 +++ b/agent99-control-plane.ps1 @@ -875,15 +875,26 @@ function Get-AgentIncidentCardModel { $targetName = [string](Get-AgentObjectValue $Data "target" (Get-AgentObjectValue $Data "alertService" "")) $alertId = [string](Get-AgentObjectValue $Data "alertId" "") $correlationKey = [string](Get-AgentObjectValue $Data "correlationKey" "") - $runKey = [string](Get-AgentObjectValue $Data "runId" (Get-AgentObjectValue $Data "automationRunId" (Get-AgentObjectValue $Data "id" ""))) + $runKey = [string](Get-AgentObjectValue $Data "runId" (Get-AgentObjectValue $Data "run_id" (Get-AgentObjectValue $Data "automationRunId" (Get-AgentObjectValue $Data "automation_run_id" (Get-AgentObjectValue $Data "id" ""))))) $problem = if ($Data -and $Data.PSObject.Properties["problem"]) { $Data.problem } else { $null } $outcome = if ($Data -and $Data.PSObject.Properties["outcome"]) { $Data.outcome } else { $null } + $identity = if ($Data -and $Data.PSObject.Properties["identity"]) { $Data.identity } else { $null } $problemKey = [string](Get-AgentObjectValue $problem "key" "") $outcomeState = [string](Get-AgentObjectValue $outcome "state" (Get-AgentObjectValue $Data "outcomeState" "")) $verifierPassed = Convert-AgentBool (Get-AgentObjectValue $outcome "verifierPassed" (Get-AgentObjectValue $Data "verifierPassed" $false)) $sourceResolved = Convert-AgentBool (Get-AgentObjectValue $outcome "sourceEventResolved" (Get-AgentObjectValue $Data "sourceEventResolved" $false)) $executionOk = Convert-AgentBool (Get-AgentObjectValue $Data "executionOk" (Get-AgentObjectValue $Data "ok" $false)) $postVerifyOk = Convert-AgentBool (Get-AgentObjectValue $Data "postVerifyOk" $false) + $traceKey = [string](Get-AgentObjectValue $Data "traceId" (Get-AgentObjectValue $Data "trace_id" (Get-AgentObjectValue $outcome "traceId" (Get-AgentObjectValue $outcome "trace_id" (Get-AgentObjectValue $identity "trace_id" ""))))) + $workItemKey = [string](Get-AgentObjectValue $Data "workItemId" (Get-AgentObjectValue $Data "work_item_id" (Get-AgentObjectValue $outcome "workItemId" (Get-AgentObjectValue $outcome "work_item_id" (Get-AgentObjectValue $identity "work_item_id" ""))))) + $verifierName = [string](Get-AgentObjectValue $Data "verifierName" (Get-AgentObjectValue $Data "verifier_name" (Get-AgentObjectValue $outcome "verifierName" (Get-AgentObjectValue $outcome "verifier_name" "")))) + $executorName = [string](Get-AgentObjectValue $Data "executorName" (Get-AgentObjectValue $Data "executor_name" (Get-AgentObjectValue $Data "executor" ""))) + if (-not $executorName -and $modeName) { + $executorName = "Agent99:$modeName" + } elseif (-not $executorName -and $identity) { + $routeId = [string](Get-AgentObjectValue $identity "route_id" "") + if ($routeId) { $executorName = "Agent99:$routeId" } + } $family = switch -Regex ($EventType) { "^performance_|^load_shed_" { "host_performance"; break } @@ -937,6 +948,68 @@ function Get-AgentIncidentCardModel { $lifecycle = if ($executionOk -and ($postVerifyOk -or $verifierPassed)) { "recovered" } else { "verifying" } } + # Project only evidence already present in this Agent99 event. Missing stage + # receipts remain PENDING instead of being inferred green from card delivery. + $allowedLifecycleStageStates = @("pending", "running", "passed", "failed", "not_applicable") + $receivedStatus = "passed" + $checkStatus = "pending" + $applyStatus = "pending" + $verifierStatus = "pending" + $closureStatus = "pending" + $outcomeChecks = if ($outcome -and $outcome.PSObject.Properties["checks"]) { @($outcome.checks) } else { @() } + if ($outcomeChecks.Count -gt 0) { + $requiredCheckFailures = @( + $outcomeChecks | Where-Object { + (Convert-AgentBool (Get-AgentObjectValue $_ "required" $true)) -and + -not (Convert-AgentBool (Get-AgentObjectValue $_ "passed" $false)) + } + ) + $checkStatus = if ($requiredCheckFailures.Count -gt 0) { "failed" } else { "passed" } + } + + $controlledApplyPresent = [bool]($Data -and $Data.PSObject.Properties["controlledApply"]) + $controlledApply = [bool]($controlledApplyPresent -and (Convert-AgentBool $Data.controlledApply)) + $exitCodePresent = [bool]($Data -and $Data.PSObject.Properties["exitCode"]) + $executionOkPresent = [bool]($Data -and ($Data.PSObject.Properties["executionOk"] -or $Data.PSObject.Properties["ok"])) + if ($controlledApplyPresent -and -not $controlledApply) { + $applyStatus = "not_applicable" + } elseif ($controlledApply -and $outcomeState -eq "candidate_ready") { + $applyStatus = "pending" + } elseif ($controlledApply -and $exitCodePresent) { + $applyStatus = if ([int]$Data.exitCode -eq 0) { "passed" } else { "failed" } + } elseif ($controlledApply -and $executionOkPresent) { + $applyStatus = if ($executionOk) { "passed" } else { "failed" } + } + + if ($verifierPassed -or $postVerifyOk) { + $verifierStatus = "passed" + } elseif ($outcomeState -in @("failed", "blocked", "degraded")) { + $verifierStatus = "failed" + } elseif ($lifecycle -eq "blocked") { + $verifierStatus = "failed" + } elseif ($lifecycle -eq "verifying") { + $verifierStatus = "running" + } + + if (($outcomeState -eq "resolved" -and $sourceResolved) -or $lifecycle -eq "recovered") { + $closureStatus = "passed" + } elseif ($outcomeState -in @("failed", "blocked", "degraded") -or $lifecycle -eq "blocked") { + $closureStatus = "failed" + } elseif ($lifecycle -in @("remediating", "verifying")) { + $closureStatus = "running" + } + + $explicitReceivedStatus = [string](Get-AgentObjectValue $Data "receivedStatus" (Get-AgentObjectValue $Data "received_status" "")) + $explicitCheckStatus = [string](Get-AgentObjectValue $Data "checkStatus" (Get-AgentObjectValue $Data "check_status" "")) + $explicitApplyStatus = [string](Get-AgentObjectValue $Data "applyStatus" (Get-AgentObjectValue $Data "apply_status" "")) + $explicitVerifierStatus = [string](Get-AgentObjectValue $Data "verifierStatus" (Get-AgentObjectValue $Data "verifier_status" "")) + $explicitClosureStatus = [string](Get-AgentObjectValue $Data "closureStatus" (Get-AgentObjectValue $Data "closure_status" "")) + if ($explicitReceivedStatus -in $allowedLifecycleStageStates) { $receivedStatus = $explicitReceivedStatus } + if ($explicitCheckStatus -in $allowedLifecycleStageStates) { $checkStatus = $explicitCheckStatus } + if ($explicitApplyStatus -in $allowedLifecycleStageStates) { $applyStatus = $explicitApplyStatus } + if ($explicitVerifierStatus -in $allowedLifecycleStageStates) { $verifierStatus = $explicitVerifierStatus } + if ($explicitClosureStatus -in $allowedLifecycleStageStates) { $closureStatus = $explicitClosureStatus } + $lifecycleLabel = switch ($lifecycle) { "detected" { "已偵測 / DETECTED" } "investigating" { "調查中 / INVESTIGATING" } @@ -1174,6 +1247,15 @@ function Get-AgentIncidentCardModel { lifecycleLabel = $lifecycleLabel stateKey = "$lifecycle|$Severity" runKey = $runKey + traceKey = $traceKey + workItemKey = $workItemKey + executorName = if ($executorName) { Limit-AgentTextLine $executorName 150 } else { "" } + verifierName = if ($verifierName) { Limit-AgentTextLine $verifierName 150 } else { "" } + receivedStatus = $receivedStatus + checkStatus = $checkStatus + applyStatus = $applyStatus + verifierStatus = $verifierStatus + closureStatus = $closureStatus title = Limit-AgentTextLine $title 96 target = Limit-AgentTextLine $target 96 impact = Limit-AgentTextLine $impact 240 @@ -1912,7 +1994,24 @@ function Send-AgentTelegram { $attempt.visualError = "card_image_render_failed" } - $deliverySource = "$($card.incidentId)|$($card.fingerprint)|$($card.stateKey)|$($card.runKey)" + # A stage receipt transition is a new visible lifecycle state even when the + # coarse lifecycle label is unchanged. Include exact stage/identity fields + # so send-once dedup cannot suppress Check/Apply/Verifier/Closure progress. + $deliverySource = @( + $card.incidentId, + $card.fingerprint, + $card.stateKey, + $card.runKey, + $card.traceKey, + $card.workItemKey, + $card.receivedStatus, + $card.checkStatus, + $card.applyStatus, + $card.verifierStatus, + $card.closureStatus, + $card.executorName, + $card.verifierName + ) -join "|" $deliveryHashBytes = [Security.Cryptography.SHA256]::Create().ComputeHash([Text.Encoding]::UTF8.GetBytes($deliverySource)) $deliveryHash = [BitConverter]::ToString($deliveryHashBytes).Replace("-", "").ToLowerInvariant() $deliveryId = "agent99-lifecycle-$($deliveryHash.Substring(0, 40))" @@ -1928,6 +2027,15 @@ function Send-AgentTelegram { lifecycle = if ($card.lifecycle -in @("detected", "investigating", "remediating", "verifying", "recovered", "blocked")) { [string]$card.lifecycle } else { "unknown" } state_key = [string]$card.stateKey run_id = if ($card.runKey) { [string]$card.runKey } else { $null } + trace_id = if ($card.traceKey) { [string]$card.traceKey } else { $null } + work_item_id = if ($card.workItemKey) { [string]$card.workItemKey } else { $null } + executor_name = if ($card.executorName) { [string]$card.executorName } else { $null } + verifier_name = if ($card.verifierName) { [string]$card.verifierName } else { $null } + received_status = [string]$card.receivedStatus + check_status = [string]$card.checkStatus + apply_status = [string]$card.applyStatus + verifier_status = [string]$card.verifierStatus + closure_status = [string]$card.closureStatus title = [string]$card.title target = [string]$card.target impact = [string]$card.impact diff --git a/apps/api/src/api/v1/telegram.py b/apps/api/src/api/v1/telegram.py index f83ed62f6..019f98ce4 100644 --- a/apps/api/src/api/v1/telegram.py +++ b/apps/api/src/api/v1/telegram.py @@ -9,7 +9,7 @@ Phase 5.5: Long Polling 重構 (內網修復) - 新: Long Polling 模式 (主動輪詢) - 適用內網環境 Endpoints: -- POST /api/v1/telegram/webhook - [已棄用] 接收 Telegram Bot Update +- POST /api/v1/telegram/webhook - canonical owner: api.v1.telegram_webhook - POST /api/v1/telegram/test-push - 測試推送 (僅開發模式) - GET /api/v1/telegram/health - Gateway 健康檢查 @@ -35,6 +35,7 @@ from src.services.security_interceptor import ( NonceReplayError, UserNotWhitelistedError, ) +from src.services.telegram_button_registry import telegram_button_registry_snapshot from src.services.telegram_gateway import ( TelegramGatewayError, _telegram_send_delivery_succeeded, @@ -268,17 +269,15 @@ async def _sync_telegram_rejection(approval_id: str) -> bool: # Endpoints # ============================================================================= -@router.post( - "/webhook", - summary="[已棄用] Telegram Bot Webhook", - description="⚠️ 已棄用：內網環境請使用 Long Polling 模式。此端點保留供外網環境或測試使用。", - deprecated=True, -) async def telegram_webhook( update: TelegramUpdate, ) -> dict: """ - 接收 Telegram Bot Update + Legacy callback implementation retained for direct compatibility tests. + + This function is deliberately not registered as an API route. The sole + production owner of ``/api/v1/telegram/webhook`` is + ``src.api.v1.telegram_webhook``. 處理流程: 1. 驗證 Update 類型 (僅處理 callback_query) @@ -563,13 +562,26 @@ async def test_push( summary="Telegram Gateway 健康檢查", ) async def telegram_health() -> dict: - """Telegram Gateway 健康狀態 (含 Long Polling 狀態)""" + """Telegram Gateway health with callback transport truth.""" gateway = get_telegram_gateway() + ingress = gateway.callback_ingress_status() + configured = bool(settings.OPENCLAW_TG_BOT_TOKEN and settings.SRE_GROUP_CHAT_ID) + interactive_ready = bool(ingress.get("interactive_buttons_ready") is True) return { - "status": "configured" if settings.OPENCLAW_TG_BOT_TOKEN else "not_configured", - "mode": "long_polling", # Phase 5.5: 已從 webhook 切換至 long_polling - "polling_active": gateway._polling_active, + "status": ( + "ready" + if interactive_ready + else "degraded_callback_ingress_unverified" + if configured + else "not_configured" + ), + "configuration_status": "configured" if configured else "not_configured", + "mode": ingress["mode"], + "polling_active": ingress["polling_active"], + "interactive_buttons_ready": interactive_ready, + "callback_ingress": ingress, + "button_registry": telegram_button_registry_snapshot(), "bot_token_set": bool(settings.OPENCLAW_TG_BOT_TOKEN), "chat_id_set": bool(settings.SRE_GROUP_CHAT_ID), "sre_group_chat_id_set": bool(settings.SRE_GROUP_CHAT_ID), diff --git a/apps/api/src/api/v1/telegram_webhook.py b/apps/api/src/api/v1/telegram_webhook.py index fbdf8bb84..4f6392fae 100644 --- a/apps/api/src/api/v1/telegram_webhook.py +++ b/apps/api/src/api/v1/telegram_webhook.py @@ -32,6 +32,12 @@ def _verify_secret_token( """ expected = getattr(settings, "TELEGRAM_WEBHOOK_SECRET", "") if not expected: + if settings.ENVIRONMENT == "prod": + logger.error("telegram_webhook_secret_not_configured_fail_closed") + raise HTTPException( + status_code=503, + detail="Telegram webhook secret is not configured", + ) # dev 環境：未配置 secret → 跳過驗證 return if x_telegram_bot_api_secret_token != expected: @@ -47,7 +53,7 @@ async def telegram_webhook(request: Request) -> dict: """ 接收 Telegram Bot API webhook update。 - 目前：記錄 update 類型 + 回傳 200 OK。 + Callback Query：交給唯一 TelegramGateway handler，完成 mirror、驗證與 action。 WS4 Hermes NL：在此呼叫 Hermes NL router 處理自然語言指令。 Telegram Bot API 要求此端點在 1 秒內回傳 200，耗時操作須非同步排程。 @@ -73,7 +79,6 @@ async def telegram_webhook(request: Request) -> dict: if update_type == "callback_query": callback = body.get("callback_query", {}) or {} - message = callback.get("message", {}) or {} user = callback.get("from", {}) or {} callback_query_id = callback.get("id") callback_data = callback.get("data") @@ -82,17 +87,11 @@ async def telegram_webhook(request: Request) -> dict: from src.services.telegram_gateway import get_telegram_gateway gateway = get_telegram_gateway() - mirror_callback = getattr(gateway, "mirror_callback_query_received", None) - if callable(mirror_callback): - await mirror_callback( - update_id=body.get("update_id"), - callback_query_id=callback_query_id, - callback_data=callback_data, - user_id=user_id, - username=user.get("username") or user.get("first_name") or str(user_id), - message_id=message.get("message_id"), - chat_id=(message.get("chat") or {}).get("id"), - ) + await gateway._handle_callback_query( + body.get("update_id"), + callback, + ingress_source="canonical_webhook", + ) # WS5: chat_member 同步 Approvers 白名單（ADR-093） if update_type in ("chat_member", "my_chat_member") or ( diff --git a/apps/api/src/models/agent99_completion.py b/apps/api/src/models/agent99_completion.py index ac8244215..98bd3eb09 100644 --- a/apps/api/src/models/agent99_completion.py +++ b/apps/api/src/models/agent99_completion.py @@ -12,6 +12,10 @@ from typing import Literal from pydantic import BaseModel, ConfigDict, Field, model_validator +Agent99LifecycleStageStatus = Literal[ + "pending", "running", "passed", "failed", "not_applicable" +] + class Agent99TelegramReceipt(BaseModel): model_config = ConfigDict(extra="forbid", str_strip_whitespace=True) @@ -180,6 +184,15 @@ class Agent99TelegramLifecycleRequest(BaseModel): ] state_key: str = Field(min_length=3, max_length=96) run_id: str | None = Field(default=None, max_length=180) + trace_id: str | None = Field(default=None, max_length=180) + work_item_id: str | None = Field(default=None, max_length=180) + executor_name: str | None = Field(default=None, max_length=160) + verifier_name: str | None = Field(default=None, max_length=160) + received_status: Agent99LifecycleStageStatus | None = None + check_status: Agent99LifecycleStageStatus | None = None + apply_status: Agent99LifecycleStageStatus | None = None + verifier_status: Agent99LifecycleStageStatus | None = None + closure_status: Agent99LifecycleStageStatus | None = None title: str = Field(min_length=1, max_length=120) target: str = Field(min_length=1, max_length=120) impact: str = Field(min_length=1, max_length=300) diff --git a/apps/api/src/services/agent99_telegram_lifecycle.py b/apps/api/src/services/agent99_telegram_lifecycle.py index 8351b59e4..4bfd7e0dd 100644 --- a/apps/api/src/services/agent99_telegram_lifecycle.py +++ b/apps/api/src/services/agent99_telegram_lifecycle.py @@ -25,9 +25,62 @@ _SEVERITY_LABELS = { "info": "資訊 / INFO", } +_STAGE_STATUS_LABELS = { + "pending": "待回報 / PENDING", + "running": "執行中 / RUNNING", + "passed": "完成 / PASSED", + "failed": "失敗 / FAILED", + "not_applicable": "不適用 / N/A", +} + +_STAGE_STATUS_COMPACT = { + "pending": "PENDING", + "running": "RUNNING", + "passed": "PASSED", + "failed": "FAILED", + "not_applicable": "N/A", +} + +_LIFECYCLE_STAGES = ( + ("Received", "received_status"), + ("Check", "check_status"), + ("Apply", "apply_status"), + ("Verifier", "verifier_status"), + ("Closure", "closure_status"), +) + _DURABLE_RECONCILE_DELAYS_SECONDS = (0.25, 0.5, 1.0) +def _escape_bounded(value: object, limit: int) -> str: + """Escape without cutting an HTML entity or overflowing Telegram.""" + + rendered: list[str] = [] + rendered_length = 0 + truncated = False + for character in str(value or ""): + encoded = html.escape(character) + if rendered_length + len(encoded) > limit - 3: + truncated = True + break + rendered.append(encoded) + rendered_length += len(encoded) + return "".join(rendered) + ("..." if truncated else "") + + +def _stage_status( + payload: Agent99TelegramLifecycleRequest, + field_name: str, +) -> str: + """Missing evidence stays visibly pending; it is never inferred green.""" + + return str(getattr(payload, field_name) or "pending") + + +def _pending_value(value: object | None) -> object: + return value if value not in (None, "") else "未提供 / PENDING" + + def _bounded_reconcile_receipt( payload: Agent99TelegramLifecycleRequest, *, @@ -49,28 +102,57 @@ def _bounded_reconcile_receipt( def _format_lifecycle_card(payload: Agent99TelegramLifecycleRequest) -> str: - def esc(value: object) -> str: - return html.escape(str(value or "")) + stage_lines = [ + ( + f"├ {stage_name}：" + f"{_STAGE_STATUS_LABELS[_stage_status(payload, field_name)]}" + ) + for stage_name, field_name in _LIFECYCLE_STAGES + ] return "\n".join( [ - f"Agent99 事件｜{esc(payload.title)}", + f"Agent99 事件｜{_escape_bounded(payload.title, 180)}", ( - f"狀態：{esc(_LIFECYCLE_LABELS[payload.lifecycle])}" - f"　等級：{esc(_SEVERITY_LABELS[payload.severity])}" + f"狀態：{_LIFECYCLE_LABELS[payload.lifecycle]}" + f"　等級：{_SEVERITY_LABELS[payload.severity]}" ), "────────────────────", - f"資產　{esc(payload.target)}", - f"影響　{esc(payload.impact)}", - f"AI 處置　{esc(payload.action)}", - f"驗證　{esc(payload.verification)}", + f"資產　{_escape_bounded(payload.target, 180)}", + f"影響　{_escape_bounded(payload.impact, 300)}", + f"AI 處置　{_escape_bounded(payload.action, 360)}", + f"驗證　{_escape_bounded(payload.verification, 300)}", + "────────────────────", + "AI Agent 執行鏈", + *stage_lines, + ( + "執行器：" + f"{_escape_bounded(_pending_value(payload.executor_name), 140)}　" + "Verifier：" + f"{_escape_bounded(_pending_value(payload.verifier_name), 140)}" + ), + "同一執行識別", + ( + "Run：" + f"{_escape_bounded(_pending_value(payload.run_id), 150)}" + ), + ( + "Trace：" + f"{_escape_bounded(_pending_value(payload.trace_id), 150)}" + ), + ( + "Work Item：" + f"{_escape_bounded(_pending_value(payload.work_item_id), 150)}" + ), "────────────────────", ( - f"事件：{esc(payload.incident_id)}　" - f"發生：{payload.occurrences} 次　耗時：{esc(payload.duration_text)}" + "事件：" + f"{_escape_bounded(payload.incident_id, 160)}　" + f"發生：{payload.occurrences} 次　" + f"耗時：{_escape_bounded(payload.duration_text, 80)}" ), - f"下一次更新：{esc(payload.next_update)}", - f"時間：{esc(payload.occurred_at.isoformat())}", + f"下一次更新：{_escape_bounded(payload.next_update, 300)}", + f"時間：{_escape_bounded(payload.occurred_at.isoformat(), 64)}", ] )[:4096] @@ -78,32 +160,44 @@ def _format_lifecycle_card(payload: Agent99TelegramLifecycleRequest) -> str: def _format_lifecycle_caption(payload: Agent99TelegramLifecycleRequest) -> str: """Keep the visual card caption scannable on mobile Telegram clients.""" - def esc(value: object, limit: int) -> str: - rendered: list[str] = [] - rendered_length = 0 - truncated = False - for character in str(value or ""): - encoded = html.escape(character) - if rendered_length + len(encoded) > limit - 3: - truncated = True - break - rendered.append(encoded) - rendered_length += len(encoded) - return "".join(rendered) + ("..." if truncated else "") + stage_summary = "｜".join( + ( + f"{stage_name}:" + f"{_STAGE_STATUS_COMPACT[_stage_status(payload, field_name)]}" + ) + for stage_name, field_name in _LIFECYCLE_STAGES + ) return "\n".join( [ ( - f"{esc(_LIFECYCLE_LABELS[payload.lifecycle], 48)}｜" - f"{esc(_SEVERITY_LABELS[payload.severity], 32)}" + f"{_escape_bounded(_LIFECYCLE_LABELS[payload.lifecycle], 48)}｜" + f"{_escape_bounded(_SEVERITY_LABELS[payload.severity], 32)}" ), - f"{esc(payload.title, 100)}", - f"資產：{esc(payload.target, 80)}", - f"Agent99：{esc(payload.action, 220)}", - f"驗證：{esc(payload.verification, 180)}", + f"{_escape_bounded(payload.title, 90)}", + f"資產：{_escape_bounded(payload.target, 64)}", + f"鏈路：{stage_summary}", + f"Agent99：{_escape_bounded(payload.action, 115)}", + f"驗證：{_escape_bounded(payload.verification, 90)}", ( - f"事件 {esc(payload.incident_id, 180)}｜" - f"發生 {payload.occurrences} 次｜{esc(payload.duration_text, 60)}" + "執行器：" + f"{_escape_bounded(_pending_value(payload.executor_name), 52)}｜" + "Verifier：" + f"{_escape_bounded(_pending_value(payload.verifier_name), 52)}" + ), + ( + "ID｜Run:" + f"{_escape_bounded(_pending_value(payload.run_id), 48)}｜" + "Trace:" + f"{_escape_bounded(_pending_value(payload.trace_id), 48)}｜" + "WI:" + f"{_escape_bounded(_pending_value(payload.work_item_id), 48)}" + ), + ( + "事件 " + f"{_escape_bounded(payload.incident_id, 80)}｜" + f"發生 {payload.occurrences} 次｜" + f"{_escape_bounded(payload.duration_text, 40)}" ), ] ) diff --git a/apps/api/src/services/security_interceptor.py b/apps/api/src/services/security_interceptor.py index 65280a149..f128df079 100644 --- a/apps/api/src/services/security_interceptor.py +++ b/apps/api/src/services/security_interceptor.py @@ -554,9 +554,13 @@ class TelegramSecurityInterceptor: # 2026-04-01 Claude Code (ADR-050): 支援 read-only info actions (2-part format) # 2026-04-20 P0.1 ogt + Claude Opus 4.7: drift_view_page 納入 INFO_ACTIONS # payload 格式: drift_view_page:{report_id}_{page}（底線分隔，不跟冒號衝突） - INFO_ACTIONS = {"detail", "reanalyze", "history", "drift_view_page"} + from src.services.telegram_button_registry import ( + registered_info_callback_actions, + ) + + info_actions = registered_info_callback_actions() parts = callback_data.split(":") - if len(parts) == 2 and parts[0] in INFO_ACTIONS: + if len(parts) == 2 and parts[0] in info_actions: return { "action": parts[0], "incident_id": parts[1], diff --git a/apps/api/src/services/telegram_button_registry.py b/apps/api/src/services/telegram_button_registry.py new file mode 100644 index 000000000..7b2171f28 --- /dev/null +++ b/apps/api/src/services/telegram_button_registry.py @@ -0,0 +1,222 @@ +"""Machine-readable Telegram inline-button contracts. + +Only callback actions that resolve to one explicit handler contract may be +rendered. Transport readiness is evaluated separately by TelegramGateway; +this registry prevents an enabled transport from reviving legacy ghost +buttons. +""" + +from __future__ import annotations + +import re +from dataclasses import asdict, dataclass + + +@dataclass(frozen=True) +class TelegramButtonActionContract: + action: str + callback_format: str + handler: str + risk: str + executor: str + verifier: str + state: str = "active" + + +_STATIC_ACTIONS: dict[str, TelegramButtonActionContract] = { + "detail": TelegramButtonActionContract( + "detail", "info", "TelegramGateway._send_incident_detail", "low", + "telegram_info_query", "telegram_callback_reply_receipt", + ), + "history": TelegramButtonActionContract( + "history", "info", "TelegramGateway._send_incident_history", "low", + "telegram_info_query", "telegram_callback_reply_receipt", + ), + "reanalyze": TelegramButtonActionContract( + "reanalyze", "info", "TelegramGateway._send_reanalyze_result", "low", + "telegram_reanalysis_query", "telegram_callback_reply_receipt", + ), + "drift_view_page": TelegramButtonActionContract( + "drift_view_page", "info", "TelegramGateway._send_drift_diff_detail", "low", + "telegram_info_query", "telegram_callback_reply_receipt", + ), + "approve": TelegramButtonActionContract( + "approve", "nonce", "TelegramGateway.handle_callback", "high", + "approval_control_plane", "approval_and_execution_receipt", + ), + "reject": TelegramButtonActionContract( + "reject", "nonce", "TelegramGateway.handle_callback", "medium", + "approval_control_plane", "approval_state_receipt", + ), + "silence": TelegramButtonActionContract( + "silence", "nonce", "TelegramGateway._handle_silence", "medium", + "telegram_silence_store", "telegram_callback_reply_receipt", + ), + "snooze": TelegramButtonActionContract( + "snooze", "nonce", "TelegramGateway._handle_snooze", "low", + "telegram_snooze_store", "telegram_callback_reply_receipt", + ), + "tune": TelegramButtonActionContract( + "tune", "nonce", "TelegramGateway._handle_auto_tuning", "low", + "shadow_candidate_recorder", "candidate_record_receipt", + state="shadow_candidate_only", + ), + "log_manual_fix": TelegramButtonActionContract( + "log_manual_fix", "nonce", "TelegramGateway.handle_callback:log_manual_fix", "low", + "incident_timeline", "timeline_write_receipt", + ), + "drift_view": TelegramButtonActionContract( + "drift_view", "nonce", "TelegramGateway._handle_drift_action", "low", + "drift_readback", "telegram_callback_reply_receipt", + ), + "drift_adopt": TelegramButtonActionContract( + "drift_adopt", "nonce", "TelegramGateway._handle_drift_action", "high", + "drift_control_plane", "drift_post_verifier", + ), + "drift_revert": TelegramButtonActionContract( + "drift_revert", "nonce", "TelegramGateway._handle_drift_action", "high", + "drift_control_plane", "drift_post_verifier", + ), + "la": TelegramButtonActionContract( + "la", "llm_short_id", "TelegramGateway._handle_llm_action_callback", "dynamic", + "callback_dispatcher", "callback_dispatch_receipt", + ), +} + +_AI_ADVISORY_ACTIONS = { + "ai_advisory_handled", + "ai_advisory_snooze", + "ai_advisory_view", + "ai_advisory_produce_cmd", +} +_LLM_SHORT_ID_RE = re.compile(r"^[0-9a-f]{16}$") + + +def _callback_action(callback_data: str) -> str: + return str(callback_data or "").split(":", 1)[0].strip() + + +def _registry_action_contract(action: str) -> TelegramButtonActionContract | None: + contract = _STATIC_ACTIONS.get(action) + if contract is not None: + return contract + if action in _AI_ADVISORY_ACTIONS: + return TelegramButtonActionContract( + action, + "advisory", + "TelegramGateway._handle_ai_advisory_action", + "low", + "ai_advisory_control_plane", + "telegram_callback_reply_receipt", + ) + + # The YAML registry is the existing source of truth for category actions. + # Import lazily so registry validation never creates a module cycle. + from src.services.callback_dispatcher import get_action_spec + + spec = get_action_spec(action) + if spec is None: + return None + return TelegramButtonActionContract( + action=spec.name, + callback_format=spec.callback_format, + handler="callback_dispatcher.dispatch_action", + risk=spec.risk, + executor=f"mcp:{spec.mcp_provider}/{spec.mcp_tool}", + verifier="callback_dispatch_receipt", + ) + + +def resolve_callback_action_contract( + callback_data: str, +) -> TelegramButtonActionContract | None: + """Resolve and validate a callback payload without persisting its nonce.""" + raw = str(callback_data or "").strip() + if not raw or len(raw.encode("utf-8")) > 64: + return None + parts = raw.split(":") + action = _callback_action(raw) + contract = _registry_action_contract(action) + if contract is None: + return None + + if contract.callback_format == "info": + return contract if len(parts) == 2 and bool(parts[1]) else None + if contract.callback_format == "nonce": + return contract if len(parts) in {4, 5} and all(parts[1:4]) else None + if contract.callback_format == "llm_short_id": + return ( + contract + if len(parts) == 2 and _LLM_SHORT_ID_RE.fullmatch(parts[1]) + else None + ) + if contract.callback_format == "advisory": + return contract if len(parts) == 3 and all(parts[1:]) else None + return None + + +def callback_action_is_registered(callback_data: str) -> bool: + return resolve_callback_action_contract(callback_data) is not None + + +def registered_info_callback_actions() -> frozenset[str]: + """Return every two-part callback action accepted by the dispatcher.""" + from src.services.callback_dispatcher import load_action_registry + + names = { + action + for action, contract in _STATIC_ACTIONS.items() + if contract.callback_format == "info" + } + names.update( + spec.name + for spec in load_action_registry().values() + if spec.callback_format == "info" + ) + return frozenset(names) + + +def telegram_button_registry_snapshot() -> dict[str, object]: + """Return a redaction-safe registry readback for health and tests.""" + from src.services.callback_dispatcher import load_action_registry + + contracts = list(_STATIC_ACTIONS.values()) + contracts.extend( + TelegramButtonActionContract( + action=action, + callback_format="advisory", + handler="TelegramGateway._handle_ai_advisory_action", + risk="low", + executor="ai_advisory_control_plane", + verifier="telegram_callback_reply_receipt", + ) + for action in sorted(_AI_ADVISORY_ACTIONS) + ) + contracts.extend( + TelegramButtonActionContract( + action=spec.name, + callback_format=spec.callback_format, + handler="callback_dispatcher.dispatch_action", + risk=spec.risk, + executor=f"mcp:{spec.mcp_provider}/{spec.mcp_tool}", + verifier="callback_dispatch_receipt", + ) + for spec in load_action_registry().values() + if spec.name not in _STATIC_ACTIONS + ) + rows = sorted((asdict(item) for item in contracts), key=lambda item: item["action"]) + return { + "schema_version": "telegram_button_action_registry_v1", + "registered_action_count": len(rows), + "unknown_action_policy": "do_not_render", + "actions": rows, + } + + +__all__ = [ + "TelegramButtonActionContract", + "callback_action_is_registered", + "registered_info_callback_actions", + "resolve_callback_action_contract", + "telegram_button_registry_snapshot", +] diff --git a/apps/api/src/services/telegram_gateway.py b/apps/api/src/services/telegram_gateway.py index 7b4191aa5..faad17160 100644 --- a/apps/api/src/services/telegram_gateway.py +++ b/apps/api/src/services/telegram_gateway.py @@ -32,6 +32,7 @@ import secrets from collections.abc import Mapping from dataclasses import dataclass, field from datetime import UTC, datetime +from urllib.parse import urlsplit from uuid import NAMESPACE_URL, UUID, uuid5 import httpx @@ -75,6 +76,7 @@ from src.services.security_interceptor import ( UserNotWhitelistedError, get_security_interceptor, ) +from src.services.telegram_button_registry import callback_action_is_registered # ============================================================================= # Snooze/Silence Redis Keys (2026-03-27 P1 優化) @@ -133,6 +135,7 @@ POLLING_LEADER_KEY = "telegram:polling:leader" POLLING_LEADER_TTL = 45 # seconds - Pod 宕掉後 45s 自動轉移 POLLING_LEADER_RENEW = 20 # seconds - 每 20s 續約 POLLING_LEADER_WATCH = 30 # seconds - 非 Leader Pod 每 30s 嘗試接管 +CALLBACK_INGRESS_RECEIPT_MAX_AGE_SECONDS = 24 * 60 * 60 logger = structlog.get_logger(__name__) _TELEGRAM_BOT_URL_RE = re.compile(r"(api\.telegram\.org/bot)[^/\s]+") @@ -1024,6 +1027,65 @@ def normalize_telegram_send_message_payload(method: str, payload: dict) -> dict: return normalized_payload +def _verified_telegram_button_url(value: object) -> bool: + """Accept only credential-free HTTPS links to the canonical AwoooP UI.""" + raw = str(value or "").strip() + if not raw: + return False + parsed = urlsplit(raw) + canonical = urlsplit(AWOOOP_WEB_BASE_URL) + return bool( + parsed.scheme == "https" + and parsed.hostname + and canonical.hostname + and parsed.hostname == canonical.hostname + and parsed.port == canonical.port + and parsed.username is None + and parsed.password is None + ) + + +def filter_telegram_reply_markup( + reply_markup: object, + *, + interactive_buttons_ready: bool, +) -> dict | None: + """Fail closed when callback ingress or an action contract is missing. + + URL buttons remain available because they do not depend on Telegram + callback transport. Unknown callbacks are removed even when the transport + is healthy, preventing legacy builders from creating ghost controls. + """ + if not isinstance(reply_markup, dict): + return None + inline_keyboard = reply_markup.get("inline_keyboard") + if not isinstance(inline_keyboard, list): + return dict(reply_markup) + + safe_rows: list[list[dict]] = [] + for row in inline_keyboard: + if not isinstance(row, list): + continue + safe_row: list[dict] = [] + for button in row: + if not isinstance(button, dict): + continue + text = str(button.get("text") or "").strip() + callback_data = str(button.get("callback_data") or "").strip() + url = str(button.get("url") or "").strip() + if ( + interactive_buttons_ready + and callback_data + and callback_action_is_registered(callback_data) + ): + safe_row.append({"text": text, "callback_data": callback_data}) + elif not callback_data and _verified_telegram_button_url(url): + safe_row.append({"text": text, "url": url}) + if safe_row: + safe_rows.append(safe_row) + return {"inline_keyboard": safe_rows} if safe_rows else None + + def _telegram_send_delivery_succeeded(result: object) -> bool: """Require affirmative provider delivery evidence from ``_send_request``. @@ -4604,7 +4666,7 @@ class SentryErrorMessage: Sentry 錯誤訊息 (SENTRY_ERROR) 2026-03-29 ogt: 新增，用於 Sentry 錯誤通知 - 按鈕: [🔍 查看詳情] [🔕 靜默 1h] + 按鈕: [🔕 靜默 1h]；Sentry URL 由訊息中的 HTTPS link 提供 """ error_id: str # Sentry Issue ID error_type: str # TypeError, ValueError, etc. @@ -4664,7 +4726,7 @@ class ResourceWarnMessage: 資源告警訊息 (RESOURCE_WARN) 2026-03-29 ogt: 新增，用於資源耗盡警告 - 按鈕: [⚡ 自動擴展] [🔕 靜默 1h] + 按鈕: [🔕 靜默 1h]；未綁 executor/verifier 前不渲染擴展操作 """ resource_id: str # RES-YYYYMMDD-XXXX pod_name: str # Pod 名稱 @@ -5493,6 +5555,8 @@ class TelegramGateway: self._polling_active = False self._polling_task: asyncio.Task | None = None self._last_update_id = 0 + self._last_callback_ingress_receipt_at: datetime | None = None + self._last_callback_ingress_source = "" # 2026-04-01 Claude Code: 分散式 Leader Election (防 2-Pod 409 互搶) self._pod_id = os.environ.get("POD_NAME", os.urandom(8).hex()) self._leader_task: asyncio.Task | None = None @@ -5545,6 +5609,116 @@ class TelegramGateway: """Runtime binding; new monitoring sends are still policy-gated.""" return settings.SRE_GROUP_CHAT_ID + def record_callback_ingress_receipt( + self, + *, + source: str, + update_id: int | None, + ) -> None: + """Record one process-local callback receipt without callback payload.""" + self._last_callback_ingress_receipt_at = datetime.now(UTC) + self._last_callback_ingress_source = str(source or "unknown")[:80] + if isinstance(update_id, int): + self._last_update_id = max(int(self._last_update_id or 0), update_id) + + def callback_ingress_status(self) -> dict[str, object]: + """Return callback transport truth used by health and render gates.""" + now = datetime.now(UTC) + last_receipt = getattr(self, "_last_callback_ingress_receipt_at", None) + source = str(getattr(self, "_last_callback_ingress_source", "") or "") + age_seconds: int | None = None + if isinstance(last_receipt, datetime): + normalized = ( + last_receipt.replace(tzinfo=UTC) + if last_receipt.tzinfo is None + else last_receipt.astimezone(UTC) + ) + age_seconds = max(0, int((now - normalized).total_seconds())) + receipt_fresh = bool( + age_seconds is not None + and age_seconds <= CALLBACK_INGRESS_RECEIPT_MAX_AGE_SECONDS + ) + polling_declared_active = bool(getattr(self, "_polling_active", False)) + polling_task = getattr(self, "_polling_task", None) + polling_task_active = bool( + polling_task is not None + and hasattr(polling_task, "done") + and not polling_task.done() + ) + polling_active = bool(polling_declared_active and polling_task_active) + if polling_active: + mode = "api_long_polling" + owner = "awoooi_api" + elif receipt_fresh: + mode = "external_callback_forward" + owner = source or "external_forwarder" + elif polling_declared_active: + mode = "api_long_polling_stale" + owner = "awoooi_api" + elif bool(settings.TELEGRAM_ENABLE_POLLING): + mode = "api_long_polling_inactive" + owner = "awoooi_api" + else: + mode = "external_callback_forward_unverified" + owner = "external_forwarder_unverified" + configured = bool(self.bot_token and self.chat_id) + interactive_ready = bool(configured and (polling_active or receipt_fresh)) + return { + "schema_version": "telegram_callback_ingress_status_v1", + "mode": mode, + "owner": owner, + "configured": configured, + "polling_active": polling_active, + "polling_declared_active": polling_declared_active, + "polling_task_active": polling_task_active, + "receipt_available": isinstance(last_receipt, datetime), + "receipt_fresh": receipt_fresh, + "receipt_source": source or None, + "last_receipt_at": ( + last_receipt.astimezone(UTC).isoformat() + if isinstance(last_receipt, datetime) and last_receipt.tzinfo + else last_receipt.replace(tzinfo=UTC).isoformat() + if isinstance(last_receipt, datetime) + else None + ), + "receipt_age_seconds": age_seconds, + "receipt_max_age_seconds": CALLBACK_INGRESS_RECEIPT_MAX_AGE_SECONDS, + "stale": bool(isinstance(last_receipt, datetime) and not receipt_fresh), + "interactive_buttons_ready": interactive_ready, + "fail_closed_reason": ( + None if interactive_ready else "callback_ingress_not_verified" + ), + } + + def _apply_callback_ingress_render_gate(self, payload: dict) -> dict: + reply_markup = payload.get("reply_markup") + if not isinstance(reply_markup, dict): + return payload + ingress = self.callback_ingress_status() + filtered = filter_telegram_reply_markup( + reply_markup, + interactive_buttons_ready=bool( + ingress.get("interactive_buttons_ready") is True + ), + ) + if filtered == reply_markup: + return payload + guarded = dict(payload) + if filtered is None: + guarded.pop("reply_markup", None) + else: + guarded["reply_markup"] = filtered + logger.warning( + "telegram_callback_button_render_fail_closed", + callback_ingress_mode=ingress.get("mode"), + interactive_buttons_ready=ingress.get("interactive_buttons_ready"), + original_button_count=_reply_markup_summary( + {"reply_markup": reply_markup} + ).get("button_count"), + rendered_button_count=_reply_markup_summary(guarded).get("button_count"), + ) + return guarded + @staticmethod def _blocked_route_receipt( classifier: str, @@ -6605,6 +6779,7 @@ class TelegramGateway: raise TelegramGatewayError("HTTP client not initialized") payload = normalize_telegram_send_message_payload(method, payload) + payload = self._apply_callback_ingress_render_gate(payload) source_envelope_extra = payload.pop(_AWOOOP_SOURCE_ENVELOPE_EXTRA_KEY, None) if canonical_route_receipt is not None: source_envelope_extra = dict(source_envelope_extra or {}) @@ -7233,7 +7408,7 @@ class TelegramGateway: # 自動調優按鈕 (v7.0) if include_auto_tuning and auto_tuning_command: tuning_nonce = self._security.generate_callback_nonce(approval_id, "tune") - buttons.append([{"text": "⚡ 執行自動調優", "callback_data": tuning_nonce}]) + buttons.append([{"text": "📝 記錄調優候選", "callback_data": tuning_nonce}]) awooop_row = _awooop_truth_chain_button_row(incident_id) if awooop_row: @@ -7318,7 +7493,7 @@ class TelegramGateway: if include_auto_tuning and auto_tuning_command: tuning_nonce = self._security.generate_callback_nonce(approval_id, "tune") buttons.append([ - {"text": "⚡ 執行自動調優", "callback_data": tuning_nonce} + {"text": "📝 記錄調優候選", "callback_data": tuning_nonce} ]) return {"inline_keyboard": buttons} @@ -8741,13 +8916,11 @@ class TelegramGateway: def _build_sentry_keyboard(self, error_id: str) -> dict: """建立 Sentry 錯誤訊息按鈕""" - view_nonce = self._security.generate_callback_nonce(error_id, "view") silence_nonce = self._security.generate_callback_nonce(error_id, "silence") return { "inline_keyboard": [ [ - {"text": "🔍 查看詳情", "callback_data": view_nonce}, {"text": "🔕 靜默 1h", "callback_data": silence_nonce}, ] ] @@ -8755,13 +8928,11 @@ class TelegramGateway: def _build_resource_keyboard(self, resource_id: str) -> dict: """建立資源告警按鈕""" - scale_nonce = self._security.generate_callback_nonce(resource_id, "scale") silence_nonce = self._security.generate_callback_nonce(resource_id, "silence") return { "inline_keyboard": [ [ - {"text": "⚡ 自動擴展", "callback_data": scale_nonce}, {"text": "🔕 靜默 1h", "callback_data": silence_nonce}, ] ] @@ -9593,7 +9764,7 @@ class TelegramGateway: await self._answer_callback( callback_query_id, "tune", - text="⚡ 調優指令已記錄 (Shadow Mode)", + text="📝 調優候選已記錄 (Shadow Mode)", ) # 更新訊息 await self._update_message_after_action( @@ -10448,7 +10619,7 @@ class TelegramGateway: elif action == "reject": stamp = f"❌ 已由 @{username} 拒絕執行" elif action == "tune": - stamp = f"⚡ 已由 @{username} 觸發自動調優 (Shadow Mode)" + stamp = f"📝 已由 @{username} 記錄調優候選 (Shadow Mode)" if extra_info: stamp += "\n📝 指令已記錄" elif action == "snooze": @@ -12902,7 +13073,13 @@ class TelegramGateway: elif message: await self._handle_chat_message(update_id, message) - async def _handle_callback_query(self, update_id: int, callback_query: dict) -> None: + async def _handle_callback_query( + self, + update_id: int, + callback_query: dict, + *, + ingress_source: str = "api_long_polling", + ) -> None: """處理按鈕點擊更新""" callback_query_id = callback_query.get("id") callback_data = callback_query.get("data") @@ -12913,6 +13090,11 @@ class TelegramGateway: logger.warning("telegram_callback_invalid", update_id=update_id) return + self.record_callback_ingress_receipt( + source=ingress_source, + update_id=update_id, + ) + username = user.get("username") or user.get("first_name") or str(user_id) original_text = callback_query.get("message", {}).get("text", "") message = callback_query.get("message", {}) @@ -12947,7 +13129,7 @@ class TelegramGateway: chat_id=chat_id, ) - if result.get("success"): + if result.get("success") and not result.get("info_action"): # 執行資料庫更新 (簽核/拒絕) await self._execute_approval_action( action=result["action"], diff --git a/apps/api/tests/test_agent99_telegram_lifecycle_api.py b/apps/api/tests/test_agent99_telegram_lifecycle_api.py index dfdbbd5c4..461530ca4 100644 --- a/apps/api/tests/test_agent99_telegram_lifecycle_api.py +++ b/apps/api/tests/test_agent99_telegram_lifecycle_api.py @@ -34,6 +34,15 @@ def payload() -> dict: "lifecycle": "investigating", "state_key": "investigating|warning", "run_id": "run-agent99-20260715-01", + "trace_id": "trace-agent99-20260715-01", + "work_item_id": "work-item-agent99-20260715-01", + "executor_name": "Agent99:Recover", + "verifier_name": "agent99-recover-post-condition-v1", + "received_status": "passed", + "check_status": "passed", + "apply_status": "running", + "verifier_status": "pending", + "closure_status": "pending", "title": "Host 110 主機效能異常", "target": "192.168.0.110", "impact": "資源壓力可能降低服務回應速度。", @@ -90,6 +99,28 @@ def test_agent99_client_retries_one_202_with_exact_db_only_receipt() -> None: assert "reconcileProviderSendPerformed" in send_section assert "provider_message_id" in send_section assert "destination_binding" in send_section + for lifecycle_field in ( + "trace_id", + "work_item_id", + "executor_name", + "verifier_name", + "received_status", + "check_status", + "apply_status", + "verifier_status", + "closure_status", + ): + assert f"{lifecycle_field} =" in send_section + for card_field in ( + "receivedStatus", + "checkStatus", + "applyStatus", + "verifierStatus", + "closureStatus", + "traceKey", + "workItemKey", + ): + assert f"$card.{card_field}" in send_section assert "X-Agent99-Lifecycle-Token" not in send_section selftest_section = source.split( "function Invoke-AgentTelegramDeliverySelfTest {", @@ -137,6 +168,72 @@ def test_lifecycle_payload_rejects_unbounded_content() -> None: }, } ) + with pytest.raises(ValidationError): + Agent99TelegramLifecycleRequest.model_validate( + {**payload(), "closure_status": "resolved_without_receipt"} + ) + + +def test_lifecycle_card_exposes_agent99_stage_and_identity_receipts() -> None: + request = Agent99TelegramLifecycleRequest.model_validate( + { + **payload(), + "check_status": "failed", + "apply_status": "passed", + "verifier_status": "failed", + "closure_status": "failed", + } + ) + + card = lifecycle_service._format_lifecycle_card(request) + caption = lifecycle_service._format_lifecycle_caption(request) + + assert "AI Agent 執行鏈" in card + assert "Received：完成 / PASSED" in card + assert "Check：失敗 / FAILED" in card + assert "Apply：完成 / PASSED" in card + assert "Verifier：失敗 / FAILED" in card + assert "Closure：失敗 / FAILED" in card + assert "run-agent99-20260715-01" in card + assert "trace-agent99-20260715-01" in card + assert "work-item-agent99-20260715-01" in card + assert "Agent99:Recover" in card + assert "agent99-recover-post-condition-v1" in card + assert "Received:PASSED" in caption + assert "Check:FAILED" in caption + assert "Closure:FAILED" in caption + assert "Trace:trace-agent99-20260715-01" in caption + + +def test_lifecycle_card_missing_stage_evidence_stays_visibly_pending() -> None: + optional_receipt_fields = { + "run_id", + "trace_id", + "work_item_id", + "executor_name", + "verifier_name", + "received_status", + "check_status", + "apply_status", + "verifier_status", + "closure_status", + } + request = Agent99TelegramLifecycleRequest.model_validate( + { + key: value + for key, value in payload().items() + if key not in optional_receipt_fields + } + ) + + card = lifecycle_service._format_lifecycle_card(request) + caption = lifecycle_service._format_lifecycle_caption(request) + + assert card.count("待回報 / PENDING") >= 5 + assert card.count("未提供 / PENDING") >= 5 + assert "Received:PENDING" in caption + assert "Closure:PENDING" in caption + assert "ID｜Run:未提供 / PENDING" in caption def test_visual_caption_stays_within_telegram_limit_without_broken_html() -> None: diff --git a/apps/api/tests/test_telegram_button_consistency.py b/apps/api/tests/test_telegram_button_consistency.py index 6e0931b00..12d4e12b0 100644 --- a/apps/api/tests/test_telegram_button_consistency.py +++ b/apps/api/tests/test_telegram_button_consistency.py @@ -14,6 +14,8 @@ Telegram 按鈕一致性測試 import re +from src.services.telegram_button_registry import registered_info_callback_actions + def _read_gateway() -> str: with open("src/services/telegram_gateway.py", encoding="utf-8") as f: @@ -229,16 +231,14 @@ class TestCallbackHandlerCompleteness: """驗證 detail/history 按鈕有對應的 handler（鬼魂按鈕鐵律三條件之二）""" def test_detail_action_in_info_actions_whitelist(self): - """detail action 在 INFO_ACTIONS 白名單中""" + """detail action 在 machine-readable info registry 中""" source = _read_security_interceptor() - assert '"detail"' in source, "detail 必須在 INFO_ACTIONS 白名單" - # 確認是在 INFO_ACTIONS 集合定義中 - assert "INFO_ACTIONS" in source + assert "registered_info_callback_actions" in source + assert "detail" in registered_info_callback_actions() def test_history_action_in_info_actions_whitelist(self): - """history action 在 INFO_ACTIONS 白名單中""" - source = _read_security_interceptor() - assert '"history"' in source, "history 必須在 INFO_ACTIONS 白名單" + """history action 在 machine-readable info registry 中""" + assert "history" in registered_info_callback_actions() def test_detail_handler_exists_in_handle_callback(self): """handle_callback 有 detail handler 分支""" diff --git a/apps/api/tests/test_telegram_callback_ingress_contract.py b/apps/api/tests/test_telegram_callback_ingress_contract.py new file mode 100644 index 000000000..fc4b7ece3 --- /dev/null +++ b/apps/api/tests/test_telegram_callback_ingress_contract.py @@ -0,0 +1,366 @@ +from __future__ import annotations + +from datetime import UTC, datetime, timedelta +from types import SimpleNamespace + +import pytest +from fastapi import HTTPException + +from src.api.v1 import telegram as telegram_api +from src.api.v1 import telegram_webhook as telegram_webhook_api +from src.services import telegram_gateway as telegram_gateway_module +from src.services.callback_dispatcher import load_action_registry +from src.services.telegram_button_registry import ( + callback_action_is_registered, + registered_info_callback_actions, + resolve_callback_action_contract, + telegram_button_registry_snapshot, +) +from src.services.telegram_gateway import ( + CALLBACK_INGRESS_RECEIPT_MAX_AGE_SECONDS, + TelegramGateway, + filter_telegram_reply_markup, +) + + +class _ConfiguredGateway(TelegramGateway): + @property + def bot_token(self) -> str: + return "test-configured" + + @property + def chat_id(self) -> str: + return "test-configured" + + +class _Security: + @staticmethod + def generate_callback_nonce(approval_id: str, action: str) -> str: + return f"{action}:{approval_id}:1:nonce" + + +def _callback_rows(markup: dict | None) -> list[dict]: + return [ + button + for row in (markup or {}).get("inline_keyboard", []) + for button in row + ] + + +def test_callback_ingress_defaults_to_unverified_fail_closed() -> None: + gateway = _ConfiguredGateway() + + status = gateway.callback_ingress_status() + + assert status["mode"] == "external_callback_forward_unverified" + assert status["polling_active"] is False + assert status["receipt_available"] is False + assert status["interactive_buttons_ready"] is False + assert status["fail_closed_reason"] == "callback_ingress_not_verified" + + +def test_callback_ingress_receipt_has_24_hour_freshness_contract() -> None: + gateway = _ConfiguredGateway() + assert CALLBACK_INGRESS_RECEIPT_MAX_AGE_SECONDS == 24 * 60 * 60 + + gateway._last_callback_ingress_source = "canonical_webhook" + gateway._last_callback_ingress_receipt_at = datetime.now(UTC) - timedelta(hours=23) + fresh = gateway.callback_ingress_status() + assert fresh["receipt_fresh"] is True + assert fresh["interactive_buttons_ready"] is True + + gateway._last_callback_ingress_receipt_at = datetime.now(UTC) - timedelta(hours=25) + stale = gateway.callback_ingress_status() + assert stale["stale"] is True + assert stale["interactive_buttons_ready"] is False + + +def test_polling_flag_without_live_task_is_stale_not_ready() -> None: + gateway = _ConfiguredGateway() + gateway._polling_active = True + gateway._polling_task = None + + status = gateway.callback_ingress_status() + + assert status["mode"] == "api_long_polling_stale" + assert status["polling_declared_active"] is True + assert status["polling_task_active"] is False + assert status["polling_active"] is False + assert status["interactive_buttons_ready"] is False + + +def test_fail_closed_markup_keeps_only_verified_https_urls() -> None: + markup = { + "inline_keyboard": [[ + {"text": "詳情", "callback_data": "detail:INC-20260716-TEST"}, + {"text": "Runs", "url": "https://awoooi.wooo.work/zh-TW/awooop/runs"}, + {"text": "Unsafe", "url": "http://192.0.2.1/internal"}, + {"text": "External", "url": "https://example.com/not-canonical"}, + ]] + } + + filtered = filter_telegram_reply_markup( + markup, + interactive_buttons_ready=False, + ) + + assert _callback_rows(filtered) == [{ + "text": "Runs", + "url": "https://awoooi.wooo.work/zh-TW/awooop/runs", + }] + + +def test_gateway_render_gate_uses_live_ingress_status() -> None: + gateway = _ConfiguredGateway() + payload = { + "text": "incident", + "reply_markup": { + "inline_keyboard": [[ + {"text": "詳情", "callback_data": "detail:INC-20260716-TEST"}, + {"text": "Runs", "url": "https://awoooi.wooo.work/zh-TW/awooop/runs"}, + ]] + }, + } + + guarded = gateway._apply_callback_ingress_render_gate(payload) + + assert _callback_rows(guarded["reply_markup"]) == [{ + "text": "Runs", + "url": "https://awoooi.wooo.work/zh-TW/awooop/runs", + }] + + +def test_ready_markup_still_removes_unknown_ghost_callbacks() -> None: + markup = { + "inline_keyboard": [[ + {"text": "詳情", "callback_data": "detail:INC-20260716-TEST"}, + {"text": "Ghost view", "callback_data": "view:ERR:1:nonce"}, + {"text": "Ghost scale", "callback_data": "scale:RES:1:nonce"}, + {"text": "Runs", "url": "https://awoooi.wooo.work/zh-TW/awooop/runs"}, + ]] + } + + filtered = filter_telegram_reply_markup( + markup, + interactive_buttons_ready=True, + ) + rendered = _callback_rows(filtered) + + assert {button.get("callback_data") for button in rendered} == { + "detail:INC-20260716-TEST", + None, + } + assert all("view:" not in str(button) for button in rendered) + assert all("scale:" not in str(button) for button in rendered) + + +def test_machine_registry_excludes_ghosts_and_describes_shadow_tune() -> None: + assert callback_action_is_registered("detail:INC-20260716-TEST") is True + assert callback_action_is_registered("view:ERR:1:nonce") is False + assert callback_action_is_registered("scale:RES:1:nonce") is False + tune = resolve_callback_action_contract("tune:APR-1:1:nonce") + assert tune is not None + assert tune.state == "shadow_candidate_only" + + snapshot = telegram_button_registry_snapshot() + action_names = {row["action"] for row in snapshot["actions"]} + assert snapshot["schema_version"] == "telegram_button_action_registry_v1" + assert snapshot["unknown_action_policy"] == "do_not_render" + assert "view" not in action_names + assert "scale" not in action_names + + +def test_every_yaml_action_has_a_renderable_registry_contract() -> None: + registry = load_action_registry() + assert registry + + for action, spec in registry.items(): + callback_data = ( + f"{action}:APR-1:1:nonce" + if spec.callback_format == "nonce" + else f"{action}:INC-20260716-TEST" + ) + contract = resolve_callback_action_contract(callback_data) + assert contract is not None, action + assert contract.handler == "callback_dispatcher.dispatch_action" + assert contract.executor.startswith("mcp:") + + +def test_security_parser_accepts_every_registered_info_action() -> None: + gateway = _ConfiguredGateway() + info_actions = registered_info_callback_actions() + assert {"detail", "history", "reanalyze", "drift_view_page"} <= info_actions + + for action in info_actions: + parsed = gateway._security.parse_callback_data( + f"{action}:INC-20260716-TEST" + ) + assert parsed["action"] == action + assert parsed["is_info_action"] is True + + +def test_legacy_builders_no_longer_emit_view_or_scale() -> None: + gateway = _ConfiguredGateway() + gateway._security = _Security() + + sentry = _callback_rows(gateway._build_sentry_keyboard("ERR-1")) + resource = _callback_rows(gateway._build_resource_keyboard("RES-1")) + + assert [button["callback_data"].split(":", 1)[0] for button in sentry] == [ + "silence" + ] + assert [button["callback_data"].split(":", 1)[0] for button in resource] == [ + "silence" + ] + + +@pytest.mark.asyncio +async def test_tune_button_truthfully_says_it_records_a_candidate() -> None: + gateway = _ConfiguredGateway() + gateway._security = _Security() + + markup = await gateway._build_inline_keyboard( + approval_id="APR-1", + incident_id="INC-20260716-TEST", + include_auto_tuning=True, + auto_tuning_command="candidate-only", + ) + labels = {button["text"] for button in _callback_rows(markup)} + + assert "📝 記錄調優候選" in labels + assert "⚡ 執行自動調優" not in labels + + +class _InfoCallbackGateway(_ConfiguredGateway): + def __init__(self) -> None: + super().__init__() + self.approval_execution_called = False + + async def mirror_callback_query_received(self, **_kwargs) -> str: + return "receipt-1" + + async def handle_callback(self, **_kwargs) -> dict: + return { + "success": True, + "info_action": True, + "action": "detail", + "approval_id": "INC-20260716-TEST", + } + + async def _execute_approval_action(self, **_kwargs) -> None: + self.approval_execution_called = True + + +@pytest.mark.asyncio +async def test_polling_info_action_does_not_execute_approval_stage() -> None: + gateway = _InfoCallbackGateway() + + await gateway._handle_callback_query( + 42, + { + "id": "callback-1", + "data": "detail:INC-20260716-TEST", + "from": {"id": 7, "username": "operator"}, + "message": { + "message_id": 99, + "text": "incident", + "chat": {"id": -1001}, + }, + }, + ) + + assert gateway.approval_execution_called is False + assert gateway.callback_ingress_status()["receipt_source"] == "api_long_polling" + + +def test_webhook_route_has_one_canonical_owner() -> None: + route_paths = [route.path for route in telegram_api.router.routes] + route_paths.extend(route.path for route in telegram_webhook_api.router.routes) + + assert route_paths.count("/telegram/webhook") == 1 + + +def test_production_webhook_without_secret_fails_closed(monkeypatch) -> None: + monkeypatch.setattr( + telegram_webhook_api, + "settings", + SimpleNamespace(ENVIRONMENT="prod", TELEGRAM_WEBHOOK_SECRET=""), + ) + + with pytest.raises(HTTPException) as exc_info: + telegram_webhook_api._verify_secret_token(None) + + assert exc_info.value.status_code == 503 + + +class _CallbackRequest: + async def json(self) -> dict: + return { + "update_id": 77, + "callback_query": { + "id": "callback-77", + "data": "detail:INC-20260716-TEST", + "from": {"id": 7, "username": "operator"}, + "message": {"message_id": 9, "chat": {"id": -1001}}, + }, + } + + +class _CanonicalWebhookGateway: + def __init__(self) -> None: + self.receipts: list[dict] = [] + + async def _handle_callback_query( + self, + update_id: int, + callback_query: dict, + *, + ingress_source: str, + ) -> None: + self.receipts.append({ + "update_id": update_id, + "callback_query": callback_query, + "ingress_source": ingress_source, + }) + + +@pytest.mark.asyncio +async def test_canonical_webhook_forwards_callback_to_gateway_handler(monkeypatch) -> None: + gateway = _CanonicalWebhookGateway() + monkeypatch.setattr( + telegram_gateway_module, + "get_telegram_gateway", + lambda: gateway, + ) + + result = await telegram_webhook_api.telegram_webhook(_CallbackRequest()) + + assert result == {"ok": True} + assert len(gateway.receipts) == 1 + assert gateway.receipts[0]["update_id"] == 77 + assert gateway.receipts[0]["ingress_source"] == "canonical_webhook" + + +@pytest.mark.asyncio +async def test_health_is_degraded_when_callback_ingress_is_unverified(monkeypatch) -> None: + gateway = _ConfiguredGateway() + monkeypatch.setattr(telegram_api, "get_telegram_gateway", lambda: gateway) + monkeypatch.setattr( + telegram_api, + "settings", + SimpleNamespace( + OPENCLAW_TG_BOT_TOKEN="test-configured", + SRE_GROUP_CHAT_ID="test-configured", + OPENCLAW_TG_USER_WHITELIST=[], + ENVIRONMENT="test", + ), + ) + + health = await telegram_api.telegram_health() + + assert health["status"] == "degraded_callback_ingress_unverified" + assert health["mode"] == "external_callback_forward_unverified" + assert health["interactive_buttons_ready"] is False + assert health["callback_ingress"]["fail_closed_reason"] == ( + "callback_ingress_not_verified" + )