From 1a9497869bcd5fc8ad1611bbd45d165d89ce5b16 Mon Sep 17 00:00:00 2001 From: ogt Date: Wed, 15 Jul 2026 08:06:42 +0800 Subject: [PATCH] fix(agent99): route lifecycle alerts through canonical gateway --- agent99-control-plane.ps1 | 210 +++++++++++++++--- agent99-deploy.ps1 | 4 + agent99.config.99.example.json | 5 + apps/api/src/api/v1/agents.py | 60 ++++- apps/api/src/models/agent99_completion.py | 51 ++++- .../services/agent99_telegram_lifecycle.py | 114 ++++++++++ apps/api/src/services/telegram_gateway.py | 110 ++++++++- .../test_agent99_telegram_lifecycle_api.py | 208 +++++++++++++++++ ...nt99_transport_recovery_deploy_contract.py | 22 +- ...t_telegram_notification_egress_scanners.py | 7 +- 10 files changed, 736 insertions(+), 55 deletions(-) create mode 100644 apps/api/src/services/agent99_telegram_lifecycle.py create mode 100644 apps/api/tests/test_agent99_telegram_lifecycle_api.py diff --git a/agent99-control-plane.ps1 b/agent99-control-plane.ps1 index 69d52d61a..e16a0a0f7 100644 --- a/agent99-control-plane.ps1 +++ b/agent99-control-plane.ps1 @@ -1697,6 +1697,20 @@ function Invoke-AgentSensorGateSelfTest { exit 0 } +function Invoke-AgentCanonicalLifecycleIngress { + param( + [string]$Url, + [string]$Token, + [object]$Payload, + [int]$TimeoutSeconds + ) + + Invoke-RestMethod -Method Post -Uri $Url -Headers @{ + "X-Agent99-Lifecycle-Token" = $Token + "X-Project-ID" = "awoooi" + } -ContentType "application/json; charset=utf-8" -Body ($Payload | ConvertTo-Json -Depth 8 -Compress) -TimeoutSec $TimeoutSeconds +} + function Send-AgentTelegram { param( [string]$Severity, @@ -1707,6 +1721,10 @@ function Send-AgentTelegram { $telegram = $Config.telegram $enabled = ($telegram -and $telegram.enabled) + $gateway = if ($telegram -and $telegram.PSObject.Properties["canonicalGateway"]) { $telegram.canonicalGateway } else { $null } + $gatewayUrl = if ($gateway -and $gateway.PSObject.Properties["url"]) { [string]$gateway.url } else { "" } + $tokenEnv = if ($gateway -and $gateway.PSObject.Properties["tokenEnv"]) { [string]$gateway.tokenEnv } else { "AGENT99_SRE_RELAY_TOKEN" } + $timeoutSeconds = if ($gateway -and $gateway.PSObject.Properties["timeoutSeconds"]) { [math]::Min(30, [math]::Max(5, [int]$gateway.timeoutSeconds)) } else { 20 } $chatIdOverride = if ($Data -and $Data.PSObject.Properties["replyChatId"]) { [string]$Data.replyChatId } else { "" } $card = Get-AgentIncidentCardModel $Severity $EventType $Message $Data $incidentState = Get-AgentTelegramIncidentState $card @@ -1714,31 +1732,37 @@ function Send-AgentTelegram { $storedRootMessageId = if ($incidentState.value -and $incidentState.value.PSObject.Properties["rootMessageId"]) { [string]$incidentState.value.rootMessageId } else { "" } $replyToMessageId = if ($explicitReplyMessageId -match "^\d+$") { $explicitReplyMessageId } elseif ($storedRootMessageId -match "^\d+$") { $storedRootMessageId } else { "" } - $attempt = [pscustomobject]@{ - timestamp = (Get-Date -Format o) - enabled = [bool]$enabled - eventType = $EventType - severity = $Severity - messageFormat = "incident_card_zh_tw_v5" - incidentId = $card.incidentId - fingerprint = $card.fingerprint - lifecycle = $card.lifecycle - stateKey = $card.stateKey - configured = $false - tokenEnv = $null - chatIdEnv = $null - chatOverride = if ($chatIdOverride) { "telegram-origin" } else { $null } - replyToMessageId = if ($replyToMessageId) { $replyToMessageId } else { $null } - messageId = $null - incidentStatePath = $incidentState.path - incidentStateWritten = $false - visualPath = $null - visualSent = $false - visualError = $null - relay = $null - sent = $false - error = $null - } + $attempt = [pscustomobject]@{ + timestamp = (Get-Date -Format o) + enabled = [bool]$enabled + eventType = $EventType + severity = $Severity + messageFormat = "incident_card_zh_tw_v6" + incidentId = $card.incidentId + fingerprint = $card.fingerprint + lifecycle = $card.lifecycle + stateKey = $card.stateKey + deliveryId = $null + configured = [bool]$gatewayUrl + tokenEnv = $tokenEnv + tokenPresent = $false + chatIdEnv = $null + chatOverride = if ($chatIdOverride) { "telegram-origin" } else { $null } + replyToMessageId = if ($replyToMessageId) { $replyToMessageId } else { $null } + messageId = $null + incidentStatePath = $incidentState.path + incidentStateWritten = $false + visualPath = $null + visualSent = $false + visualError = $null + relay = $null + sent = $false + suppressed = $false + reason = $null + error = $null + rawResponseStored = $false + secretValueLogged = $false + } if (-not $enabled) { $attempt.error = "telegram_disabled" @@ -1746,16 +1770,119 @@ function Send-AgentTelegram { return $attempt } - # Fail closed before reading a sender binding, copying an attachment, or - # attempting any provider call. Agent99 events must be routed by the - # canonical API registry/gateway with a durable receipt. - $attempt.error = "canonical_telegram_gateway_transport_required" - $attempt.sent = $false - $attempt.relay = [pscustomobject]@{ - ok = $false - providerSendPerformed = $false - routeStatus = "blocked_no_egress" - error = "canonical_telegram_gateway_transport_required" + # A successful same-run dispatch is finalized by the API reconciler, which + # owns the terminal Telegram/KM/PlayBook bundle. Failed and scheduled events + # still use this lifecycle ingress so they cannot disappear silently. + $dispatchIdentity = Get-AgentObjectValue $Data "identity" $null + $outcome = Get-AgentObjectValue $Data "outcome" $null + $dispatchIdentitySchema = [string](Get-AgentObjectValue $dispatchIdentity "schemaVersion" (Get-AgentObjectValue $dispatchIdentity "schema_version" "")) + $reconcilerOwnsTerminal = [bool]( + $dispatchIdentity -and + $dispatchIdentitySchema -eq "agent99_controlled_dispatch_identity_v1" -and + [string](Get-AgentObjectValue $outcome "state" "") -eq "resolved" -and + (Convert-AgentBool (Get-AgentObjectValue $outcome "verifierPassed" $false)) -and + (Convert-AgentBool (Get-AgentObjectValue $outcome "sourceEventResolved" $false)) + ) + if ($reconcilerOwnsTerminal) { + $attempt.suppressed = $true + $attempt.reason = "canonical_reconciler_owns_terminal_receipt" + $attempt.relay = [pscustomobject]@{ + ok = $true + providerSendPerformed = $false + routeStatus = "delegated_to_verified_reconciler" + durableAck = $false + error = $null + } + $script:TelegramAttempts += $attempt + return $attempt + } + + if ($gatewayUrl -notmatch '^https://awoooi\.wooo\.work/api/v1/agents/agent99/telegram-lifecycle$') { + $attempt.error = "canonical_gateway_url_invalid" + $script:TelegramAttempts += $attempt + return $attempt + } + $token = Get-AgentEnvironmentValue $tokenEnv + $attempt.tokenPresent = [bool]$token + if (-not $token) { + $attempt.error = "canonical_gateway_token_missing" + $script:TelegramAttempts += $attempt + return $attempt + } + + $deliverySource = "$($card.incidentId)|$($card.fingerprint)|$($card.stateKey)|$($card.runKey)" + $deliveryHashBytes = [Security.Cryptography.SHA256]::Create().ComputeHash([Text.Encoding]::UTF8.GetBytes($deliverySource)) + $deliveryHash = [BitConverter]::ToString($deliveryHashBytes).Replace("-", "").ToLowerInvariant() + $deliveryId = "agent99-lifecycle-$($deliveryHash.Substring(0, 40))" + $attempt.deliveryId = $deliveryId + $payload = [ordered]@{ + schema_version = "agent99_telegram_lifecycle_v1" + delivery_id = $deliveryId + project_id = "awoooi" + incident_id = [string]$card.incidentId + fingerprint = [string]$card.fingerprint + event_type = $EventType + severity = $Severity + 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 } + title = [string]$card.title + target = [string]$card.target + impact = [string]$card.impact + action = [string]$card.action + verification = [string]$card.verification + duration_text = [string]$card.durationText + occurrences = [int]$card.occurrences + next_update = [string]$card.nextUpdate + reply_to_message_id = if ($replyToMessageId) { [int64]$replyToMessageId } else { $null } + occurred_at = (Get-Date -Format o) + } + + try { + $response = Invoke-AgentCanonicalLifecycleIngress $gatewayUrl $token $payload $timeoutSeconds + $sameIdentity = [bool]( + [string](Get-AgentObjectValue $response "delivery_id" "") -eq $deliveryId -and + [string](Get-AgentObjectValue $response "incident_id" "") -eq [string]$card.incidentId + ) + $durableAck = Convert-AgentBool (Get-AgentObjectValue $response "durable_outbound_acknowledged" $false) + $destinationVerified = Convert-AgentBool (Get-AgentObjectValue $response "destination_binding_verified" $false) + $providerMessageId = [string](Get-AgentObjectValue $response "provider_message_id" "") + $deliveryOk = [bool]( + (Convert-AgentBool (Get-AgentObjectValue $response "ok" $false)) -and + $sameIdentity -and + $durableAck -and + $destinationVerified -and + $providerMessageId + ) + $attempt.messageId = if ($providerMessageId) { $providerMessageId } else { $null } + $attempt.sent = $deliveryOk + $attempt.error = if ($deliveryOk) { $null } else { "canonical_gateway_receipt_not_verified" } + $attempt.relay = [pscustomobject]@{ + ok = $deliveryOk + providerSendPerformed = Convert-AgentBool (Get-AgentObjectValue $response "provider_send_performed" $false) + routeStatus = [string](Get-AgentObjectValue $response "delivery_status" "unknown") + durableAck = $durableAck + destinationBindingVerified = $destinationVerified + error = $attempt.error + } + if ($deliveryOk) { + $statePath = Save-AgentTelegramIncidentState $card $incidentState $providerMessageId $replyToMessageId + $attempt.incidentStatePath = $statePath + $attempt.incidentStateWritten = [bool]$statePath + } + } catch { + $failure = Get-AgentSafeHttpFailureReceipt $_ + $attempt.error = if ($failure.errorCode) { [string]$failure.errorCode } else { "canonical_gateway_http_error" } + $attempt.relay = [pscustomobject]@{ + ok = $false + providerSendPerformed = $false + routeStatus = "delivery_failed" + durableAck = $false + destinationBindingVerified = $false + httpStatus = $failure.httpStatus + retryable = $failure.retryable + error = $attempt.error + } } $script:TelegramAttempts += $attempt return $attempt @@ -3253,16 +3380,25 @@ function Test-AgentRecentTelegramDelivery { continue } if (-not $data) { continue } - $attempts = @($data.telegram) + $allAttempts = @($data.telegram) + $attempts = @($allAttempts | Where-Object { + -not ($_.PSObject.Properties["suppressed"] -and $_.suppressed -eq $true) + }) if ($attempts.Count -eq 0) { continue } - $sent = @($attempts | Where-Object { $_.sent -eq $true -or ($_.relay -and $_.relay.ok -eq $true) }) + $sent = @($attempts | Where-Object { + $_.sent -eq $true -and + $_.relay -and + $_.relay.ok -eq $true -and + $_.relay.durableAck -eq $true + }) $checks += [pscustomobject]@{ name = if ($contract.name) { [string]$contract.name } else { [string]$contract.pattern } latestPath = $file.FullName latestFile = $file.Name ageMinutes = [math]::Round(((Get-Date) - $file.LastWriteTime).TotalMinutes, 2) attempts = $attempts.Count + suppressed = $allAttempts.Count - $attempts.Count sent = $sent.Count ok = [bool]($sent.Count -gt 0) } diff --git a/agent99-deploy.ps1 b/agent99-deploy.ps1 index f336da167..3998fc0f4 100644 --- a/agent99-deploy.ps1 +++ b/agent99-deploy.ps1 @@ -507,6 +507,10 @@ try { Add-DefaultProperty $config.completionCallback "batchLimit" 5 Add-DefaultProperty $config.completionCallback "retryAfterMinutes" 5 Add-DefaultProperty $config "telegram" ([pscustomobject]@{}) + Add-DefaultProperty $config.telegram "canonicalGateway" ([pscustomobject]@{}) + Add-DefaultProperty $config.telegram.canonicalGateway "url" "https://awoooi.wooo.work/api/v1/agents/agent99/telegram-lifecycle" + Add-DefaultProperty $config.telegram.canonicalGateway "tokenEnv" "AGENT99_SRE_RELAY_TOKEN" + Add-DefaultProperty $config.telegram.canonicalGateway "timeoutSeconds" 20 Add-DefaultProperty $config.telegram "repeatAfterMinutes" 120 Add-DefaultProperty $config "performance" ([pscustomobject]@{}) Add-DefaultProperty $config.performance "processCpuHostPercentWarning" 25 diff --git a/agent99.config.99.example.json b/agent99.config.99.example.json index 7474b5fb3..92980ff42 100644 --- a/agent99.config.99.example.json +++ b/agent99.config.99.example.json @@ -193,6 +193,11 @@ }, "telegram": { "enabled": true, + "canonicalGateway": { + "url": "https://awoooi.wooo.work/api/v1/agents/agent99/telegram-lifecycle", + "tokenEnv": "AGENT99_SRE_RELAY_TOKEN", + "timeoutSeconds": 20 + }, "botTokenEnv": "AGENT99_TELEGRAM_BOT_TOKEN", "chatIdEnv": "AGENT99_TELEGRAM_CHAT_ID", "botTokenEnvAliases": [ diff --git a/apps/api/src/api/v1/agents.py b/apps/api/src/api/v1/agents.py index 7f5187f71..0a08b9f92 100644 --- a/apps/api/src/api/v1/agents.py +++ b/apps/api/src/api/v1/agents.py @@ -38,13 +38,19 @@ from src.core.config import settings from src.core.context import clear_project_context, set_project_context from src.core.logging import get_logger from src.core.sse import get_publisher -from src.models.agent99_completion import Agent99CompletionCallbackRequest +from src.models.agent99_completion import ( + Agent99CompletionCallbackRequest, + Agent99TelegramLifecycleRequest, +) from src.services.agent99_completion_callback import ( record_agent99_completion_callback, ) from src.services.agent99_enterprise_ai_automation_work_items import ( load_latest_agent99_enterprise_ai_automation_work_items, ) +from src.services.agent99_telegram_lifecycle import ( + deliver_agent99_telegram_lifecycle, +) from src.services.agent_market_governance_snapshot import ( load_latest_agent_market_governance_snapshot, ) @@ -1187,6 +1193,58 @@ async def get_agent99_enterprise_ai_automation_work_items() -> dict[str, Any]: ) from exc +@router.post( + "/agent99/telegram-lifecycle", + response_model=dict[str, Any], + summary="接收 Agent99 結構化生命週期事件", + description=( + "以 Agent99 relay token 驗證 Windows 99 結構化事件卡，交由 AWOOOI " + "canonical Telegram gateway 以 send-once outbox 傳送並回讀 durable ack。" + "此端點不接受 Bot token、任意訊息文字、raw log 或檔案內容。" + ), +) +async def post_agent99_telegram_lifecycle( + payload: Agent99TelegramLifecycleRequest, + x_agent99_lifecycle_token: str | None = Header( + default=None, + alias="X-Agent99-Lifecycle-Token", + ), +) -> dict[str, Any]: + expected_token = settings.AGENT99_SRE_ALERT_RELAY_TOKEN + if not expected_token: + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail="agent99_lifecycle_auth_not_configured", + ) + if not x_agent99_lifecycle_token or not hmac.compare_digest( + x_agent99_lifecycle_token, + expected_token, + ): + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="agent99_lifecycle_auth_failed", + ) + + context_tokens = set_project_context( + project_id=payload.project_id, + source="agent99_telegram_lifecycle", + request_id=payload.delivery_id, + ) + try: + receipt = await deliver_agent99_telegram_lifecycle(payload) + finally: + clear_project_context(context_tokens) + if receipt.get("ok") is not True: + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail=( + "agent99_lifecycle_delivery_not_durably_acknowledged:" + f"{receipt.get('delivery_status') or 'unknown'}" + ), + ) + return receipt + + @router.post( "/agent99/completion-callback", response_model=dict[str, Any], diff --git a/apps/api/src/models/agent99_completion.py b/apps/api/src/models/agent99_completion.py index 1ce7e1e08..2e87798cd 100644 --- a/apps/api/src/models/agent99_completion.py +++ b/apps/api/src/models/agent99_completion.py @@ -40,7 +40,9 @@ class Agent99CompletionCallbackRequest(BaseModel): model_config = ConfigDict(extra="forbid", str_strip_whitespace=True) schema_version: Literal["agent99_completion_callback_v1"] - callback_id: str = Field(min_length=3, max_length=180, pattern=r"^[A-Za-z0-9_.:-]+$") + callback_id: str = Field( + min_length=3, max_length=180, pattern=r"^[A-Za-z0-9_.:-]+$" + ) project_id: str = Field(default="awoooi", min_length=1, max_length=64) run_id: str = Field(min_length=3, max_length=180) trace_id: str = Field(min_length=3, max_length=180) @@ -71,3 +73,50 @@ class Agent99CompletionCallbackRequest(BaseModel): telegram: Agent99TelegramReceipt = Field(default_factory=Agent99TelegramReceipt) problem: Agent99ProblemReceipt = Field(default_factory=Agent99ProblemReceipt) occurred_at: datetime + + +class Agent99TelegramLifecycleRequest(BaseModel): + """Bounded Agent99 event card handed to the canonical Telegram gateway.""" + + model_config = ConfigDict(extra="forbid", str_strip_whitespace=True) + + schema_version: Literal["agent99_telegram_lifecycle_v1"] + delivery_id: str = Field( + min_length=12, + max_length=180, + pattern=r"^[A-Za-z0-9_.:-]+$", + ) + project_id: Literal["awoooi"] = "awoooi" + incident_id: str = Field( + min_length=3, + max_length=180, + pattern=r"^[A-Za-z0-9_.:-]+$", + ) + fingerprint: str = Field(min_length=8, max_length=160) + event_type: str = Field( + min_length=2, + max_length=96, + pattern=r"^[A-Za-z0-9_.:-]+$", + ) + severity: Literal["info", "warning", "critical"] + lifecycle: Literal[ + "detected", + "investigating", + "remediating", + "verifying", + "recovered", + "blocked", + "unknown", + ] + state_key: str = Field(min_length=3, max_length=96) + run_id: str | None = Field(default=None, max_length=180) + 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) + action: str = Field(min_length=1, max_length=360) + verification: str = Field(min_length=1, max_length=300) + duration_text: str = Field(min_length=1, max_length=80) + occurrences: int = Field(default=1, ge=1, le=1_000_000) + next_update: str = Field(min_length=1, max_length=300) + reply_to_message_id: int | None = Field(default=None, ge=1) + occurred_at: datetime diff --git a/apps/api/src/services/agent99_telegram_lifecycle.py b/apps/api/src/services/agent99_telegram_lifecycle.py new file mode 100644 index 000000000..799e93ef1 --- /dev/null +++ b/apps/api/src/services/agent99_telegram_lifecycle.py @@ -0,0 +1,114 @@ +"""Canonical, durable Telegram lifecycle delivery for Windows Agent99.""" + +from __future__ import annotations + +import html +from typing import Any + +from src.models.agent99_completion import Agent99TelegramLifecycleRequest +from src.services.telegram_gateway import get_telegram_gateway + + +_LIFECYCLE_LABELS = { + "detected": "已偵測 / DETECTED", + "investigating": "調查中 / INVESTIGATING", + "remediating": "修復中 / REMEDIATING", + "verifying": "驗證中 / VERIFYING", + "recovered": "已恢復 / RECOVERED", + "blocked": "需介入 / ACTION REQUIRED", + "unknown": "狀態未知 / UNKNOWN", +} + +_SEVERITY_LABELS = { + "critical": "重大 / CRITICAL", + "warning": "警告 / WARNING", + "info": "資訊 / INFO", +} + + +def _format_lifecycle_card(payload: Agent99TelegramLifecycleRequest) -> str: + def esc(value: object) -> str: + return html.escape(str(value or "")) + + return "\n".join( + [ + f"Agent99 事件｜{esc(payload.title)}", + ( + f"狀態：{esc(_LIFECYCLE_LABELS[payload.lifecycle])}" + f"　等級：{esc(_SEVERITY_LABELS[payload.severity])}" + ), + "────────────────────", + f"資產　{esc(payload.target)}", + f"影響　{esc(payload.impact)}", + f"AI 處置　{esc(payload.action)}", + f"驗證　{esc(payload.verification)}", + "────────────────────", + ( + f"事件：{esc(payload.incident_id)}　" + f"發生：{payload.occurrences} 次　耗時：{esc(payload.duration_text)}" + ), + f"下一次更新：{esc(payload.next_update)}", + f"時間：{esc(payload.occurred_at.isoformat())}", + ] + )[:4096] + + +async def deliver_agent99_telegram_lifecycle( + payload: Agent99TelegramLifecycleRequest, +) -> dict[str, Any]: + """Send once through the registered AWOOOI SRE route and verify its ack.""" + + response = await get_telegram_gateway().send_agent99_lifecycle_receipt( + delivery_id=payload.delivery_id, + incident_id=payload.incident_id, + state_key=payload.state_key, + text=_format_lifecycle_card(payload), + priority="P0" if payload.severity == "critical" else "P1", + project_id=payload.project_id, + reply_to_message_id=payload.reply_to_message_id, + ) + result = response.get("result") if isinstance(response, dict) else None + message_id = str(result.get("message_id") or "") if isinstance(result, dict) else "" + delivery_context = ( + response.get("_awooop_delivery_context") + if isinstance(response, dict) + and isinstance(response.get("_awooop_delivery_context"), dict) + else {} + ) + delivery_status = ( + str(response.get("_awooop_delivery_status") or "") + if isinstance(response, dict) + else "" + ) + durable_ack = bool( + isinstance(response, dict) + and response.get("_awooop_outbound_mirror_acknowledged") is True + ) + destination_verified = bool( + delivery_status == "sent_reused" + or delivery_context.get("destination_binding_verified") is True + ) + ok = bool( + durable_ack + and destination_verified + and message_id + and delivery_status in {"sent", "sent_reused"} + ) + return { + "schema_version": "agent99_telegram_lifecycle_receipt_v1", + "ok": ok, + "delivery_id": payload.delivery_id, + "incident_id": payload.incident_id, + "event_type": payload.event_type, + "lifecycle": payload.lifecycle, + "provider_message_id": message_id or None, + "delivery_status": delivery_status or "unknown", + "durable_outbound_acknowledged": durable_ack, + "destination_binding_verified": destination_verified, + "provider_send_performed": bool( + isinstance(response, dict) + and response.get("_awooop_provider_send_performed") is True + ), + "stores_secret": False, + "raw_response_stored": False, + } diff --git a/apps/api/src/services/telegram_gateway.py b/apps/api/src/services/telegram_gateway.py index 292409548..4a1ae6d0c 100644 --- a/apps/api/src/services/telegram_gateway.py +++ b/apps/api/src/services/telegram_gateway.py @@ -2343,13 +2343,14 @@ def _legacy_outbound_run_id(chat_id: str, provider_message_id: str) -> UUID: def _controlled_apply_result_delivery_identity( source_envelope_extra: object, ) -> dict[str, str] | None: - """Return the stable identity required before a controlled-result send.""" + """Return the stable identity required before a durable lifecycle send.""" if not isinstance(source_envelope_extra, dict): return None callback_reply = source_envelope_extra.get("callback_reply") if not isinstance(callback_reply, dict): return None - if callback_reply.get("action") != "controlled_apply_result": + delivery_kind = str(callback_reply.get("action") or "").strip() + if delivery_kind not in {"controlled_apply_result", "agent99_lifecycle"}: return None identity = { @@ -2359,15 +2360,22 @@ def _controlled_apply_result_delivery_identity( ).strip(), "incident_id": str(callback_reply.get("incident_id") or "").strip(), "apply_op_id": str(callback_reply.get("apply_op_id") or "").strip(), + "delivery_kind": delivery_kind, } return identity if all(identity.values()) else None def _controlled_apply_result_delivery_run_id(identity: dict[str, str]) -> UUID: """Build one stable outbox identity for one controlled apply result.""" + delivery_kind = identity.get("delivery_kind", "controlled_apply_result") + namespace = ( + "awoooi:controlled-apply-result" + if delivery_kind == "controlled_apply_result" + else f"awoooi:{delivery_kind}" + ) return uuid5( NAMESPACE_URL, - "awoooi:controlled-apply-result:" + f"{namespace}:" f"{identity['project_id']}:" f"{identity['automation_run_id']}:" f"{identity['incident_id']}:" @@ -5774,6 +5782,10 @@ class TelegramGateway: project_id = identity["project_id"] delivery_run_id = _controlled_apply_result_delivery_run_id(identity) + delivery_kind = identity.get( + "delivery_kind", + "controlled_apply_result", + ) notification_policy = source_envelope_extra.get("notification_policy") if not isinstance(notification_policy, dict): notification_policy = {} @@ -5928,9 +5940,7 @@ class TelegramGateway: callback_reply = dict( reservation_extra.get("callback_reply") or {} ) - callback_reply["status"] = ( - "controlled_apply_result_reserved" - ) + callback_reply["status"] = f"{delivery_kind}_reserved" reservation_extra["callback_reply"] = callback_reply reservation_message_id = str( await record_outbound_message( @@ -5956,7 +5966,7 @@ class TelegramGateway: ), provider_message_id=None, send_status="pending", - triggered_by_state="controlled_apply_result", + triggered_by_state=delivery_kind, is_shadow=False, ) ) @@ -6066,7 +6076,11 @@ class TelegramGateway: if not message_id or not run_id or not provider_message_id: return False - lock_key = f"telegram-controlled-apply-result:{run_id}" + delivery_kind = identity.get( + "delivery_kind", + "controlled_apply_result", + ) + lock_key = f"telegram-{delivery_kind}:{run_id}" for attempt in range(_CONTROLLED_APPLY_OUTBOUND_FINALIZE_ATTEMPTS): finalized = False try: @@ -6086,7 +6100,7 @@ class TelegramGateway: send_status = 'sent', send_error = NULL, sent_at = coalesce(sent_at, NOW()), - triggered_by_state = 'controlled_apply_result', + triggered_by_state = :triggered_by_state, source_envelope = jsonb_set( jsonb_set( source_envelope, @@ -6114,6 +6128,7 @@ class TelegramGateway: "message_id": message_id, "run_id": run_id, "provider_message_id": provider_message_id, + "triggered_by_state": delivery_kind, }, ) row = result.mappings().one_or_none() @@ -6276,7 +6291,8 @@ class TelegramGateway: ) controlled_apply_result_requested = bool( isinstance(callback_reply, dict) - and callback_reply.get("action") == "controlled_apply_result" + and callback_reply.get("action") + in {"controlled_apply_result", "agent99_lifecycle"} ) controlled_delivery_identity = ( _controlled_apply_result_delivery_identity(source_envelope_extra) @@ -11586,6 +11602,80 @@ class TelegramGateway: payload["reply_to_message_id"] = inbound_message_id return await self._send_request("sendMessage", payload) + async def send_agent99_lifecycle_receipt( + self, + *, + delivery_id: str, + incident_id: str, + state_key: str, + text: str, + priority: str, + project_id: str = "awoooi", + reply_to_message_id: int | None = None, + ) -> dict: + """Send one structured Agent99 lifecycle update with a durable ack.""" + + source_extra = _callback_reply_source_envelope_extra( + incident_id=incident_id, + failure_context="agent99_lifecycle", + status="agent99_lifecycle_pending", + chunk_index=0, + chunk_count=1, + callback_action="agent99_lifecycle", + parse_mode="HTML", + parent_message_id=reply_to_message_id, + ) + if source_extra is None: + return { + "ok": False, + "_awooop_outbound_mirror_acknowledged": False, + "_awooop_delivery_status": "incident_identity_missing", + "_awooop_provider_send_performed": False, + } + source_extra["outbound_message_type"] = ( + "final" if "recovered" in state_key else "error" + ) + source_extra["execution_kind"] = "agent99_lifecycle" + source_extra["notification_policy"] = { + "policy_version": "agent99_lifecycle_send_once_v1", + "provider_delivery": "immediate", + "disposition": "send_once", + "provider_send_performed": None, + "details_retained_in": [ + "awooop_outbound_message", + "telegram_outbound_receipts", + ], + } + callback_reply = source_extra.get("callback_reply") + if isinstance(callback_reply, dict): + callback_reply["automation_run_id"] = delivery_id + callback_reply["apply_op_id"] = state_key + callback_reply["project_id"] = project_id or "awoooi" + source_refs = source_extra.get("source_refs") + if isinstance(source_refs, dict): + source_refs["automation_run_ids"] = [delivery_id] + + payload: dict[str, object] = { + "text": text[:4096], + "parse_mode": "HTML", + "disable_web_page_preview": True, + _CANONICAL_ROUTE_CONTEXT_KEY: { + "product_id": "awoooi", + "signal_family": "incident_lifecycle", + "severity": priority, + }, + _AWOOOP_SOURCE_ENVELOPE_EXTRA_KEY: source_extra, + } + reply_markup = incident_truth_chain_reply_markup( + incident_id, + project_id=project_id or "awoooi", + ) + if reply_markup: + payload["reply_markup"] = reply_markup + if reply_to_message_id is not None: + payload["reply_to_message_id"] = reply_to_message_id + return await self._send_request("sendMessage", payload) + async def send_controlled_apply_result_receipt( self, *, diff --git a/apps/api/tests/test_agent99_telegram_lifecycle_api.py b/apps/api/tests/test_agent99_telegram_lifecycle_api.py new file mode 100644 index 000000000..1a0996409 --- /dev/null +++ b/apps/api/tests/test_agent99_telegram_lifecycle_api.py @@ -0,0 +1,208 @@ +from __future__ import annotations + +# ruff: noqa: E402 +import os + +os.environ.setdefault("DATABASE_URL", "postgresql+asyncpg://test:test@localhost/test") + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient +from pydantic import ValidationError + +from src.api.v1 import agents as agents_api +from src.core.config import settings +from src.models.agent99_completion import Agent99TelegramLifecycleRequest +from src.services import agent99_telegram_lifecycle as lifecycle_service +from src.services import telegram_gateway as gateway_service + + +def payload() -> dict: + return { + "schema_version": "agent99_telegram_lifecycle_v1", + "delivery_id": "agent99-lifecycle-0123456789abcdef0123456789abcdef01234567", + "project_id": "awoooi", + "incident_id": "INC-20260715-AG9901", + "fingerprint": "0123456789abcdef0123456789abcdef", + "event_type": "performance_warning", + "severity": "warning", + "lifecycle": "investigating", + "state_key": "investigating|warning", + "run_id": "run-agent99-20260715-01", + "title": "Host 110 主機效能異常", + "target": "192.168.0.110", + "impact": "資源壓力可能降低服務回應速度。", + "action": "Agent99 正在執行 allowlisted 診斷與降載。", + "verification": "等待 CPU、記憶體與磁碟 post-verifier。", + "duration_text": "12.4 秒", + "occurrences": 2, + "next_update": "狀態改變時更新，相同狀態不重複推播。", + "occurred_at": "2026-07-15T04:20:00+08:00", + } + + +def app_client() -> TestClient: + app = FastAPI() + app.include_router(agents_api.router, prefix="/api/v1") + return TestClient(app) + + +def test_lifecycle_payload_rejects_unbounded_content() -> None: + with pytest.raises(ValidationError): + Agent99TelegramLifecycleRequest.model_validate( + {**payload(), "raw_log": "must-not-enter-lifecycle-ingress"} + ) + with pytest.raises(ValidationError): + Agent99TelegramLifecycleRequest.model_validate( + {**payload(), "incident_id": r"C:\Wooo\Agent99\evidence\raw.json"} + ) + with pytest.raises(ValidationError): + Agent99TelegramLifecycleRequest.model_validate( + {**payload(), "project_id": "unregistered-product"} + ) + + +def test_lifecycle_endpoint_rejects_wrong_token(monkeypatch) -> None: + monkeypatch.setattr(settings, "AGENT99_SRE_ALERT_RELAY_TOKEN", "expected") + + response = app_client().post( + "/api/v1/agents/agent99/telegram-lifecycle", + headers={"X-Agent99-Lifecycle-Token": "wrong"}, + json=payload(), + ) + + assert response.status_code == 401 + assert response.json()["detail"] == "agent99_lifecycle_auth_failed" + + +def test_lifecycle_endpoint_requires_durable_gateway_ack(monkeypatch) -> None: + monkeypatch.setattr(settings, "AGENT99_SRE_ALERT_RELAY_TOKEN", "expected") + + async def failed_delivery(request): + return { + "ok": False, + "delivery_id": request.delivery_id, + "delivery_status": "pending_unknown", + } + + monkeypatch.setattr( + agents_api, + "deliver_agent99_telegram_lifecycle", + failed_delivery, + ) + response = app_client().post( + "/api/v1/agents/agent99/telegram-lifecycle", + headers={"X-Agent99-Lifecycle-Token": "expected"}, + json=payload(), + ) + + assert response.status_code == 503 + assert "pending_unknown" in response.json()["detail"] + + +def test_lifecycle_endpoint_returns_same_delivery_receipt(monkeypatch) -> None: + monkeypatch.setattr(settings, "AGENT99_SRE_ALERT_RELAY_TOKEN", "expected") + + async def successful_delivery(request): + return { + "schema_version": "agent99_telegram_lifecycle_receipt_v1", + "ok": True, + "delivery_id": request.delivery_id, + "incident_id": request.incident_id, + "delivery_status": "sent", + "provider_message_id": "8123", + "durable_outbound_acknowledged": True, + } + + monkeypatch.setattr( + agents_api, + "deliver_agent99_telegram_lifecycle", + successful_delivery, + ) + response = app_client().post( + "/api/v1/agents/agent99/telegram-lifecycle", + headers={"X-Agent99-Lifecycle-Token": "expected"}, + json=payload(), + ) + + assert response.status_code == 200 + assert response.json()["delivery_id"] == payload()["delivery_id"] + assert response.json()["provider_message_id"] == "8123" + + +def test_lifecycle_outbox_identity_is_stable_and_separate() -> None: + base = { + "callback_reply": { + "project_id": "awoooi", + "automation_run_id": "agent99-lifecycle-0123456789", + "incident_id": "INC-20260715-AG9901", + "apply_op_id": "investigating|warning", + } + } + lifecycle_source = { + "callback_reply": { + **base["callback_reply"], + "action": "agent99_lifecycle", + } + } + controlled_source = { + "callback_reply": { + **base["callback_reply"], + "action": "controlled_apply_result", + } + } + + lifecycle_identity = gateway_service._controlled_apply_result_delivery_identity( + lifecycle_source + ) + controlled_identity = gateway_service._controlled_apply_result_delivery_identity( + controlled_source + ) + + assert lifecycle_identity is not None + assert controlled_identity is not None + assert lifecycle_identity["delivery_kind"] == "agent99_lifecycle" + assert controlled_identity["delivery_kind"] == "controlled_apply_result" + assert gateway_service._controlled_apply_result_delivery_run_id( + lifecycle_identity + ) == gateway_service._controlled_apply_result_delivery_run_id(lifecycle_identity) + assert gateway_service._controlled_apply_result_delivery_run_id( + lifecycle_identity + ) != gateway_service._controlled_apply_result_delivery_run_id(controlled_identity) + + +@pytest.mark.asyncio +async def test_lifecycle_service_uses_durable_canonical_sender(monkeypatch) -> None: + calls: list[dict] = [] + + class Gateway: + async def send_agent99_lifecycle_receipt(self, **kwargs): + calls.append(kwargs) + return { + "ok": True, + "result": {"message_id": 9123}, + "_awooop_outbound_mirror_acknowledged": True, + "_awooop_delivery_status": "sent", + "_awooop_provider_send_performed": True, + "_awooop_delivery_context": { + "destination_binding_verified": True, + }, + } + + monkeypatch.setattr( + lifecycle_service, + "get_telegram_gateway", + lambda: Gateway(), + ) + request = Agent99TelegramLifecycleRequest.model_validate(payload()) + + receipt = await lifecycle_service.deliver_agent99_telegram_lifecycle(request) + + assert receipt["ok"] is True + assert receipt["durable_outbound_acknowledged"] is True + assert receipt["destination_binding_verified"] is True + assert receipt["provider_message_id"] == "9123" + assert calls[0]["priority"] == "P1" + assert "Agent99 事件" in calls[0]["text"] + assert "AI 處置" in calls[0]["text"] + assert "raw_log" not in calls[0]["text"] diff --git a/apps/api/tests/test_agent99_transport_recovery_deploy_contract.py b/apps/api/tests/test_agent99_transport_recovery_deploy_contract.py index 3ed3908d6..db4cf9002 100644 --- a/apps/api/tests/test_agent99_transport_recovery_deploy_contract.py +++ b/apps/api/tests/test_agent99_transport_recovery_deploy_contract.py @@ -111,9 +111,11 @@ def test_agent99_telegram_lifecycle_fails_closed_without_direct_sender() -> None assert "lastRunKey = $Card.runKey" in source assert "$previousRunKey -eq [string]$incidentCard.runKey" in source assert '$dedupeParts += "run=$($incidentCard.runKey)"' in source - assert 'error = "canonical_telegram_gateway_transport_required"' in source - assert 'providerSendPerformed = $false' in source - assert 'routeStatus = "blocked_no_egress"' in source + assert "agent99/telegram-lifecycle" in source + assert '"X-Agent99-Lifecycle-Token" = $Token' in source + assert "durable_outbound_acknowledged" in source + assert "destination_binding_verified" in source + assert 'reason = "canonical_reconciler_owns_terminal_receipt"' in source assert "function Send-AgentTelegramRelay" not in source assert "function Send-AgentTelegramPhotoDirect" not in source assert "telegram_receipt_b64=" not in source @@ -403,10 +405,13 @@ def test_agent99_completion_callback_waits_for_durable_same_trace_readback() -> config = json.loads((ROOT / "agent99.config.99.example.json").read_text()) callback = config["completionCallback"] + lifecycle = config["telegram"]["canonicalGateway"] assert callback["enabled"] is True assert callback["url"].startswith("https://awoooi.wooo.work/") assert callback["tokenEnv"] == "AGENT99_SRE_RELAY_TOKEN" assert callback["retryAfterMinutes"] == 5 + assert lifecycle["url"].endswith("/api/v1/agents/agent99/telegram-lifecycle") + assert lifecycle["tokenEnv"] == "AGENT99_SRE_RELAY_TOKEN" assert 'schema_version = "agent99_completion_callback_v1"' in source assert '"X-Agent99-Completion-Token" = $token' in source assert '$receipt.durableReadback' in source @@ -415,10 +420,21 @@ def test_agent99_completion_callback_waits_for_durable_same_trace_readback() -> assert 'state\\completion-callbacks\\pending' in source assert 'Invoke-AgentPendingCompletionCallbacks' in source assert 'Add-DefaultProperty $config "completionCallback"' in deploy + assert 'Add-DefaultProperty $config.telegram "canonicalGateway"' in deploy assert 'rawResponseStored = $false' in source assert 'secretValueLogged = $false' in source +def test_agent99_selfhealth_ignores_suppressed_lifecycle_attempts() -> None: + source = CONTROL.read_text(encoding="utf-8") + function = source[source.index("function Test-AgentRecentTelegramDelivery") :] + function = function[: function.index("function Test-AgentRuntimeManifest")] + + assert 'PSObject.Properties["suppressed"]' in function + assert "$_.suppressed -eq $true" in function + assert "$_.relay.durableAck -eq $true" in function + + def test_agent99_object_reader_preserves_ordered_callback_identity() -> None: source = CONTROL.read_text(encoding="utf-8") helper = source[source.index("function Get-AgentObjectValue") :] diff --git a/apps/api/tests/test_telegram_notification_egress_scanners.py b/apps/api/tests/test_telegram_notification_egress_scanners.py index 3959d99d5..2156e6264 100644 --- a/apps/api/tests/test_telegram_notification_egress_scanners.py +++ b/apps/api/tests/test_telegram_notification_egress_scanners.py @@ -490,6 +490,7 @@ def test_agent99_has_no_direct_or_custom_telegram_sender_source() -> None: ) assert "function Send-AgentTelegramRelay" not in source assert "function Send-AgentTelegramPhotoDirect" not in source - assert "canonical_telegram_gateway_transport_required" in source - assert "providerSendPerformed = $false" in source - assert 'routeStatus = "blocked_no_egress"' in source + assert "agent99/telegram-lifecycle" in source + assert '"X-Agent99-Lifecycle-Token" = $Token' in source + assert "durable_outbound_acknowledged" in source + assert "destination_binding_verified" in source