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/core/config.py b/apps/api/src/core/config.py index 8a310cbaf..2aec50780 100644 --- a/apps/api/src/core/config.py +++ b/apps/api/src/core/config.py @@ -298,6 +298,16 @@ class Settings(BaseSettings): "processes that share a constrained production role." ), ) + DATABASE_NULL_POOL_CONCURRENCY_LIMIT: int = Field( + default=0, + ge=0, + le=50, + description=( + "Per-process concurrent DB session cap when NullPool is enabled. " + "Zero leaves concurrency unchanged; bounded workers set this to " + "their verified role budget." + ), + ) DATABASE_BOOTSTRAP_DDL_ENABLED: bool = Field( default=True, description=( diff --git a/apps/api/src/db/base.py b/apps/api/src/db/base.py index 652f310b5..684405284 100644 --- a/apps/api/src/db/base.py +++ b/apps/api/src/db/base.py @@ -50,9 +50,54 @@ class Base(DeclarativeBase): _engine: AsyncEngine | None = None _session_factory: async_sessionmaker[AsyncSession] | None = None +_null_pool_session_limiter: asyncio.BoundedSemaphore | None = None +_null_pool_session_limiter_size = 0 logger = get_logger("awoooi.db") +def _get_null_pool_session_limiter() -> asyncio.BoundedSemaphore | None: + """Honor the configured connection budget even when NullPool is active.""" + + global _null_pool_session_limiter, _null_pool_session_limiter_size + limit = int(settings.DATABASE_NULL_POOL_CONCURRENCY_LIMIT) + if not settings.DATABASE_NULL_POOL or limit <= 0: + return None + if _null_pool_session_limiter is None or _null_pool_session_limiter_size != limit: + _null_pool_session_limiter = asyncio.BoundedSemaphore(limit) + _null_pool_session_limiter_size = limit + return _null_pool_session_limiter + + +@asynccontextmanager +async def _database_session_slot() -> AsyncGenerator[None, None]: + """Bound concurrent NullPool sessions without retaining DB connections.""" + + limiter = _get_null_pool_session_limiter() + if limiter is None: + yield + return + try: + await asyncio.wait_for( + limiter.acquire(), + timeout=settings.DATABASE_POOL_TIMEOUT_SECONDS, + ) + except TimeoutError as exc: + logger.warning( + "database_null_pool_session_concurrency_limit_reached", + concurrency_limit=int( + settings.DATABASE_NULL_POOL_CONCURRENCY_LIMIT + ), + timeout_seconds=settings.DATABASE_POOL_TIMEOUT_SECONDS, + ) + raise SQLAlchemyTimeoutError( + "NullPool database session concurrency limit reached" + ) from exc + try: + yield + finally: + limiter.release() + + def _raise_unauthorized_db_context(msg: str) -> None: context = get_current_project_context() logger.error( @@ -133,27 +178,28 @@ async def get_db() -> AsyncGenerator[AsyncSession, None]: ... """ factory = get_session_factory() - async with factory() as session: - try: - from src.core.context import get_current_project_id + async with _database_session_slot(): + async with factory() as session: + try: + from src.core.context import get_current_project_id - # AwoooP Phase 2.3 (2026-05-04 ogt): SET LOCAL app.project_id 讓 RLS Policy 生效 - # Fail-Closed RLS: 遇到未授權情境拋出錯誤而非回退到 "awoooi" - pid = get_current_project_id() - if not pid: - _raise_unauthorized_db_context( - "Unauthorized: project_id is missing in context (Fail-Closed RLS)" + # AwoooP Phase 2.3 (2026-05-04 ogt): SET LOCAL app.project_id 讓 RLS Policy 生效 + # Fail-Closed RLS: 遇到未授權情境拋出錯誤而非回退到 "awoooi" + pid = get_current_project_id() + if not pid: + _raise_unauthorized_db_context( + "Unauthorized: project_id is missing in context (Fail-Closed RLS)" + ) + + await session.execute( + text("SELECT set_config('app.project_id', :pid, TRUE)"), + {"pid": pid}, ) - - await session.execute( - text("SELECT set_config('app.project_id', :pid, TRUE)"), - {"pid": pid}, - ) - yield session - await session.commit() - except Exception: - await session.rollback() - raise + yield session + await session.commit() + except Exception: + await session.rollback() + raise @asynccontextmanager @@ -178,17 +224,18 @@ async def get_db_context(project_id: str | None = None) -> AsyncGenerator[AsyncS _raise_unauthorized_db_context("Unauthorized: project_id is missing in context (Fail-Closed RLS)") factory = get_session_factory() - async with factory() as session: - try: - await session.execute( - text("SELECT set_config('app.project_id', :pid, TRUE)"), - {"pid": effective_pid}, - ) - yield session - await session.commit() - except Exception: - await session.rollback() - raise + async with _database_session_slot(): + async with factory() as session: + try: + await session.execute( + text("SELECT set_config('app.project_id', :pid, TRUE)"), + {"pid": effective_pid}, + ) + yield session + await session.commit() + except Exception: + await session.rollback() + raise # ============================================================================= diff --git a/apps/api/src/jobs/agent99_controlled_dispatch_reconciler_job.py b/apps/api/src/jobs/agent99_controlled_dispatch_reconciler_job.py index 55ae37c7a..728d64997 100644 --- a/apps/api/src/jobs/agent99_controlled_dispatch_reconciler_job.py +++ b/apps/api/src/jobs/agent99_controlled_dispatch_reconciler_job.py @@ -5,6 +5,7 @@ from __future__ import annotations import asyncio import hashlib from collections.abc import Awaitable, Callable +from datetime import UTC, datetime from typing import Any import structlog @@ -54,6 +55,25 @@ LegacyIncidentFetcher = Callable[..., Awaitable[list[dict[str, Any]]]] Agent99Dispatcher = Callable[..., Awaitable[dict[str, Any]]] +def _dispatch_timeout_elapsed(value: Any) -> bool: + if not isinstance(value, datetime): + return False + normalized = value if value.tzinfo is not None else value.replace(tzinfo=UTC) + return normalized <= datetime.now(UTC) + + +async def _terminalize_expired_outcome( + ledger: Any, + item: dict[str, Any], +) -> bool: + if not _dispatch_timeout_elapsed(item.get("timeout_at")): + return False + terminal = await ledger.record_outcome_timeout_no_write_terminal( + identity=item["identity"], + ) + return terminal.get("receipt_persisted") is True + + def _is_bound_no_write_status_outcome( outcome: dict[str, Any] | None, *, @@ -594,6 +614,7 @@ async def reconcile_agent99_controlled_dispatches_once( "status_reconcile_requested": 0, "status_reconcile_pending": 0, "external_no_write_reconciled": 0, + "outcome_timeout_terminalized": 0, "verifier_written": 0, "closed": 0, } @@ -622,7 +643,10 @@ async def reconcile_agent99_controlled_dispatches_once( source_receipt.get("resolved") is not True or not source_receipt_ref ): - counters["source_not_resolved"] += 1 + if await _terminalize_expired_outcome(ledger, item): + counters["outcome_timeout_terminalized"] += 1 + else: + counters["source_not_resolved"] += 1 continue outcome = await asyncio.to_thread( read_agent99_sre_outcome, @@ -633,6 +657,9 @@ async def reconcile_agent99_controlled_dispatches_once( identity=identity, source_receipt_ref=source_receipt_ref, ): + if await _terminalize_expired_outcome(ledger, item): + counters["outcome_timeout_terminalized"] += 1 + continue request_receipt = await asyncio.to_thread( request_agent99_same_run_status_reconcile, identity, @@ -674,6 +701,8 @@ async def reconcile_agent99_controlled_dispatches_once( str(identity.run_id), ) if not isinstance(outcome, dict): + if await _terminalize_expired_outcome(ledger, item): + counters["outcome_timeout_terminalized"] += 1 continue mode = str(outcome.get("mode") or mode) outcome_contract = ( @@ -707,6 +736,8 @@ async def reconcile_agent99_controlled_dispatches_once( evidence_refs=evidence_refs, ) if verifier.get("receipt_persisted") is not True: + if await _terminalize_expired_outcome(ledger, item): + counters["outcome_timeout_terminalized"] += 1 continue counters["verifier_written"] += 1 receipt = verifier @@ -735,7 +766,20 @@ async def run_agent99_controlled_dispatch_reconciler_loop() -> None: error=str(exc), ) try: - await reconcile_agent99_controlled_dispatches_once() + reconciled = await reconcile_agent99_controlled_dispatches_once() + if any( + reconciled[key] > 0 + for key in ( + "external_no_write_reconciled", + "outcome_timeout_terminalized", + "verifier_written", + "closed", + ) + ): + logger.info( + "agent99_controlled_dispatch_reconciler_tick", + **reconciled, + ) except asyncio.CancelledError: raise except Exception as exc: diff --git a/apps/api/src/jobs/asset_capability_reconciliation_job.py b/apps/api/src/jobs/asset_capability_reconciliation_job.py index 33625980f..8073fbed3 100644 --- a/apps/api/src/jobs/asset_capability_reconciliation_job.py +++ b/apps/api/src/jobs/asset_capability_reconciliation_job.py @@ -33,6 +33,8 @@ logger = structlog.get_logger(__name__) _FIRST_DELAY_SECONDS = 90 _LOOP_BACKOFF_SECONDS = 30 * 60 +_MAX_RECONCILIATION_CANDIDATES = 20_000 +_RETRYABLE_RESULT_STATUSES = {"blocked_safe_no_write", "degraded_no_write"} _DAILY_TRIGGER_HOUR_TAIPEI = 3 _DAILY_TRIGGER_MINUTE_TAIPEI = 30 _TAIPEI = ZoneInfo("Asia/Taipei") @@ -187,17 +189,30 @@ async def run_asset_capability_reconciliation_loop() -> None: daily_trigger_minute_taipei=_DAILY_TRIGGER_MINUTE_TAIPEI, ) await asyncio.sleep(_FIRST_DELAY_SECONDS) + triggered_by = "startup" while True: try: - await reconcile_once() + result = await reconcile_once(triggered_by=triggered_by) except Exception as exc: logger.exception( "asset_capability_reconciliation_loop_error", error_type=type(exc).__name__, ) await asyncio.sleep(_LOOP_BACKOFF_SECONDS) + triggered_by = "retry" + continue + if str(result.get("status") or "") in _RETRYABLE_RESULT_STATUSES: + logger.warning( + "asset_capability_reconciliation_retry_scheduled", + status=result.get("status"), + reason=result.get("reason"), + retry_seconds=_LOOP_BACKOFF_SECONDS, + ) + await asyncio.sleep(_LOOP_BACKOFF_SECONDS) + triggered_by = "retry" continue await asyncio.sleep(_seconds_until_next_trigger()) + triggered_by = "cron" async def reconcile_once( @@ -231,6 +246,19 @@ async def reconcile_once( } candidates = build_reconciliation_candidates(matrix) + if len(candidates) > _MAX_RECONCILIATION_CANDIDATES: + logger.error( + "asset_capability_reconciliation_candidate_bound_exceeded", + candidate_count=len(candidates), + max_candidate_count=_MAX_RECONCILIATION_CANDIDATES, + external_runtime_write_performed=False, + ) + return { + "status": "blocked_safe_no_write", + "reason": "candidate_count_above_controlled_execution_bound", + "candidate_count": len(candidates), + "max_candidate_count": _MAX_RECONCILIATION_CANDIDATES, + } receipt = await _persist_reconciliation( matrix, candidates, @@ -284,6 +312,8 @@ async def _persist_reconciliation( AND COALESCE(input ->> 'semantic_operation_type', operation_type) = :semantic_operation_type AND input ->> 'idempotency_key' = :idempotency_key + AND status = 'success' + AND output ->> 'closed' = 'true' ORDER BY created_at DESC LIMIT 1 """ @@ -298,28 +328,38 @@ async def _persist_reconciliation( if isinstance(prior, dict): return prior - for candidate_raw in candidates: - candidate = dict(candidate_raw) + candidate_fingerprints = [ + str(candidate.get("candidate_fingerprint") or "") + for candidate in candidates + ] + existing_fingerprints: set[str] = set() + if candidate_fingerprints: existing = await db.execute( text( """ - SELECT op_id::text + SELECT DISTINCT input ->> 'candidate_fingerprint' FROM automation_operation_log WHERE actor = :actor AND COALESCE(input ->> 'semantic_operation_type', operation_type) = :semantic_operation_type - AND input ->> 'candidate_fingerprint' = :candidate_fingerprint - ORDER BY created_at DESC - LIMIT 1 + AND input ->> 'candidate_fingerprint' + = ANY(CAST(:candidate_fingerprints AS text[])) """ ), { "actor": RECONCILIATION_ACTOR, "semantic_operation_type": RECONCILIATION_WORK_ITEM_OPERATION, - "candidate_fingerprint": candidate["candidate_fingerprint"], + "candidate_fingerprints": candidate_fingerprints, }, ) - if existing.scalar_one_or_none(): + existing_fingerprints = { + str(row[0]) for row in existing.fetchall() if row[0] + } + + insert_rows: list[dict[str, Any]] = [] + for candidate_raw in candidates: + candidate = dict(candidate_raw) + if candidate["candidate_fingerprint"] in existing_fingerprints: continue input_payload = build_reconciliation_work_item_input( candidate, @@ -334,6 +374,29 @@ async def _persist_reconciliation( "post_verifier": "pending_batch_readback", "rollback_or_no_write_terminal": "no_external_runtime_write", } + insert_rows.append( + { + "asset_id": candidate.get("asset_inventory_id"), + "discovery_run_id": discovery_run_id, + "actor": RECONCILIATION_ACTOR, + "input": _canonical_json(input_payload), + "output": _canonical_json(output_payload), + "dry_run_result": _canonical_json( + { + "would_create_work_item": True, + "external_runtime_write": False, + "candidate_fingerprint": candidate["candidate_fingerprint"], + } + ), + "tags": [ + "asset_capability_reconciliation", + "work_item", + "aia_p0_006", + ], + } + ) + + if insert_rows: await db.execute( text( """ @@ -360,50 +423,37 @@ async def _persist_reconciliation( ) """ ), - { - "asset_id": candidate.get("asset_inventory_id"), - "discovery_run_id": discovery_run_id, - "actor": RECONCILIATION_ACTOR, - "input": _canonical_json(input_payload), - "output": _canonical_json(output_payload), - "dry_run_result": _canonical_json( - { - "would_create_work_item": True, - "external_runtime_write": False, - "candidate_fingerprint": candidate["candidate_fingerprint"], - } - ), - "tags": [ - "asset_capability_reconciliation", - "work_item", - "aia_p0_006", - ], - }, + insert_rows, ) - created_count += 1 + created_count = len(insert_rows) - rows_result = await db.execute( - text( - """ - SELECT input ->> 'candidate_fingerprint' AS candidate_fingerprint - FROM automation_operation_log - WHERE actor = :actor - AND COALESCE(input ->> 'semantic_operation_type', operation_type) - = :semantic_operation_type - """ - ), - { - "actor": RECONCILIATION_ACTOR, - "semantic_operation_type": RECONCILIATION_WORK_ITEM_OPERATION, - }, - ) - persisted_fingerprints = { - str(row[0]) for row in rows_result.fetchall() if row[0] - } expected_fingerprints = { str(candidate.get("candidate_fingerprint") or "") for candidate in candidates } + persisted_fingerprints: set[str] = set() + if candidate_fingerprints: + rows_result = await db.execute( + text( + """ + SELECT DISTINCT input ->> 'candidate_fingerprint' + FROM automation_operation_log + WHERE actor = :actor + AND COALESCE(input ->> 'semantic_operation_type', operation_type) + = :semantic_operation_type + AND input ->> 'candidate_fingerprint' + = ANY(CAST(:candidate_fingerprints AS text[])) + """ + ), + { + "actor": RECONCILIATION_ACTOR, + "semantic_operation_type": RECONCILIATION_WORK_ITEM_OPERATION, + "candidate_fingerprints": candidate_fingerprints, + }, + ) + persisted_fingerprints = { + str(row[0]) for row in rows_result.fetchall() if row[0] + } repository_readback_count = len(expected_fingerprints & persisted_fingerprints) summary_output = build_reconciliation_summary_payload( matrix, diff --git a/apps/api/src/main.py b/apps/api/src/main.py index f9ca12fce..ed734b331 100644 --- a/apps/api/src/main.py +++ b/apps/api/src/main.py @@ -191,6 +191,7 @@ def _resolve_request_project_context(request: Request) -> tuple[str | None, str] return None, "request.project_id.missing" + # ============================================================================= # Sentry SDK Initialization (Error Tracking - 補強 SignOz) # Self-Hosted @ 192.168.0.110 @@ -274,6 +275,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: # AwoooP Phase 2.4 (2026-05-04 ogt): 設定 startup handler 的 project_id context # asyncio.create_task() 自動繼承父任務的 ContextVar → 31 個 background loop 全部標記為 awoooi from src.core.context import PROJECT_ID + PROJECT_ID.set("awoooi") # Startup @@ -345,6 +347,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: # ADR-015: MCP Provider 註冊 (DI 模式) from src.plugins.mcp.providers import register_all_providers + register_all_providers() logger.info("mcp_providers_registered") @@ -352,6 +355,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: # 2026-04-16 ogt + Claude Sonnet 4.6: 修復 sensors=0/0 根因 — init 從未在 startup 被呼叫 try: from src.services.mcp_tool_registry import init_mcp_tool_registry + await init_mcp_tool_registry() logger.info("mcp_tool_registry_initialized") except Exception as e: @@ -367,7 +371,9 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: ) logger.info("telegram_heartbeat_monitor_started", interval_minutes=30) else: - logger.warning("telegram_heartbeat_monitor_skipped", reason="OPENCLAW_TG_BOT_TOKEN not set") + logger.warning( + "telegram_heartbeat_monitor_skipped", reason="OPENCLAW_TG_BOT_TOKEN not set" + ) # Reboot Recovery: Warm-up Redis Working Memory from PostgreSQL # 2026-04-05 ogt: 重開機後 Redis 清空，從 DB restore 未解決的 incidents @@ -392,10 +398,12 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: async with get_db_context() as db: result = await db.execute( select(IncidentRecord).where( - IncidentRecord.status.in_([ - IncidentStatus.INVESTIGATING, - IncidentStatus.MITIGATING, - ]) + IncidentRecord.status.in_( + [ + IncidentStatus.INVESTIGATING, + IncidentStatus.MITIGATING, + ] + ) ) ) records = result.scalars().all() @@ -446,6 +454,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: # 必須在 embedding indexing 之前，確保 playbook 表有資料 try: from src.services.playbook_seed_service import seed_playbooks_from_rules + schedule_api_background_task(seed_playbooks_from_rules()) logger.info("playbook_seed_scheduled") except Exception as e: @@ -456,6 +465,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: # 2026-04-15 ogt + Claude Sonnet 4.6（亞太）: Phase 3.5 AI 學習成果持久化 try: from src.repositories.playbook_repository import get_playbook_repository + schedule_api_background_task(get_playbook_repository().backfill_redis_to_pg()) logger.info("playbook_pg_backfill_scheduled") except Exception as e: @@ -465,6 +475,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: from src.services.playbook_embedding_service import ( ensure_playbook_embeddings_indexed, ) + schedule_api_background_task(ensure_playbook_embeddings_indexed()) logger.info("playbook_embedding_indexing_scheduled") except Exception as e: @@ -486,6 +497,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: # dedup TTL 10 分鐘過期後，ready decisions 就沒有補送機制 → 長期卡在 ready 無人審核 try: from src.services.decision_manager import get_decision_manager + schedule_api_background_task(get_decision_manager().resend_stale_ready_tokens()) logger.info("stale_ready_tokens_resend_scheduled") except Exception as e: @@ -497,6 +509,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: # Sweeper 定期掃描無 decision token 的 INVESTIGATING incidents → 背景觸發 try: from src.jobs.incident_analysis_sweeper import run_incident_analysis_sweeper + schedule_api_background_task(run_incident_analysis_sweeper()) logger.info("incident_analysis_sweeper_scheduled", interval_sec=90) except Exception as e: @@ -507,6 +520,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: # 解開 8 張 0 writer 表的第一個 (asset_inventory / asset_discovery_run / asset_coverage_snapshot) try: from src.jobs.asset_scanner_job import run_asset_scanner_loop + schedule_api_background_task(run_asset_scanner_loop()) logger.info("asset_scanner_loop_scheduled", interval_sec=3600) except Exception as e: @@ -517,6 +531,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: # 解鎖 E3 Hermes 自動建規則: AI 需要 alert_rule_catalog 作為 baseline 才能提案修正 try: from src.jobs.rule_catalog_sync_job import run_rule_catalog_sync_loop + schedule_api_background_task(run_rule_catalog_sync_loop()) logger.info("rule_catalog_sync_loop_scheduled", interval_sec=3600) except Exception as e: @@ -527,6 +542,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: # 解鎖: Phase 4 Holt-Winters 預測有歷史資料 / 容量趨勢分析 try: from src.jobs.capacity_scanner_job import run_capacity_scanner_loop + schedule_api_background_task(run_capacity_scanner_loop()) logger.info("capacity_scanner_loop_scheduled", daily_trigger_hour_taipei=2) except Exception as e: @@ -537,6 +553,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: # MVP: secret_rotated 真實檢查,其他 6 維占位 'unknown',後續 agent 補 try: from src.jobs.compliance_scanner_job import run_compliance_scanner_loop + schedule_api_background_task(run_compliance_scanner_loop()) logger.info("compliance_scanner_loop_scheduled", daily_trigger_hour_taipei=3) except Exception as e: @@ -546,6 +563,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: # 消費 signals:aider:events stream → 建 incident + 寫 aider_events 表 try: from src.workers.aider_event_processor import run_aider_event_processor_loop + logger.info("aider_event_processor: starting Redis stream consumer") schedule_api_background_task(run_aider_event_processor_loop()) except Exception as e: @@ -556,6 +574,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: # 依據: Prometheus targets / alert_rule_catalog labels / knowledge_entries 覆蓋 try: from src.jobs.coverage_evaluator_job import run_coverage_evaluator_loop + schedule_api_background_task(run_coverage_evaluator_loop()) logger.info("coverage_evaluator_loop_scheduled", interval_sec=3600) except Exception as e: @@ -566,6 +585,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: # 解鎖 E3 Hermes: noise_rate > 0.5 的 rule 可被 AI 提案 deprecate try: from src.jobs.rule_stats_updater_job import run_rule_stats_updater_loop + schedule_api_background_task(run_rule_stats_updater_loop()) logger.info("rule_stats_updater_loop_scheduled", interval_sec=3600) except Exception as e: @@ -576,35 +596,26 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: # 解鎖: 資產變化歷史 (added/removed/lifecycle_changed),AI 可追蹤集群演進 try: from src.jobs.asset_change_tracker_job import run_asset_change_tracker_loop + schedule_api_background_task(run_asset_change_tracker_loop()) logger.info("asset_change_tracker_loop_scheduled", interval_sec=3600) except Exception as e: logger.warning("asset_change_tracker_loop_schedule_failed", error=str(e)) - # AIA-P0-006: reconcile the declared all-asset catalog with live ADR-090 - # inventory and project added/removed/drift capability gaps as durable work items. - try: - from src.jobs.asset_capability_reconciliation_job import ( - run_asset_capability_reconciliation_loop, - ) - - schedule_api_background_task(run_asset_capability_reconciliation_loop()) - logger.info( - "asset_capability_reconciliation_loop_scheduled", - daily_trigger_hour_taipei=3, - daily_trigger_minute_taipei=30, - ) - except Exception as e: - logger.warning( - "asset_capability_reconciliation_loop_schedule_failed", - error=str(e), - ) + # AIA-P0-006 has one runtime owner. Production API pods reserve their DB + # budget for requests, so the singleton signal worker owns reconciliation. + logger.info( + "asset_capability_reconciliation_owner_delegated", + owner="signal_worker", + reason="api_db_pool_reserved_for_serving_requests", + ) # ADR-090 § Hermes Rule Quality Advisor (2026-04-19 ogt + Claude Opus 4.7 Asia/Taipei) # 每日 04:00 Taipei 分析 alert_rule_catalog.noise_rate,對高噪音規則推 Telegram 建議 # 統帥鐵律: AI 只推建議不自動改 review_status,人工決策 deprecate try: from src.jobs.hermes_rule_quality_job import run_hermes_rule_quality_loop + schedule_api_background_task(run_hermes_rule_quality_loop()) logger.info("hermes_rule_quality_loop_scheduled", daily_trigger_hour_taipei=4) except Exception as e: @@ -615,6 +626,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: # 高風險 host 寫 aol(capacity_recommendation) + Telegram 建議 try: from src.jobs.capacity_forecaster_job import run_capacity_forecaster_loop + schedule_api_background_task(run_capacity_forecaster_loop()) logger.info("capacity_forecaster_loop_scheduled", daily_trigger_hour_taipei=5) except Exception as e: @@ -628,6 +640,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: run_monthly_report_loop, run_weekly_report_loop, ) + schedule_api_background_task(run_daily_report_loop()) schedule_api_background_task(run_weekly_report_loop()) schedule_api_background_task(run_monthly_report_loop()) @@ -644,6 +657,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: # 確保 PENDING approval 超過 48h 後觸發 resolve_incident → KM 學習鏈閉環 try: from src.jobs.approval_timeout_resolver import run_approval_timeout_resolver + schedule_api_background_task(run_approval_timeout_resolver()) logger.info("approval_timeout_resolver_scheduled", interval_sec=3600) except Exception as e: @@ -657,6 +671,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: from src.jobs.incident_lifecycle_reconciler import ( INTERVAL_SECONDS as INCIDENT_LIFECYCLE_RECONCILER_INTERVAL, ) + logger.info( "incident_lifecycle_reconciler_owner_delegated", interval_sec=INCIDENT_LIFECYCLE_RECONCILER_INTERVAL, @@ -676,6 +691,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: from src.jobs.awooop_ansible_candidate_backfill_job import ( run_awooop_ansible_candidate_backfill_loop, ) + scheduled_task = schedule_api_background_task( run_awooop_ansible_candidate_backfill_loop() ) @@ -687,7 +703,9 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: interval_seconds=settings.AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_INTERVAL_SECONDS, ) except Exception as e: - logger.warning("awooop_ansible_candidate_backfill_worker_schedule_failed", error=str(e)) + logger.warning( + "awooop_ansible_candidate_backfill_worker_schedule_failed", error=str(e) + ) # AwoooP Ansible check-mode worker. # 先執行 ansible-playbook --check --diff 並回寫 automation_operation_log； @@ -696,6 +714,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: from src.jobs.awooop_ansible_check_mode_job import ( run_awooop_ansible_check_mode_loop, ) + scheduled_task = schedule_api_background_task( run_awooop_ansible_check_mode_loop() ) @@ -713,6 +732,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: # 2026-04-15 ogt + Claude Sonnet 4.6（亞太）: Phase 3 初始建立 try: from src.services.playbook_evolver import run_evolver_loop + schedule_api_background_task(run_evolver_loop()) logger.info("evolver_loop_scheduled", interval_sec=86400) except Exception as e: @@ -723,18 +743,22 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: from src.jobs.playbook_generation_governance_job import ( run_playbook_generation_governance_loop, ) + schedule_api_background_task(run_playbook_generation_governance_loop()) logger.info( "playbook_generation_governance_loop_scheduled", interval_sec=settings.PLAYBOOK_DRAFT_GOVERNANCE_INTERVAL_SECONDS, ) except Exception as e: - logger.warning("playbook_generation_governance_loop_schedule_failed", error=str(e)) + logger.warning( + "playbook_generation_governance_loop_schedule_failed", error=str(e) + ) # ADR-083 Phase 3: 知識遺忘 Job（每日）— 30d 未引用 KB entry 標記 archived # 2026-04-15 ogt + Claude Sonnet 4.6（亞太）: Phase 3 初始建立 try: from src.jobs.knowledge_decay_job import run_knowledge_decay_loop + schedule_api_background_task(run_knowledge_decay_loop()) logger.info("knowledge_decay_loop_scheduled", interval_sec=86400) except Exception as e: @@ -745,6 +769,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: # Feature Flag: ENABLE_KM_BACKFILL_RECONCILER=false 停用（回滾用） try: from src.jobs.km_backfill_reconciler_job import run_km_backfill_reconciler_loop + schedule_api_background_task(run_km_backfill_reconciler_loop()) logger.info("km_backfill_reconciler_loop_scheduled", interval_sec=300) except Exception as e: @@ -756,6 +781,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: # ADR-091 Task T2 try: from src.jobs.aol_to_catalog_writeback_job import run_aol_writeback_loop + schedule_api_background_task(run_aol_writeback_loop()) logger.info("aol_to_catalog_writeback_loop_scheduled", interval_sec=3600) except Exception as e: @@ -771,28 +797,48 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: import asyncio as _asyncio from src.jobs.kb_rot_cleaner import get_kb_rot_cleaner + while True: try: now = now_taipei() # 計算下次月初 3 點（台北時間） if now.day == 1 and now.hour < 3: - next_run = now.replace(hour=3, minute=0, second=0, microsecond=0) + next_run = now.replace( + hour=3, minute=0, second=0, microsecond=0 + ) elif now.month == 12: next_run = now.replace( - year=now.year + 1, month=1, day=1, - hour=3, minute=0, second=0, microsecond=0, + year=now.year + 1, + month=1, + day=1, + hour=3, + minute=0, + second=0, + microsecond=0, ) else: next_run = now.replace( - month=now.month + 1, day=1, - hour=3, minute=0, second=0, microsecond=0, + month=now.month + 1, + day=1, + hour=3, + minute=0, + second=0, + microsecond=0, ) sleep_sec = (next_run - now).total_seconds() - logger.info("kb_rot_cleaner_next_run", next_run=next_run.isoformat(), sleep_sec=int(sleep_sec)) + logger.info( + "kb_rot_cleaner_next_run", + next_run=next_run.isoformat(), + sleep_sec=int(sleep_sec), + ) await _asyncio.sleep(sleep_sec) try: result = await get_kb_rot_cleaner().run() - logger.info("kb_rot_cleaner_completed", stale_count=result.stale_count, total=result.total_scanned) + logger.info( + "kb_rot_cleaner_completed", + stale_count=result.stale_count, + total=result.total_scanned, + ) except Exception as _e: logger.exception("kb_rot_cleaner_failed", error=str(_e)) except _asyncio.CancelledError: @@ -810,6 +856,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: # 2026-04-15 ogt + Claude Sonnet 4.6（亞太）: Phase 3 初始建立 try: from src.services.finetune_exporter import run_finetune_export_loop + schedule_api_background_task(run_finetune_export_loop()) logger.info("finetune_export_loop_scheduled", interval_sec=604800) except Exception as e: @@ -821,6 +868,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: # 2026-04-15 ogt + Claude Sonnet 4.6（亞太）: Phase 4 初始建立 try: from src.services.proactive_inspector import run_proactive_inspector_loop + schedule_api_background_task(run_proactive_inspector_loop()) logger.info("proactive_inspector_loop_scheduled", interval_sec=300) except Exception as e: @@ -830,6 +878,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: # 2026-04-15 ogt + Claude Sonnet 4.6（亞太）: Phase 6 初始建立 try: from src.jobs.offline_replay_service import run_offline_replay_loop + schedule_api_background_task(run_offline_replay_loop()) logger.info("offline_replay_loop_scheduled", interval_sec=604800) except Exception as e: @@ -839,6 +888,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: # MASTER §1.1：系統必須能感知自身故障，W-1 SLO + W-2 Telegram靜默 + W-3 飛輪成功率 try: from src.jobs.ai_slo_watchdog_job import run_ai_slo_watchdog_loop + schedule_api_background_task(run_ai_slo_watchdog_loop()) logger.info("ai_slo_watchdog_scheduled", interval_sec=900) except Exception as e: @@ -848,6 +898,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: # MASTER P2.2：trust_drift / knowledge_degradation / llm_hallucination / execution_blast_radius try: from src.services.governance_agent import run_governance_loop + schedule_api_background_task(run_governance_loop()) logger.info("governance_agent_scheduled", interval_sec=3600) except Exception as e: @@ -856,6 +907,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: # 2026-05-03 ogt + Claude Sonnet 4.6（亞太）: GovernanceDispatcher Wave 2E（每 30s poll） try: from src.services.governance_dispatcher import run_governance_dispatcher_loop + schedule_api_background_task(run_governance_dispatcher_loop()) logger.info("governance_dispatcher_scheduled", interval_sec=30) except Exception as e: @@ -866,6 +918,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: # 只產生 REVIEW 草稿並停在 owner review，不直接批准或發布 KM。 try: from src.jobs.hermes_kb_growth_worker import run_hermes_kb_growth_loop + schedule_api_background_task(run_hermes_kb_growth_loop()) logger.info("hermes_kb_growth_worker_scheduled", interval_sec=300) except Exception as e: @@ -893,6 +946,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: try: from src.core.redis_client import get_redis from src.services.failover_alerter import configure_alerter + configure_alerter(get_redis()) logger.info("failover_alerter_configured") except Exception as _alerter_err: @@ -909,8 +963,10 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: # 探測 5 Provider（ollama/ollama_local/gemini/claude/openclaw_nemo）版本 # 寫入 ai_provider_version_history；版本變更時 log warning，P3.2.3 alerter 後續整合 try: + async def _run_model_version_tracker_loop() -> None: from src.services.model_version_tracker import get_model_version_tracker + tracker = get_model_version_tracker() while True: try: @@ -924,7 +980,9 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: except asyncio.CancelledError: break except Exception as _loop_err: - logger.exception("model_version_tracker_loop_error", error=str(_loop_err)) + logger.exception( + "model_version_tracker_loop_error", error=str(_loop_err) + ) await asyncio.sleep(60) # 錯誤後 1 分鐘重試 schedule_api_background_task(_run_model_version_tracker_loop()) @@ -937,6 +995,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: # Shadow mode：is_shadow=True，0 user-visible response，0 destructive tool call try: from src.workers.platform_worker import start_platform_worker + if api_background_tasks_enabled: await start_platform_worker() logger.info("platform_worker_started", mode="shadow") @@ -955,6 +1014,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: # 必須在 Redis pool 關閉之前停止（recovery service 可能仍在寫 Redis） try: from src.services.ollama_auto_recovery import get_ollama_auto_recovery_service + await get_ollama_auto_recovery_service().stop() logger.info("ollama_failover_system_stopped") except Exception as e: @@ -965,6 +1025,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: # 確保 _verify_and_learn / runbook_generator 寫入不被 SIGTERM cancel try: from src.services.auto_repair_service import get_auto_repair_service + _repair_svc = get_auto_repair_service() if hasattr(_repair_svc, "drain_pending_tasks"): _drain_result = await _repair_svc.drain_pending_tasks(timeout=60.0) @@ -975,6 +1036,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: # AwoooP Phase 4: Platform Worker 優雅停機（2026-05-04 ogt） try: from src.workers.platform_worker import stop_platform_worker + await stop_platform_worker() logger.info("platform_worker_stopped") except Exception as e: @@ -991,6 +1053,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: await telegram_gw.close() # Phase 33: Close RAG Service httpx client (ADR-067) from src.services.knowledge_rag_service import get_knowledge_rag_service + await get_knowledge_rag_service().close() # Phase 5: Close HTTP Clients (統帥鐵律: 連線池回收) await close_all_http_clients() @@ -1178,7 +1241,9 @@ async def http_exception_handler(_request: Request, exc: HTTPException) -> JSONR This is critical for P1-1 fail-closed evidence; without it, all HTTPException is swallowed by the generic exception handler and downgraded to 500. """ - return JSONResponse(status_code=exc.status_code, content={"detail": exc.detail}, headers=exc.headers) + return JSONResponse( + status_code=exc.status_code, content={"detail": exc.detail}, headers=exc.headers + ) @app.exception_handler(Exception) @@ -1219,15 +1284,25 @@ app.include_router(csrf_v1.router, prefix="/api/v1", tags=["Security"]) # Phase app.include_router(dashboard_v1.router, prefix="/api/v1", tags=["Dashboard"]) app.include_router(approvals_v1.router, prefix="/api/v1", tags=["HITL Approvals"]) app.include_router(ai_v1.router, prefix="/api/v1", tags=["AI Decision"]) -app.include_router(ai_governance_v1.router, prefix="/api/v1", tags=["AI Governance"]) # 2026-05-02: /governance 頁面 -app.include_router(ai_slo_v1.router, prefix="/api/v1", tags=["AI SLO"]) # Phase 6 ADR-087 -app.include_router(aiops_kpi_v1.router, prefix="/api/v1", tags=["AIOps KPI"]) # ADR-090 § Phase 7 Dashboard -app.include_router(aiops_timeline_v1.router, prefix="/api/v1", tags=["AIOps Timeline"]) # 2026-04-27 Wave8-X3 B4 +app.include_router( + ai_governance_v1.router, prefix="/api/v1", tags=["AI Governance"] +) # 2026-05-02: /governance 頁面 +app.include_router( + ai_slo_v1.router, prefix="/api/v1", tags=["AI SLO"] +) # Phase 6 ADR-087 +app.include_router( + aiops_kpi_v1.router, prefix="/api/v1", tags=["AIOps KPI"] +) # ADR-090 § Phase 7 Dashboard +app.include_router( + aiops_timeline_v1.router, prefix="/api/v1", tags=["AIOps Timeline"] +) # 2026-04-27 Wave8-X3 B4 app.include_router(webhooks_v1.router, prefix="/api/v1", tags=["Webhooks"]) app.include_router(timeline_v1.router, prefix="/api/v1", tags=["Timeline"]) app.include_router(audit_logs_v1.router, prefix="/api/v1", tags=["Audit Logs"]) # 2026-04-09 Claude Sonnet 4.6: alert_operation_log 查詢 API (Sprint 5.2) -app.include_router(alert_operation_logs_v1.router, prefix="/api/v1", tags=["Alert Operation Logs"]) +app.include_router( + alert_operation_logs_v1.router, prefix="/api/v1", tags=["Alert Operation Logs"] +) app.include_router( aider_events_v1.router, prefix="/api/v1", @@ -1310,7 +1385,9 @@ app.include_router( notifications.router, prefix="/api/v1/notifications", tags=["Notifications"] ) # AwoooP Phase 4 (2026-05-04 ogt): Platform Shell — Shadow Mode Run API -app.include_router(platform_v1.router, prefix="/api/v1/platform", tags=["AwoooP Platform"]) +app.include_router( + platform_v1.router, prefix="/api/v1/platform", tags=["AwoooP Platform"] +) # ============================================================================= 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_controlled_dispatch_ledger.py b/apps/api/src/services/agent99_controlled_dispatch_ledger.py index af334421b..e64ac799a 100644 --- a/apps/api/src/services/agent99_controlled_dispatch_ledger.py +++ b/apps/api/src/services/agent99_controlled_dispatch_ledger.py @@ -1752,6 +1752,200 @@ class PostgresAgent99DispatchLedger: "runtime_closure_verified": False, } + async def record_outcome_timeout_no_write_terminal( + self, + *, + identity: Agent99DispatchIdentity, + ) -> dict[str, Any]: + """Fail an expired run without replaying transport or claiming closure.""" + + now = _utc_now_naive() + error_code = "E-AGENT99-OUTCOME-TIMEOUT" + try: + async with get_db_context(identity.project_id) as db: + current_result = await db.execute( + select( + AwoooPRunState.state, + AwoooPRunState.error_detail, + AwoooPRunState.timeout_at, + ) + .where( + AwoooPRunState.run_id == identity.run_id, + AwoooPRunState.project_id == identity.project_id, + ) + .with_for_update() + ) + current = current_result.one_or_none() + envelope = _parse_envelope( + current.error_detail if current is not None else None + ) + identity_matches = bool( + isinstance(envelope, dict) + and _stage_identity_matches( + identity, + envelope.get("identity") + if isinstance(envelope.get("identity"), dict) + else {}, + ) + ) + if ( + current is not None + and current.state == "failed" + and identity_matches + and envelope.get("outcome_timeout_no_write_terminal") is True + ): + return { + **envelope, + "status": "outcome_timeout_no_write_terminal_idempotent", + "receipt_persisted": True, + "runtime_closure_verified": False, + } + if current is None or not identity_matches: + return { + "status": "outcome_timeout_run_identity_invalid_fail_closed", + "receipt_persisted": False, + "runtime_closure_verified": False, + } + timeout_at = current.timeout_at + if ( + current.state != "waiting_tool" + or timeout_at is None + or timeout_at > now + ): + return { + "status": "outcome_timeout_not_due_no_write", + "receipt_persisted": False, + "runtime_closure_verified": False, + } + + prior_controlled_apply_authorized = bool( + envelope.get("controlled_apply_authorized") is True + ) + prior_runtime_execution_authorized = bool( + envelope.get("runtime_execution_authorized") is True + ) + prior_runtime_execution_attempted = bool( + envelope.get("runtime_execution_attempted") is True + ) + terminal_receipt = { + "schema_version": "agent99_outcome_timeout_no_write_v1", + "status": "authenticated_outcome_timeout", + **_stage_identity(identity), + "timeout_at": timeout_at.isoformat(), + "terminal_at": now.isoformat(), + "terminalizer_runtime_write_performed": False, + "original_runtime_write_status": ( + "unknown_without_authenticated_outcome" + ), + "transport_replayed": False, + "incident_resolution_authorized": False, + "learning_writeback_authorized": False, + "telegram_authorized": False, + "stores_raw_evidence": False, + } + envelope.update({ + "status": "outcome_timeout_no_write_terminal", + "run_terminal_state": "failed", + "outcome_timeout_no_write_terminal": True, + "prior_controlled_apply_authorized": ( + prior_controlled_apply_authorized + ), + "prior_runtime_execution_authorized": ( + prior_runtime_execution_authorized + ), + "runtime_execution_attempted": ( + prior_runtime_execution_attempted + ), + "controlled_apply_authorized": False, + "runtime_execution_authorized": False, + "terminalizer_runtime_write_performed": False, + "outcome_runtime_write_status": ( + "unknown_without_authenticated_outcome" + ), + "transport_replayed": False, + "automation_execution_success": False, + "post_verifier_passed": False, + "runtime_closure_verified": False, + "closure_complete": False, + "incident_resolution_authorized": False, + "safe_next_action": ( + "await_new_source_recurrence_after_agent99_relay_recovery" + ), + "verifier": { + **_stage_identity(identity), + "step_seq": 2, + "status": "failed_authenticated_outcome_timeout", + "post_verifier_passed": False, + "receipt": terminal_receipt, + }, + "learning_writeback": { + **_stage_identity(identity), + "step_seq": 3, + "status": "not_applicable_outcome_timeout_no_write", + "required_receipts": [], + "receipt_refs": {}, + "stores_raw_evidence": False, + }, + }) + envelope_json = _stable_json(envelope) + updated = await db.execute( + update(AwoooPRunState) + .where( + AwoooPRunState.run_id == identity.run_id, + AwoooPRunState.project_id == identity.project_id, + AwoooPRunState.state == "waiting_tool", + AwoooPRunState.timeout_at <= now, + ) + .values( + state="failed", + output_sha256=_sha256(envelope_json), + error_code=error_code, + error_detail=envelope_json, + heartbeat_at=now, + completed_at=now, + lease_until=None, + next_attempt_at=None, + worker_id=None, + ) + .returning(AwoooPRunState.run_id) + ) + persisted = updated.scalar_one_or_none() is not None + if persisted: + await db.execute( + update(AwoooPRunStepJournal) + .where( + AwoooPRunStepJournal.run_id == identity.run_id, + AwoooPRunStepJournal.step_seq.in_([2, 3]), + AwoooPRunStepJournal.result_status == "pending", + ) + .values( + output_hash=_sha256(_stable_json(terminal_receipt)), + compensation_json=terminal_receipt, + result_status="failed", + error_code=error_code, + was_blocked=True, + block_reason="authenticated_outcome_timeout_no_write", + completed_at=now, + ) + ) + except Exception as exc: + logger.warning( + "agent99_outcome_timeout_terminal_failed_fail_closed", + incident_id=identity.incident_id, + run_id=str(identity.run_id), + error=str(exc), + ) + return { + "status": "outcome_timeout_terminal_persistence_failed", + "receipt_persisted": False, + "runtime_closure_verified": False, + } + return { + **envelope, + "receipt_persisted": persisted, + "runtime_closure_verified": False, + } + async def record_learning_writeback( self, *, @@ -2236,6 +2430,7 @@ class PostgresAgent99DispatchLedger: AwoooPRunState.run_id, AwoooPRunState.state, AwoooPRunState.error_detail, + AwoooPRunState.timeout_at, ) .where( AwoooPRunState.project_id == (project_id or "awoooi"), @@ -2309,6 +2504,7 @@ class PostgresAgent99DispatchLedger: "identity": identity, "run_state": str(row.state), "receipt": envelope, + "timeout_at": getattr(row, "timeout_at", None), }) return items 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/ai_automation_asset_capability_matrix.py b/apps/api/src/services/ai_automation_asset_capability_matrix.py index 27549f312..2f35dba01 100644 --- a/apps/api/src/services/ai_automation_asset_capability_matrix.py +++ b/apps/api/src/services/ai_automation_asset_capability_matrix.py @@ -631,6 +631,7 @@ def _receipt_controls( fresh = bool( row and str(row.get("status") or "") == "success" + and output.get("closed") is True and _is_fresh( created_at, now=now, 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/src/workers/signal_worker.py b/apps/api/src/workers/signal_worker.py index ef9124ded..55b84a433 100644 --- a/apps/api/src/workers/signal_worker.py +++ b/apps/api/src/workers/signal_worker.py @@ -71,6 +71,7 @@ GRACEFUL_SHUTDOWN_TIMEOUT_S = 75 # K8s terminationGracePeriodSeconds: 90 # Signal Worker # ============================================================================= + class SignalWorker: """ Redis Streams 訊號消費者 @@ -366,7 +367,9 @@ class SignalWorker: span.set_attribute("messaging.destination", STREAM_KEY) span.set_attribute("messaging.message_id", message_id) span.set_attribute("signal.source", data.get("source", "unknown")) - span.set_attribute("signal.alert_name", data.get("alert_name", "unknown")) + span.set_attribute( + "signal.alert_name", data.get("alert_name", "unknown") + ) span.set_attribute("signal.severity", data.get("severity", "unknown")) logger.info( @@ -400,7 +403,9 @@ class SignalWorker: severity=incident.severity.value, signal_count=len(incident.signals), affected_services=incident.affected_services, - persisted_to_pg=getattr(incident, "persisted_to_pg", False), # 2026-04-01 ogt: BrainIncident 無此欄位 (ADR-046 P2-01) + persisted_to_pg=getattr( + incident, "persisted_to_pg", False + ), # 2026-04-01 ogt: BrainIncident 無此欄位 (ADR-046 P2-01) ) try: from src.services.signal_observation_service import ( @@ -505,6 +510,7 @@ def get_signal_worker() -> SignalWorker: # Standalone Entry Point (for K8s Worker Deployment) # ============================================================================= + async def _write_health_files() -> None: """ Write health check files for K8s probes. @@ -542,10 +548,7 @@ async def _heartbeat_loop(shutdown_event: asyncio.Event) -> None: # 等待下次心跳或收到關閉信號 try: - await asyncio.wait_for( - shutdown_event.wait(), - timeout=HEARTBEAT_INTERVAL - ) + await asyncio.wait_for(shutdown_event.wait(), timeout=HEARTBEAT_INTERVAL) break # 收到關閉信號 except TimeoutError: continue # 超時，繼續下次心跳 @@ -575,7 +578,9 @@ async def _main() -> None: "signal_worker_standalone_starting", environment=settings.ENVIRONMENT, redis_url=settings.REDIS_URL.split("@")[-1] if settings.REDIS_URL else "N/A", - database_url=settings.DATABASE_URL.split("@")[-1] if settings.DATABASE_URL else "N/A", + database_url=settings.DATABASE_URL.split("@")[-1] + if settings.DATABASE_URL + else "N/A", ) # Initialize Redis (API pool + Worker 專屬長連線池) @@ -695,6 +700,9 @@ async def _main() -> None: telegram_dispatch_performed=False, ) if settings.ENABLE_SECURITY_CONTROL_PLANE_MAINTENANCE_WORKER: + from src.jobs.asset_capability_reconciliation_job import ( + run_asset_capability_reconciliation_loop, + ) from src.jobs.asset_change_tracker_job import run_asset_change_tracker_loop from src.jobs.asset_scanner_job import run_asset_scanner_loop from src.jobs.compliance_scanner_job import run_compliance_scanner_loop @@ -707,6 +715,10 @@ async def _main() -> None: ("run_compliance_scanner_loop", run_compliance_scanner_loop), ("run_coverage_evaluator_loop", run_coverage_evaluator_loop), ("run_asset_change_tracker_loop", run_asset_change_tracker_loop), + ( + "run_asset_capability_reconciliation_loop", + run_asset_capability_reconciliation_loop, + ), ) security_maintenance_tasks = [ asyncio.create_task(loop(), name=task_name) @@ -791,12 +803,8 @@ async def _main() -> None: "signal_worker_agent99_controlled_dispatch_reconciler_skipped_fail_closed", owner="signal_worker", reason="authenticated_relay_not_configured", - relay_url_configured=bool( - settings.AGENT99_SRE_ALERT_RELAY_URL.strip() - ), - relay_token_configured=bool( - settings.AGENT99_SRE_ALERT_RELAY_TOKEN.strip() - ), + relay_url_configured=bool(settings.AGENT99_SRE_ALERT_RELAY_URL.strip()), + relay_token_configured=bool(settings.AGENT99_SRE_ALERT_RELAY_TOKEN.strip()), ) # Setup graceful shutdown shutdown_event = asyncio.Event() @@ -873,6 +881,7 @@ async def _main() -> None: # Remove health files from pathlib import Path + Path("/tmp/worker-healthy").unlink(missing_ok=True) Path("/tmp/worker-ready").unlink(missing_ok=True) diff --git a/apps/api/tests/test_agent99_controlled_dispatch_reconciler_job.py b/apps/api/tests/test_agent99_controlled_dispatch_reconciler_job.py index b230d7722..fc5b707a5 100644 --- a/apps/api/tests/test_agent99_controlled_dispatch_reconciler_job.py +++ b/apps/api/tests/test_agent99_controlled_dispatch_reconciler_job.py @@ -101,6 +101,7 @@ async def test_reconciler_consumes_authenticated_outcome_and_closes_same_run( "status_reconcile_requested": 0, "status_reconcile_pending": 0, "external_no_write_reconciled": 0, + "outcome_timeout_terminalized": 0, "verifier_written": 1, "closed": 1, } @@ -130,6 +131,7 @@ async def test_reconciler_does_not_finalize_without_outcome(monkeypatch) -> None "status_reconcile_requested": 0, "status_reconcile_pending": 0, "external_no_write_reconciled": 0, + "outcome_timeout_terminalized": 0, "verifier_written": 0, "closed": 0, } diff --git a/apps/api/tests/test_agent99_same_run_reconcile.py b/apps/api/tests/test_agent99_same_run_reconcile.py index 44d246dbb..d987aba8f 100644 --- a/apps/api/tests/test_agent99_same_run_reconcile.py +++ b/apps/api/tests/test_agent99_same_run_reconcile.py @@ -2,6 +2,7 @@ from __future__ import annotations import json from copy import deepcopy +from datetime import UTC, datetime, timedelta from pathlib import Path from types import SimpleNamespace from unittest.mock import AsyncMock @@ -812,6 +813,184 @@ async def test_external_no_write_terminal_rejects_malformed_source_counts( assert result["receipt_persisted"] is False +def test_dispatch_timeout_elapsed_handles_naive_and_aware_datetimes() -> None: + past = datetime.now(UTC) - timedelta(minutes=1) + future = datetime.now(UTC) + timedelta(minutes=1) + + assert job._dispatch_timeout_elapsed(past) is True + assert job._dispatch_timeout_elapsed(past.replace(tzinfo=None)) is True + assert job._dispatch_timeout_elapsed(future) is False + assert job._dispatch_timeout_elapsed(None) is False + + +@pytest.mark.asyncio +async def test_reconciler_terminalizes_expired_run_without_outcome( + monkeypatch, +) -> None: + identity = build_agent99_dispatch_identity( + project_id="awoooi", + incident_id="INC-BRR-EXPIRED", + source_fingerprint="backup-restore:" + "b" * 64, + route_id="agent99:backup_health:Status", + work_item_id="agent99-dispatch:awoooi:INC-BRR-EXPIRED", + ) + + class TimeoutLedger(_Ledger): + def __init__(self) -> None: + super().__init__() + self.timeout_calls: list[object] = [] + + async def list_reconcilable(self, **_kwargs): # type: ignore[no-untyped-def] + return [{ + "identity": identity, + "receipt": self.list_receipt, + "timeout_at": datetime.now(UTC) - timedelta(minutes=1), + }] + + async def record_outcome_timeout_no_write_terminal( + self, + **kwargs, + ): # type: ignore[no-untyped-def] + self.timeout_calls.append(kwargs["identity"]) + return { + "status": "outcome_timeout_no_write_terminal", + "receipt_persisted": True, + "runtime_closure_verified": False, + } + + ledger = TimeoutLedger() + monkeypatch.setattr(job, "get_agent99_dispatch_ledger", lambda: ledger) + monkeypatch.setattr(job, "read_agent99_sre_outcome", lambda _run: None) + + result = await job.reconcile_agent99_controlled_dispatches_once() + + assert result["outcome_timeout_terminalized"] == 1 + assert result["verifier_written"] == 0 + assert ledger.timeout_calls == [identity] + + +@pytest.mark.asyncio +async def test_outcome_timeout_terminal_does_not_resolve_incident_or_replay( + monkeypatch, +) -> None: + envelope = build_agent99_dispatch_receipt_envelope( + identity=IDENTITY, + dispatch_receipt={ + "accepted": True, + "inbox_triggered": True, + "queue_accepted": True, + "dispatch_identity_matched": True, + "delivery_certainty": "delivered", + }, + controlled_apply_authorized=True, + ) + + class TimeoutDB: + def __init__(self) -> None: + self.call = 0 + self.statements: list[str] = [] + + async def execute(self, statement): # type: ignore[no-untyped-def] + self.call += 1 + self.statements.append(str(statement)) + if self.call == 1: + return _ScalarResult( + row=SimpleNamespace( + state="waiting_tool", + error_detail=json.dumps(envelope), + timeout_at=( + datetime.now(UTC).replace(tzinfo=None) + - timedelta(minutes=1) + ), + ) + ) + if self.call == 2: + return _ScalarResult(value=IDENTITY.run_id) + return _ScalarResult() + + db = TimeoutDB() + monkeypatch.setattr( + ledger_module, + "get_db_context", + lambda _project_id: _Context(db), + ) + + result = ( + await PostgresAgent99DispatchLedger() + .record_outcome_timeout_no_write_terminal(identity=IDENTITY) + ) + + assert result["receipt_persisted"] is True + assert result["run_terminal_state"] == "failed" + assert result["prior_controlled_apply_authorized"] is True + assert result["controlled_apply_authorized"] is False + assert result["terminalizer_runtime_write_performed"] is False + assert result["outcome_runtime_write_status"] == ( + "unknown_without_authenticated_outcome" + ) + assert result["transport_replayed"] is False + assert result["automation_execution_success"] is False + assert result["post_verifier_passed"] is False + assert result["runtime_closure_verified"] is False + assert result["incident_resolution_authorized"] is False + assert result["learning_writeback"]["status"] == ( + "not_applicable_outcome_timeout_no_write" + ) + sql = "\n".join(db.statements).lower() + assert "incident_records" not in sql + assert "alert_operation_logs" not in sql + assert "telegram" not in sql + + +@pytest.mark.asyncio +async def test_outcome_timeout_terminal_rejects_unexpired_run( + monkeypatch, +) -> None: + envelope = build_agent99_dispatch_receipt_envelope( + identity=IDENTITY, + dispatch_receipt={ + "accepted": True, + "inbox_triggered": True, + "queue_accepted": True, + "dispatch_identity_matched": True, + }, + controlled_apply_authorized=False, + ) + + class FutureTimeoutDB: + def __init__(self) -> None: + self.call = 0 + + async def execute(self, _statement): # type: ignore[no-untyped-def] + self.call += 1 + return _ScalarResult( + row=SimpleNamespace( + state="waiting_tool", + error_detail=json.dumps(envelope), + timeout_at=( + datetime.now(UTC).replace(tzinfo=None) + + timedelta(minutes=1) + ), + ) + ) + + db = FutureTimeoutDB() + monkeypatch.setattr( + ledger_module, + "get_db_context", + lambda _project_id: _Context(db), + ) + + result = ( + await PostgresAgent99DispatchLedger() + .record_outcome_timeout_no_write_terminal(identity=IDENTITY) + ) + + assert result["status"] == "outcome_timeout_not_due_no_write" + assert result["receipt_persisted"] is False + assert db.call == 1 + + def test_windows_relay_and_control_plane_enforce_same_run_no_write_contract() -> None: root = Path(__file__).resolve().parents[3] relay = (root / "agent99-sre-alert-relay.ps1").read_text(encoding="utf-8") 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_ai_automation_asset_capability_matrix.py b/apps/api/tests/test_ai_automation_asset_capability_matrix.py index e36e7d95f..ad2680394 100644 --- a/apps/api/tests/test_ai_automation_asset_capability_matrix.py +++ b/apps/api/tests/test_ai_automation_asset_capability_matrix.py @@ -1,12 +1,15 @@ from __future__ import annotations import asyncio +import inspect from datetime import UTC, datetime, timedelta from pathlib import Path from typing import Any +import src.jobs.asset_capability_reconciliation_job as reconciliation_job import src.services.ai_automation_asset_capability_matrix as matrix_service from src.jobs.asset_capability_reconciliation_job import ( + _persist_reconciliation, build_reconciliation_summary_payload, build_reconciliation_work_item_input, ) @@ -212,6 +215,7 @@ def test_live_asset_has_per_stage_covered_status_and_closes_after_receipt() -> N "candidate_count": 0, "repository_readback_count": 0, "repository_readback_verified": True, + "closed": True, }, } @@ -232,6 +236,38 @@ def test_live_asset_has_per_stage_covered_status_and_closes_after_receipt() -> N assert matrix["summary"]["gap_asset_count"] == 0 +def test_reconciliation_receipt_requires_closed_terminal() -> None: + now = datetime(2026, 7, 11, 12, 0, tzinfo=UTC) + initial = build_asset_capability_matrix( + _single_host_scope(), + live_assets=[_fully_covered_host(now)], + latest_run={"run_id": "run-1", "status": "success", "ended_at": now}, + repo_root=_REPO_ROOT, + generated_at=now, + ) + matrix = build_asset_capability_matrix( + _single_host_scope(), + live_assets=[_fully_covered_host(now)], + latest_run={"run_id": "run-1", "status": "success", "ended_at": now}, + reconciliation_receipt={ + "status": "success", + "created_at": now, + "output": { + "matrix_fingerprint": initial["matrix_fingerprint"], + "candidate_count": 0, + "repository_readback_count": 0, + "repository_readback_verified": True, + "closed": False, + }, + }, + repo_root=_REPO_ROOT, + generated_at=now, + ) + + assert matrix["controls"]["daily_reconciliation_receipt"] is False + assert matrix["closed"] is False + + def test_outdated_live_receipts_are_classified_stale_per_stage() -> None: now = datetime(2026, 7, 11, 12, 0, tzinfo=UTC) old = now - timedelta(hours=40) @@ -342,6 +378,51 @@ def test_reconciliation_work_item_uses_same_trace_and_run_as_summary() -> None: ) +def test_reconciliation_persistence_batches_candidate_queries_and_inserts() -> None: + source = inspect.getsource(_persist_reconciliation) + loop_start = source.index("for candidate_raw in candidates:") + loop_end = source.index("if insert_rows:") + + assert "ANY(CAST(:candidate_fingerprints AS text[]))" in source + assert "insert_rows" in source + assert "await db.execute" not in source[loop_start:loop_end] + assert "AND output ->> 'closed' = 'true'" in source + + +def test_reconciliation_loop_retries_degraded_result_before_daily_wait( + monkeypatch: Any, +) -> None: + triggered_by_values: list[str] = [] + sleep_values: list[int] = [] + + async def _fake_reconcile_once(*, triggered_by: str) -> dict[str, object]: + triggered_by_values.append(triggered_by) + if len(triggered_by_values) == 1: + return {"status": "degraded_no_write", "reason": "transient_db_pressure"} + return {"status": "success", "closed": True} + + async def _fake_sleep(seconds: int) -> None: + sleep_values.append(seconds) + if len(sleep_values) == 3: + raise asyncio.CancelledError + + monkeypatch.setattr(reconciliation_job, "reconcile_once", _fake_reconcile_once) + monkeypatch.setattr(reconciliation_job.asyncio, "sleep", _fake_sleep) + monkeypatch.setattr(reconciliation_job, "_FIRST_DELAY_SECONDS", 90) + monkeypatch.setattr(reconciliation_job, "_LOOP_BACKOFF_SECONDS", 1_800) + monkeypatch.setattr(reconciliation_job, "_seconds_until_next_trigger", lambda: 1) + + try: + asyncio.run(reconciliation_job.run_asset_capability_reconciliation_loop()) + except asyncio.CancelledError: + pass + else: + raise AssertionError("loop should stop at the test cancellation point") + + assert triggered_by_values == ["startup", "retry"] + assert sleep_values == [90, 1_800, 1] + + def test_priority_overlay_projects_aia_p0_006_runtime_controls() -> None: matrix = build_asset_capability_matrix( _single_host_scope(), diff --git a/apps/api/tests/test_config_url_validation.py b/apps/api/tests/test_config_url_validation.py index 27489672d..86269675b 100644 --- a/apps/api/tests/test_config_url_validation.py +++ b/apps/api/tests/test_config_url_validation.py @@ -172,3 +172,4 @@ def test_database_pool_budget_defaults_to_production_safe_values(): assert s.DATABASE_MAX_OVERFLOW == 0 assert s.DATABASE_POOL_TIMEOUT_SECONDS == 5.0 assert s.DATABASE_NULL_POOL is False + assert s.DATABASE_NULL_POOL_CONCURRENCY_LIMIT == 0 diff --git a/apps/api/tests/test_runtime_bootstrap_guards.py b/apps/api/tests/test_runtime_bootstrap_guards.py index d42a70e7f..60b6f52df 100644 --- a/apps/api/tests/test_runtime_bootstrap_guards.py +++ b/apps/api/tests/test_runtime_bootstrap_guards.py @@ -8,6 +8,7 @@ Runtime bootstrap guard tests. from __future__ import annotations +import asyncio import inspect from collections.abc import Awaitable from pathlib import Path @@ -223,6 +224,52 @@ def test_get_engine_uses_null_pool_for_bounded_background_processes(monkeypatch) assert "pool_timeout" not in captured +@pytest.mark.asyncio +async def test_null_pool_session_limiter_serializes_worker_db_sessions( + monkeypatch, +) -> None: + from src.db import base as db_base + + assert "async with _database_session_slot()" in inspect.getsource( + db_base.get_db + ) + assert "async with _database_session_slot()" in inspect.getsource( + db_base.get_db_context + ) + monkeypatch.setattr(db_base.settings, "DATABASE_NULL_POOL", True) + monkeypatch.setattr( + db_base.settings, + "DATABASE_NULL_POOL_CONCURRENCY_LIMIT", + 1, + ) + monkeypatch.setattr(db_base.settings, "DATABASE_POOL_TIMEOUT_SECONDS", 1.0) + monkeypatch.setattr(db_base, "_null_pool_session_limiter", None) + monkeypatch.setattr(db_base, "_null_pool_session_limiter_size", 0) + first_entered = asyncio.Event() + release_first = asyncio.Event() + second_entered = asyncio.Event() + + async def first_session() -> None: + async with db_base._database_session_slot(): + first_entered.set() + await release_first.wait() + + async def second_session() -> None: + async with db_base._database_session_slot(): + second_entered.set() + + first_task = asyncio.create_task(first_session()) + await first_entered.wait() + second_task = asyncio.create_task(second_session()) + await asyncio.sleep(0) + + assert second_entered.is_set() is False + + release_first.set() + await asyncio.gather(first_task, second_task) + assert second_entered.is_set() is True + + @pytest.mark.asyncio async def test_init_db_serializes_bootstrap_ddl_with_advisory_lock(monkeypatch): from src.db import base as db_base diff --git a/apps/api/tests/test_signal_worker_ansible_executor_binding.py b/apps/api/tests/test_signal_worker_ansible_executor_binding.py index 21245814e..7eaf00e8a 100644 --- a/apps/api/tests/test_signal_worker_ansible_executor_binding.py +++ b/apps/api/tests/test_signal_worker_ansible_executor_binding.py @@ -21,6 +21,7 @@ def test_signal_worker_produces_candidates_and_owns_security_maintenance() -> No assert "run_compliance_scanner_loop" in source assert "run_coverage_evaluator_loop" in source assert "run_asset_change_tracker_loop" in source + assert "run_asset_capability_reconciliation_loop" in source assert "signal_worker_security_control_plane_maintenance_started" in source assert "task.cancel()" in source assert "await asyncio.gather(*security_maintenance_tasks" in source @@ -33,6 +34,25 @@ def test_signal_worker_produces_candidates_and_owns_security_maintenance() -> No assert "signal_worker_ansible_check_mode_loop_started" not in source +def test_signal_worker_is_single_asset_capability_reconciliation_owner() -> None: + from src import main as api_main + + api_source = inspect.getsource(api_main.lifespan) + worker_source = inspect.getsource(signal_worker._main) + + assert "run_asset_capability_reconciliation_loop" not in api_source + assert "asset_capability_reconciliation_owner_delegated" in api_source + assert "run_asset_capability_reconciliation_loop" in worker_source + maintenance_start = worker_source.index("security_maintenance_loops =") + maintenance_end = worker_source.index( + "security_maintenance_tasks =", maintenance_start + ) + assert ( + "run_asset_capability_reconciliation_loop" + in worker_source[maintenance_start:maintenance_end] + ) + + def test_dedicated_broker_owns_ansible_executor_loop() -> None: source = inspect.getsource(ansible_executor_broker._main) @@ -71,6 +91,8 @@ def test_worker_manifest_has_no_execution_transport_or_write_loop() -> None: assert worker_env["MCP_FEDERATION_STARTUP_SLEEP_SECONDS"] == "45" assert worker_env["MCP_FEDERATION_SOURCE_TIMEOUT_SECONDS"] == "10" assert worker_env["AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WINDOW_HOURS"] == "168" + assert worker_env["DATABASE_NULL_POOL"] == "true" + assert worker_env["DATABASE_NULL_POOL_CONCURRENCY_LIMIT"] == "1" def test_execution_broker_is_only_workload_with_ssh_transport() -> None: @@ -198,7 +220,10 @@ def test_cd_prunes_and_verifies_api_executor_boundary_in_production() -> None: assert "legacy cluster-wide executor binding still exists" in workflow assert "awoooi-executor-boundary-verification" in workflow assert "awoooi_executor_boundary_verification_v4" in workflow - assert "single_writer_source_verifier=cd_tests_and_production_source_sha_v1" in workflow + assert ( + "single_writer_source_verifier=cd_tests_and_production_source_sha_v1" + in workflow + ) assert "decision_manager_queue_only=true" in workflow assert "webhook_router_queue_only=true" in workflow assert "failure_watcher_queue_only=true" in workflow @@ -206,24 +231,27 @@ def test_cd_prunes_and_verifies_api_executor_boundary_in_production() -> None: assert "candidate_idempotency_contract_verified=true" in workflow assert "apply_idempotency_contract_verified=true" in workflow assert "critical_break_glass_contract_verified=true" in workflow - assert "independent_post_verifier_source_verifier=cd_tests_and_production_source_sha_v1" in workflow + assert ( + "independent_post_verifier_source_verifier=cd_tests_and_production_source_sha_v1" + in workflow + ) assert "asset_postcondition_registry_complete=true" in workflow assert "independent_post_verifier_read_only=true" in workflow assert "executor_returncode_not_success_source=true" in workflow assert "verifier_failure_marks_apply_failed=true" in workflow assert "verifier_failure_retry_first=true" in workflow assert "post_verifier_raw_output_not_persisted=true" in workflow - assert "canonical_learning_source_verifier=cd_tests_and_production_source_sha_v1" in workflow + assert ( + "canonical_learning_source_verifier=cd_tests_and_production_source_sha_v1" + in workflow + ) assert "canonical_playbook_resolver_complete=true" in workflow assert "km_row_readback_required=true" in workflow assert "playbook_row_readback_required=true" in workflow assert "learning_failure_fail_closed=true" in workflow assert "trust_failure_fail_closed=true" in workflow assert "learning_receipts_public_safe=true" in workflow - assert ( - "k8s/awoooi-prod/08-deployment-ansible-executor-broker.yaml" - in workflow - ) + assert "k8s/awoooi-prod/08-deployment-ansible-executor-broker.yaml" in workflow assert "executor_boundary_public_readback=verified_ready" in workflow assert "EXECUTOR_BOUNDARY_READBACK_ATTEMPTS" in workflow assert "range(1, attempts + 1)" in workflow @@ -234,9 +262,7 @@ def test_cd_prunes_and_verifies_api_executor_boundary_in_production() -> None: assert "autonomous_single_writer_source_boundary" in workflow assert "independent_post_verifier_source_boundary" in workflow assert "canonical_learning_source_boundary" in workflow - worker_env = worker_deployment["spec"]["template"]["spec"]["containers"][0][ - "env" - ] + worker_env = worker_deployment["spec"]["template"]["spec"]["containers"][0]["env"] assert any(item.get("name") == "AWOOOI_BUILD_COMMIT_SHA" for item in worker_env) assert "k8s/awoooi-prod/08-deployment-worker.yaml" in workflow assert '"source_boundary_verified"' in workflow @@ -254,9 +280,18 @@ def test_cd_prunes_and_verifies_api_executor_boundary_in_production() -> None: assert "current_user AS role_name" in workflow assert "hashlib.sha256" in workflow assert "representative_db_preflights_verified" in workflow - assert 'api_required_connection_budget="$API_DB_REQUIRED_CONNECTION_BUDGET"' in workflow - assert 'worker_required_connection_budget="$WORKER_DB_REQUIRED_CONNECTION_BUDGET"' in workflow - assert 'broker_required_connection_budget="$BROKER_DB_REQUIRED_CONNECTION_BUDGET"' in workflow + assert ( + 'api_required_connection_budget="$API_DB_REQUIRED_CONNECTION_BUDGET"' + in workflow + ) + assert ( + 'worker_required_connection_budget="$WORKER_DB_REQUIRED_CONNECTION_BUDGET"' + in workflow + ) + assert ( + 'broker_required_connection_budget="$BROKER_DB_REQUIRED_CONNECTION_BUDGET"' + in workflow + ) assert "api_representative_connection_probe_count" in workflow assert "worker_representative_connection_probe_count" in workflow assert "broker_representative_connection_probe_count" in workflow @@ -330,7 +365,5 @@ def test_api_log_does_not_claim_skipped_executor_was_scheduled() -> None: assert "awooop_ansible_check_mode_worker_schedule_evaluated" in main_source assert "scheduled_in_api_process=scheduled_task is not None" in main_source assert 'production_process_owner="awoooi-worker"' in main_source - assert ( - 'production_process_owner="awoooi-ansible-executor-broker"' in main_source - ) + assert 'production_process_owner="awoooi-ansible-executor-broker"' in main_source assert "awooop_ansible_candidate_backfill_worker_schedule_evaluated" in main_source 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 diff --git a/config/security/runtime-image-mirror-staging-policy.json b/config/security/runtime-image-mirror-staging-policy.json index 5fff9af60..0242a73ed 100644 --- a/config/security/runtime-image-mirror-staging-policy.json +++ b/config/security/runtime-image-mirror-staging-policy.json @@ -80,6 +80,19 @@ "name": "local-path-provisioner", "container": "local-path-provisioner" } + }, + { + "asset_id": "runtime-image:sealed-secrets-controller-0.26.0-canary", + "source_index_digest": "sha256:74cd190f424b591dd82a234685c8c81d50b33af807e03470bc12b23d202e0106", + "platform_digest": "sha256:1827b36853e124ac0dd2554f7cd55c04952ecb7c38ff19a4a8d93d633811ed68", + "source_config_digest": "sha256:bbfc8314cb4978b252fb313b84e1d51a7b0c040befd79ecc903ea952485a3348", + "push_ref": "registry.wooo.work/awoooi/runtime-mirror/sealed-secrets-controller:0.26.0-linux-amd64", + "workload": { + "kind": "deployment", + "namespace": "kube-system", + "name": "sealed-secrets-controller", + "container": "sealed-secrets-controller" + } } ] } diff --git a/k8s/awoooi-prod/06-deployment-api.yaml b/k8s/awoooi-prod/06-deployment-api.yaml index c826f1171..3af3f9487 100644 --- a/k8s/awoooi-prod/06-deployment-api.yaml +++ b/k8s/awoooi-prod/06-deployment-api.yaml @@ -160,12 +160,12 @@ spec: - name: AWOOOI_BUILD_COMMIT_SHA # 2026-06-29 Codex: CD rewrites this to the deployed image tag so # production deploy readback does not rely on a stale static snapshot. - value: "ca49401c6af361b0128498c7355074538ebfe61e" + value: "0f2ddc0ce81ab52c7fbe33cb2e14dc54d0d33ed5" - name: AWOOOI_DESIRED_API_IMAGE_TAG # 2026-06-30 Codex: CD rewrites this alongside AWOOOI_BUILD_COMMIT_SHA. # Production readback compares runtime image truth against this # GitOps desired tag instead of doing a slow Gitea raw fetch. - value: "ca49401c6af361b0128498c7355074538ebfe61e" + value: "0f2ddc0ce81ab52c7fbe33cb2e14dc54d0d33ed5" - name: DATABASE_POOL_SIZE # 2026-07-01 Codex: production role `awoooi` currently has a low # connection limit. Keep API pool conservative until DB role diff --git a/k8s/awoooi-prod/08-deployment-ansible-executor-broker.yaml b/k8s/awoooi-prod/08-deployment-ansible-executor-broker.yaml index ec3737ae6..643fc8be7 100644 --- a/k8s/awoooi-prod/08-deployment-ansible-executor-broker.yaml +++ b/k8s/awoooi-prod/08-deployment-ansible-executor-broker.yaml @@ -66,7 +66,7 @@ spec: - name: DATABASE_NULL_POOL value: "true" - name: AWOOOI_BUILD_COMMIT_SHA - value: "ca49401c6af361b0128498c7355074538ebfe61e" + value: "0f2ddc0ce81ab52c7fbe33cb2e14dc54d0d33ed5" - name: ENABLE_AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WORKER value: "false" - name: ENABLE_AWOOOP_ANSIBLE_CHECK_MODE_WORKER diff --git a/k8s/awoooi-prod/08-deployment-worker.yaml b/k8s/awoooi-prod/08-deployment-worker.yaml index f9510f202..edad2bbb6 100644 --- a/k8s/awoooi-prod/08-deployment-worker.yaml +++ b/k8s/awoooi-prod/08-deployment-worker.yaml @@ -31,9 +31,9 @@ spec: strategy: type: RollingUpdate rollingUpdate: - # 2026-07-01 Codex: DB role `awoooi` is capped at two connections and - # worker imports the same API DB stack. Do not run old+new worker pods - # during rollout until the DB role/pool budget is raised and verified. + # 2026-07-15 Codex: keep one worker owner during rollout. This prevents + # duplicate consumers and keeps the isolated worker DB role inside its + # verified connection budget while the old pod terminates. maxSurge: 0 maxUnavailable: 1 template: @@ -163,17 +163,21 @@ spec: fieldRef: fieldPath: metadata.name - name: DATABASE_POOL_SIZE - # 2026-07-01 Codex: keep worker DB usage inside the current - # production role connection budget during reboot rollouts. + # NullPool releases each session; the explicit concurrency cap + # below preserves the intended one-connection worker budget. value: "1" - name: DATABASE_MAX_OVERFLOW value: "0" - name: DATABASE_NULL_POOL value: "true" + - name: DATABASE_NULL_POOL_CONCURRENCY_LIMIT + # NullPool releases each connection after use, while this cap + # keeps concurrent background loops inside the worker role budget. + value: "1" - name: DATABASE_BOOTSTRAP_DDL_ENABLED value: "false" - name: AWOOOI_BUILD_COMMIT_SHA - value: "ca49401c6af361b0128498c7355074538ebfe61e" + value: "0f2ddc0ce81ab52c7fbe33cb2e14dc54d0d33ed5" - name: ENABLE_AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WORKER value: "true" - name: ENABLE_SECURITY_CONTROL_PLANE_MAINTENANCE_WORKER diff --git a/k8s/awoooi-prod/kustomization.yaml b/k8s/awoooi-prod/kustomization.yaml index c0c28dda3..2d19b025a 100644 --- a/k8s/awoooi-prod/kustomization.yaml +++ b/k8s/awoooi-prod/kustomization.yaml @@ -40,9 +40,9 @@ resources: # ⚠️ 重要: name 必須與 deployment YAML 中的 image 完全匹配 (含 tag) # newName + newTag 由 CI 透過 kustomize edit set image 注入 images: -- digest: sha256:ce9009bb35292ae029d6bb78de3c8b889be8b5d1f03d6f313b2bec2932621c50 +- digest: sha256:1d195016008e4f5ec795170055aca2194161f74fc44fb373d387f26417f0b4af name: 192.168.0.110:5000/library/api:IMAGE_TAG_PLACEHOLDER newName: 192.168.0.110:5000/awoooi/api -- digest: sha256:a5f5e6dfacaec46c5f94477460c6e9aed8db8f7ac93be6915d1ee3e930b74bda +- digest: sha256:a1422caaefe56ce45d185cb85147ab3591e25251dc421cf738f1439565348c4d name: 192.168.0.110:5000/library/web:IMAGE_TAG_PLACEHOLDER newName: 192.168.0.110:5000/awoooi/web diff --git a/scripts/security/runtime_image_mirror_controller.py b/scripts/security/runtime_image_mirror_controller.py index 0050f514a..4b441df17 100644 --- a/scripts/security/runtime_image_mirror_controller.py +++ b/scripts/security/runtime_image_mirror_controller.py @@ -603,9 +603,21 @@ def _registry_descriptor(image_ref: str) -> RegistryDescriptor | None: ) -def _local_config_digest(image_ref: str) -> str: - output = _run(["docker", "image", "inspect", "--format={{.Id}}", image_ref]).stdout - return _require_digest((output or "").strip(), "local_config_digest") +def _local_platform_config_digest(image_ref: str, platform: str) -> str: + output = _run( + [ + "docker", + "image", + "inspect", + "--platform", + platform, + "--format={{.Id}}", + image_ref, + ] + ).stdout + return _require_digest( + (output or "").strip(), "local_platform_config_digest" + ) def _local_image_present(image_ref: str) -> bool: @@ -675,14 +687,30 @@ def mirror_images( if not _archive_contains_digest(archive, image.platform_digest): raise ControllerError("runtime_cache_export_digest_missing") _run( - ["docker", "load", "--input", str(archive)], + [ + "docker", + "load", + "--platform", + policy.platform, + "--input", + str(archive), + ], quiet=True, ) _run( ["docker", "tag", source_ref, image.push_ref], quiet=True, ) - _run(["docker", "push", image.push_ref], quiet=True) + _run( + [ + "docker", + "push", + "--platform", + policy.platform, + image.push_ref, + ], + quiet=True, + ) finally: _remove_local_image(image.push_ref) if not source_preexisting: @@ -763,17 +791,42 @@ def stage_images( cache.export(source_ref, archive) if not _archive_contains_digest(archive, image.platform_digest): raise ControllerError("runtime_cache_export_digest_missing") + if not _archive_contains_digest(archive, source_config_digest): + raise ControllerError( + "runtime_cache_export_config_digest_missing" + ) _run( - ["docker", "load", "--input", str(archive)], + [ + "docker", + "load", + "--platform", + policy.platform, + "--input", + str(archive), + ], quiet=True, ) - if _local_config_digest(source_ref) != source_config_digest: + if ( + _local_platform_config_digest( + source_ref, policy.platform + ) + != source_config_digest + ): raise ControllerError("local_image_config_digest_mismatch") _run( ["docker", "tag", source_ref, image.push_ref], quiet=True, ) - _run(["docker", "push", image.push_ref], quiet=True) + _run( + [ + "docker", + "push", + "--platform", + policy.platform, + image.push_ref, + ], + quiet=True, + ) finally: _remove_local_image(image.push_ref) if not source_preexisting: diff --git a/scripts/security/tests/test_runtime_image_mirror_controller.py b/scripts/security/tests/test_runtime_image_mirror_controller.py index 74d0efcb6..8107e2496 100644 --- a/scripts/security/tests/test_runtime_image_mirror_controller.py +++ b/scripts/security/tests/test_runtime_image_mirror_controller.py @@ -238,7 +238,7 @@ class RuntimeImageMirrorControllerTest(unittest.TestCase): self.assertEqual(policy.work_item_id, "AIA-P0-006-02B") self.assertEqual(policy.risk_level, "high") - self.assertEqual(len(policy.images), 5) + self.assertEqual(len(policy.images), 6) by_asset = {image.asset_id: image for image in policy.images} self.assertEqual( set(by_asset), @@ -247,6 +247,7 @@ class RuntimeImageMirrorControllerTest(unittest.TestCase): "runtime-image:local-path-provisioner-0.0.34-canary", "runtime-image:metrics-server-0.8.1-canary", "runtime-image:redis-8.2.3-canary", + "runtime-image:sealed-secrets-controller-0.26.0-canary", "runtime-image:vpa-recommender-1.6.0-canary", }, ) @@ -335,6 +336,32 @@ class RuntimeImageMirrorControllerTest(unittest.TestCase): self.assertEqual(local_path.workload.name, "local-path-provisioner") self.assertEqual(local_path.workload.container, "local-path-provisioner") self.assertEqual(local_path.workload.container_kind, "container") + + sealed_secrets = by_asset[ + "runtime-image:sealed-secrets-controller-0.26.0-canary" + ] + self.assertEqual( + sealed_secrets.source_index_digest, + "sha256:74cd190f424b591dd82a234685c8c81d50b33af807e03470bc12b23d202e0106", + ) + self.assertEqual( + sealed_secrets.platform_digest, + "sha256:1827b36853e124ac0dd2554f7cd55c04952ecb7c38ff19a4a8d93d633811ed68", + ) + self.assertEqual( + sealed_secrets.source_config_digest, + "sha256:bbfc8314cb4978b252fb313b84e1d51a7b0c040befd79ecc903ea952485a3348", + ) + self.assertEqual( + sealed_secrets.runtime_repository, "sealed-secrets-controller" + ) + self.assertEqual(sealed_secrets.workload.kind, "deployment") + self.assertEqual(sealed_secrets.workload.namespace, "kube-system") + self.assertEqual(sealed_secrets.workload.name, "sealed-secrets-controller") + self.assertEqual( + sealed_secrets.workload.container, "sealed-secrets-controller" + ) + self.assertEqual(sealed_secrets.workload.container_kind, "container") self.assertNotIn("github.com", raw) self.assertNotIn("ghcr.io", raw) self.assertIn('"external_pull_allowed": false', raw) @@ -711,6 +738,28 @@ class RuntimeImageMirrorControllerTest(unittest.TestCase): ) ) + def test_local_platform_config_digest_reads_image_id(self) -> None: + digest = "sha256:" + "1" * 64 + completed = controller.subprocess.CompletedProcess( + args=[], + returncode=0, + stdout=digest + "\n", + stderr="", + ) + with patch.object(controller, "_run", return_value=completed) as run: + self.assertEqual( + controller._local_platform_config_digest( + "source:tag", "linux/amd64" + ), + digest, + ) + + command = run.call_args.args[0] + self.assertEqual(command[:3], ["docker", "image", "inspect"]) + self.assertIn("--platform", command) + self.assertIn("linux/amd64", command) + self.assertIn("--format={{.Id}}", command) + def test_target_digest_verifier_uses_internal_registry_transport(self) -> None: image = controller.load_policy(POLICY_PATH).images[0] with patch.object(controller.subprocess, "run") as run: @@ -888,10 +937,12 @@ class RuntimeImageMirrorControllerTest(unittest.TestCase): side_effect=[None, descriptor], ), patch.object(controller, "_local_image_present", return_value=False), - patch.object(controller, "_archive_contains_digest", return_value=True), + patch.object( + controller, "_archive_contains_digest", return_value=True + ) as archive_contains, patch.object( controller, - "_local_config_digest", + "_local_platform_config_digest", return_value=image.source_config_digest, ), patch.object(controller, "_registry_digest_present", return_value=True), @@ -916,6 +967,11 @@ class RuntimeImageMirrorControllerTest(unittest.TestCase): export.assert_called_once() commands = [call.args[0] for call in run.call_args_list] self.assertEqual([command[1] for command in commands], ["load", "tag", "push"]) + self.assertEqual(archive_contains.call_count, 2) + self.assertIn("--platform", commands[0]) + self.assertIn(policy.platform, commands[0]) + self.assertIn("--platform", commands[2]) + self.assertIn(policy.platform, commands[2]) self.assertTrue(all("pull" not in command for command in commands)) self.assertEqual(remove.call_count, 2)