From b2deba5e84112ddc3203ab30b2449febb0a2dcd7 Mon Sep 17 00:00:00 2001 From: AWOOOI CD Date: Wed, 15 Jul 2026 13:18:03 +0800 Subject: [PATCH 1/5] chore(cd): deploy 4dc5b51 [skip ci] --- k8s/awoooi-prod/06-deployment-api.yaml | 4 ++-- k8s/awoooi-prod/08-deployment-ansible-executor-broker.yaml | 2 +- k8s/awoooi-prod/08-deployment-worker.yaml | 2 +- k8s/awoooi-prod/kustomization.yaml | 4 ++-- 4 files changed, 6 insertions(+), 6 deletions(-) diff --git a/k8s/awoooi-prod/06-deployment-api.yaml b/k8s/awoooi-prod/06-deployment-api.yaml index 2372f3605..cb1747741 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: "0ba0b30f68753dfd5d2da410b1940dfc19aebd0c" + value: "4dc5b51d0f1bb1d6c628ba747ab16d5c6bc1e9bf" - 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: "0ba0b30f68753dfd5d2da410b1940dfc19aebd0c" + value: "4dc5b51d0f1bb1d6c628ba747ab16d5c6bc1e9bf" - 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 bbcb1543a..d1a0e850d 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: "0ba0b30f68753dfd5d2da410b1940dfc19aebd0c" + value: "4dc5b51d0f1bb1d6c628ba747ab16d5c6bc1e9bf" - 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 3ee54c911..3694f5625 100644 --- a/k8s/awoooi-prod/08-deployment-worker.yaml +++ b/k8s/awoooi-prod/08-deployment-worker.yaml @@ -181,7 +181,7 @@ spec: - name: DATABASE_BOOTSTRAP_DDL_ENABLED value: "false" - name: AWOOOI_BUILD_COMMIT_SHA - value: "0ba0b30f68753dfd5d2da410b1940dfc19aebd0c" + value: "4dc5b51d0f1bb1d6c628ba747ab16d5c6bc1e9bf" - 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 eaf8ec803..ec9b74e56 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:8e1731a4f2e85f2dd9144a65c9eea67ee6bcaaa7ccd87223a871ad996a66e437 +- digest: sha256:70ee43cd6e0d4eb8ee972c109dc244011d9126c0e602354d7b1fa6d514852079 name: 192.168.0.110:5000/library/api:IMAGE_TAG_PLACEHOLDER newName: 192.168.0.110:5000/awoooi/api -- digest: sha256:4264e9763e212f2b776203b6d42fd05cc2c5c4a360ddc1d9c77a98e321028dad +- digest: sha256:3931889bc1f682b088ef3e0beb87de6ac67c2451031e550de127007d4af97246 name: 192.168.0.110:5000/library/web:IMAGE_TAG_PLACEHOLDER newName: 192.168.0.110:5000/awoooi/web From 89aee2157081d196d5eb067419200141d35cc57d Mon Sep 17 00:00:00 2001 From: ogt Date: Wed, 15 Jul 2026 13:11:47 +0800 Subject: [PATCH 2/5] fix(security): route Wazuh to canonical executor --- .../controlled_alert_target_router.py | 7 +- ...3s_controlled_automation_work_items_api.py | 32 +++++++ .../api/tests/test_sre_typed_domain_router.py | 86 +++++++++++++++++++ .../15-service-registry-configmap.yaml | 16 +++- ops/config/service-registry.yaml | 14 +++ 5 files changed, 153 insertions(+), 2 deletions(-) diff --git a/apps/api/src/services/controlled_alert_target_router.py b/apps/api/src/services/controlled_alert_target_router.py index 265dd5f90..7ae678a4d 100644 --- a/apps/api/src/services/controlled_alert_target_router.py +++ b/apps/api/src/services/controlled_alert_target_router.py @@ -389,7 +389,12 @@ def resolve_typed_alert_target( _text_token(target_resource) == "sentry-self-hosted-snuba-profiling-functions-consumer-1" ) - if not exact_sentry_consumer: + exact_registry_catalog_target = bool( + domain == "host_systemd" + and _text_token(target_resource) + == _text_token(str(identity.get("service_name") or "")) + ) + if not (exact_sentry_consumer or exact_registry_catalog_target): catalog_ids = [] stateful_level = str(identity.get("stateful_level") or "") risk = ( diff --git a/apps/api/tests/test_sre_k3s_controlled_automation_work_items_api.py b/apps/api/tests/test_sre_k3s_controlled_automation_work_items_api.py index f66d3ed77..4961741e8 100644 --- a/apps/api/tests/test_sre_k3s_controlled_automation_work_items_api.py +++ b/apps/api/tests/test_sre_k3s_controlled_automation_work_items_api.py @@ -181,3 +181,35 @@ def test_typed_target_preview_is_no_write_and_fail_closed() -> None: assert agent99_payload["route"]["agent99_bridge"][ "dispatch_allowed" ] is True + + +def test_wazuh_typed_target_preview_is_bounded_and_public_safe() -> None: + app = FastAPI() + app.include_router(router, prefix="/api/v1") + client = TestClient(app) + + response = client.post( + "/api/v1/agents/controlled-alert-target-preview", + json={ + "alertname": "WazuhSecurityEvent", + "target_resource": "wazuh-manager", + "namespace": "awoooi", + "labels": {"service": "wazuh-manager", "tool": "wazuh"}, + "alert_category": "security", + }, + ) + + assert response.status_code == 200 + payload = response.json() + route = payload["route"] + assert payload["preview_only"] is True + assert payload["dispatch_performed"] is False + assert payload["runtime_closure_verified"] is False + assert route["resolution_status"] == "resolved" + assert route["canonical_asset_id"] == "service:wazuh-manager" + assert route["host"] == "host:kali-readonly" + assert route["allowed_catalog_ids"] == [ + "ansible:wazuh-manager-posture-readback" + ] + assert route["allowed_inventory_hosts"] == ["host_112"] + assert route["cross_domain_fallback_allowed"] is False diff --git a/apps/api/tests/test_sre_typed_domain_router.py b/apps/api/tests/test_sre_typed_domain_router.py index 4234c6361..01ff13681 100644 --- a/apps/api/tests/test_sre_typed_domain_router.py +++ b/apps/api/tests/test_sre_typed_domain_router.py @@ -47,6 +47,7 @@ PLAYBOOK = ( ) SENTRY_CONTAINER = "sentry-self-hosted-snuba-profiling-functions-consumer-1" SENTRY_CATALOG = "ansible:110-sentry-profiling-consumer-recovery" +WAZUH_CATALOG = "ansible:wazuh-manager-posture-readback" def _sentry_incident() -> dict: @@ -110,6 +111,49 @@ def test_sentry_container_resolves_to_exact_host110_domain_route() -> None: assert route["reasoning_contract"]["paid_call_allowed"] is False +def test_wazuh_manager_exact_service_resolves_to_host112_posture_lane() -> None: + registry = ServiceRegistryClient(REGISTRY) + route = resolve_typed_alert_target( + alertname="WazuhSecurityEvent", + target_resource="wazuh-manager", + namespace="awoooi", + labels={"service": "wazuh-manager", "tool": "wazuh"}, + alert_category="security", + registry=registry, + ) + + identity = registry.resolve_identity("wazuh-manager") + assert identity["resolution_status"] == "resolved" + assert identity["canonical_id"] == "service:wazuh-manager" + assert identity["allowed_catalog_ids"] == [WAZUH_CATALOG] + assert route["resolution_status"] == "resolved" + assert route["target_kind"] == "host_systemd" + assert route["canonical_asset_id"] == "service:wazuh-manager" + assert route["executor"] == "host_ansible_executor" + assert route["verifier"] == "iwooos_wazuh_controlled_executor_runtime" + assert route["allowed_catalog_ids"] == [WAZUH_CATALOG] + assert route["allowed_inventory_hosts"] == ["host_112"] + assert route["controlled_apply_allowed"] is True + assert route["critical_break_glass_required"] is False + assert route["cross_domain_fallback_allowed"] is False + + +def test_wazuh_systemd_alias_cannot_inherit_execution_catalog() -> None: + route = resolve_typed_alert_target( + alertname="SystemdUnitUnavailable", + target_resource="wazuh-manager.service", + namespace="awoooi", + labels={}, + alert_category="security", + registry=ServiceRegistryClient(REGISTRY), + ) + + assert route["resolution_status"] == "resolved" + assert route["canonical_asset_id"] == "service:wazuh-manager" + assert route["allowed_catalog_ids"] == [] + assert route["cross_domain_fallback_allowed"] is False + + def test_sentry_incident_cannot_fall_to_host188_or_full_convergence() -> None: hints = _catalog_hints(_sentry_incident(), None) @@ -363,6 +407,48 @@ def test_sentry_candidate_reaches_bounded_check_mode_with_exact_scope() -> None: assert claim["controlled_apply_allowed"] is True +def test_wazuh_incident_reaches_bounded_host112_check_mode() -> None: + class Incident: + incident_id = "INC-WAZUH-TYPED-ROUTE" + project_id = "awoooi" + alertname = "WazuhSecurityEvent" + alert_category = "security" + notification_type = "TYPE-2" + severity = None + affected_services = ["wazuh-manager"] + signals = [ + { + "alert_name": "WazuhSecurityEvent", + "labels": { + "service": "wazuh-manager", + "tool": "wazuh", + "namespace": "awoooi", + }, + } + ] + + payload = build_ansible_decision_audit_payload( + incident=Incident(), + proposal_data={"source": "test", "risk_level": "medium"}, + decision_path="repair_candidate_controlled_queue", + not_used_reason="test", + ) + assert payload is not None + + claim = build_ansible_check_mode_claim_input( + source_candidate_op_id="00000000-0000-0000-0000-000000112112", + candidate_input=payload["input"], + ) + + assert claim["catalog_id"] == WAZUH_CATALOG + assert claim["inventory_hosts"] == ["host_112"] + assert claim["typed_domain"] == "host_systemd" + assert claim["canonical_asset_id"] == "service:wazuh-manager" + assert claim["target_scope_validated"] is True + assert claim["cross_domain_fallback_allowed"] is False + assert claim["controlled_apply_allowed"] is True + + def test_sentry_playbook_and_post_verifier_are_exact_container_bounded() -> None: payload = yaml.safe_load(PLAYBOOK.read_text(encoding="utf-8")) source = PLAYBOOK.read_text(encoding="utf-8") diff --git a/k8s/awoooi-prod/15-service-registry-configmap.yaml b/k8s/awoooi-prod/15-service-registry-configmap.yaml index df9b37c7f..221f1203e 100644 --- a/k8s/awoooi-prod/15-service-registry-configmap.yaml +++ b/k8s/awoooi-prod/15-service-registry-configmap.yaml @@ -9,7 +9,7 @@ metadata: app: awoooi component: service-registry annotations: - awoooi.wooo.work/source-sha256: "1662532ccf9c574e9853291a6f2881377c1a5b2519c4123ae05caf45be904121" + awoooi.wooo.work/source-sha256: "5548d19a6060900d51810cb3b1c3e1dc5f2fbb2b7c77df20227499cdb68599fc" data: service-registry.yaml: | # ops/config/service-registry.yaml @@ -165,6 +165,20 @@ data: containers: ["alertmanager"] # ─── AUTO：無狀態，允許自動修復 ────────────────────────────────────── + - name: wazuh-manager + canonical_id: "service:wazuh-manager" + display_name: "Wazuh Manager (host_112)" + host: "192.168.0.112" + stateful_level: AUTO + asset_domain: host_systemd + reason: "Only the bounded no-write posture catalog is enabled; active response and arbitrary service mutation remain unavailable" + executor: host_ansible_executor + verifier: iwooos_wazuh_controlled_executor_runtime + aliases: ["wazuh-manager.service"] + containers: [] + allowed_catalog_ids: + - ansible:wazuh-manager-posture-readback + - name: nginx display_name: "Nginx (反向代理)" host: "192.168.0.110" diff --git a/ops/config/service-registry.yaml b/ops/config/service-registry.yaml index a40397fed..d38263b38 100644 --- a/ops/config/service-registry.yaml +++ b/ops/config/service-registry.yaml @@ -151,6 +151,20 @@ services: containers: ["alertmanager"] # ─── AUTO：無狀態，允許自動修復 ────────────────────────────────────── + - name: wazuh-manager + canonical_id: "service:wazuh-manager" + display_name: "Wazuh Manager (host_112)" + host: "192.168.0.112" + stateful_level: AUTO + asset_domain: host_systemd + reason: "Only the bounded no-write posture catalog is enabled; active response and arbitrary service mutation remain unavailable" + executor: host_ansible_executor + verifier: iwooos_wazuh_controlled_executor_runtime + aliases: ["wazuh-manager.service"] + containers: [] + allowed_catalog_ids: + - ansible:wazuh-manager-posture-readback + - name: nginx display_name: "Nginx (反向代理)" host: "192.168.0.110" From 862be941f6cacf6de8e673e497c31a7a2485b300 Mon Sep 17 00:00:00 2001 From: ogt Date: Wed, 15 Jul 2026 13:31:28 +0800 Subject: [PATCH 3/5] fix(aiops): reconcile typed repair generations --- agent99-sre-alert-inbox.ps1 | 205 +++++++- apps/api/src/api/v1/webhooks.py | 12 +- .../awooop_ansible_candidate_backfill_job.py | 221 ++++++-- .../ai_agent_autonomous_runtime_control.py | 10 + .../services/awooop_ansible_audit_service.py | 487 ++++++++++++++++-- ...t_agent99_alert_dispatch_order_contract.py | 126 +++++ ...est_ai_agent_autonomous_runtime_control.py | 83 ++- ...t_awooop_ansible_candidate_backfill_job.py | 48 +- .../api/tests/test_sre_typed_domain_router.py | 319 ++++++++++++ 9 files changed, 1399 insertions(+), 112 deletions(-) diff --git a/agent99-sre-alert-inbox.ps1 b/agent99-sre-alert-inbox.ps1 index 85936a403..52d5e6d09 100644 --- a/agent99-sre-alert-inbox.ps1 +++ b/agent99-sre-alert-inbox.ps1 @@ -308,6 +308,190 @@ function Get-AgentControlledApplyForAlert { return [bool]($ModeName -in $remediationModes) } +function Convert-AgentInventoryHostToken { + param([object]$Value) + + $text = ([string]$Value).Trim().ToLowerInvariant() + if (-not $text) { return "" } + if ($text -match "^host[_-](188|121|120|112|111|110|99)$") { + return "host_$($Matches[1])" + } + if ($text -match "(?:^|[^0-9])(188|121|120|112|111|110|99)(?:[^0-9]|$)") { + return "host_$($Matches[1])" + } + return "" +} + +function Test-AgentTypedDispatchEnvelope { + param( + [object]$Alert, + [string]$ModeName, + [bool]$ControlledApply + ) + + if ($ModeName -notin $remediationModes) { + return [pscustomobject]@{ ok = $true; reason = "no_mutating_dispatch" } + } + if (-not $ControlledApply) { + return [pscustomobject]@{ ok = $false; reason = "mutating_mode_requires_controlled_apply" } + } + + $routing = Get-AgentField $Alert "routing" $null + $typedRoute = Get-AgentField $routing "typedTargetRoute" $null + if (-not $routing -or -not $typedRoute) { + return [pscustomobject]@{ ok = $false; reason = "typed_dispatch_route_missing" } + } + if ([string](Get-AgentField $routing "schemaVersion" "") -ne "agent99_alert_route_v1") { + return [pscustomobject]@{ ok = $false; reason = "typed_dispatch_envelope_schema_invalid" } + } + if ([string](Get-AgentField $typedRoute "schema_version" "") -ne "typed_domain_target_route_v2") { + return [pscustomobject]@{ ok = $false; reason = "typed_dispatch_route_schema_invalid" } + } + if ([string](Get-AgentField $typedRoute "resolution_status" "") -ne "resolved") { + return [pscustomobject]@{ ok = $false; reason = "typed_dispatch_asset_unresolved" } + } + + $typedDispatchAllowed = Get-AgentField $routing "typedDispatchAllowed" $null + $controlledApplyAllowed = Get-AgentField $typedRoute "controlled_apply_allowed" $null + $routingCrossDomain = Get-AgentField $routing "crossDomainFallbackAllowed" $null + $routeCrossDomain = Get-AgentField $typedRoute "cross_domain_fallback_allowed" $null + if (-not ($typedDispatchAllowed -is [bool]) -or -not [bool]$typedDispatchAllowed) { + return [pscustomobject]@{ ok = $false; reason = "typed_dispatch_not_allowed" } + } + if (-not ($controlledApplyAllowed -is [bool]) -or -not [bool]$controlledApplyAllowed) { + return [pscustomobject]@{ ok = $false; reason = "typed_dispatch_controlled_apply_not_allowed" } + } + if ( + -not ($routingCrossDomain -is [bool]) -or + [bool]$routingCrossDomain -or + -not ($routeCrossDomain -is [bool]) -or + [bool]$routeCrossDomain + ) { + return [pscustomobject]@{ ok = $false; reason = "typed_dispatch_cross_domain_forbidden" } + } + + $targetKind = [string](Get-AgentField $typedRoute "target_kind" "") + $executor = [string](Get-AgentField $typedRoute "executor" "") + $allowedModesForDomain = @{ + "control_plane_recovery" = @("Recover") + "windows_vmware" = @("Recover", "StartVMs") + } + if (-not $allowedModesForDomain.ContainsKey($targetKind) -or $executor -ne "Agent99") { + return [pscustomobject]@{ ok = $false; reason = "typed_dispatch_agent99_domain_not_allowed" } + } + if ($ModeName -notin @($allowedModesForDomain[$targetKind])) { + return [pscustomobject]@{ ok = $false; reason = "typed_dispatch_mode_domain_mismatch" } + } + + $bridge = Get-AgentField $typedRoute "agent99_bridge" $null + $bridgeDispatchAllowed = Get-AgentField $bridge "dispatch_allowed" $null + if ( + -not $bridge -or + [string](Get-AgentField $bridge "schema_version" "") -ne "agent99_host_operations_bridge_v1" -or + -not ($bridgeDispatchAllowed -is [bool]) -or + -not [bool]$bridgeDispatchAllowed + ) { + return [pscustomobject]@{ ok = $false; reason = "typed_dispatch_bridge_not_allowed" } + } + + $canonicalAssetId = ([string](Get-AgentField $typedRoute "canonical_asset_id" "")).Trim() + if (-not $canonicalAssetId) { + return [pscustomobject]@{ ok = $false; reason = "typed_dispatch_canonical_asset_missing" } + } + if ( + ($targetKind -eq "control_plane_recovery" -and $canonicalAssetId -notmatch "^(control-plane|host-guest):") -or + ($targetKind -eq "windows_vmware" -and $canonicalAssetId -notmatch "^windows-vmware:") + ) { + return [pscustomobject]@{ ok = $false; reason = "typed_dispatch_canonical_asset_domain_mismatch" } + } + + $awoooiMetadata = Get-AgentField $Alert "awoooi" $null + if ( + [string](Get-AgentField $awoooiMetadata "canonicalAssetId" "") -ne $canonicalAssetId -or + [string](Get-AgentField $awoooiMetadata "typedDomain" "") -ne $targetKind -or + [string](Get-AgentField $awoooiMetadata "assetResolutionStatus" "") -ne "resolved" + ) { + return [pscustomobject]@{ ok = $false; reason = "typed_dispatch_metadata_mismatch" } + } + + $allowedInventoryHosts = @( + @(Get-AgentField $typedRoute "allowed_inventory_hosts" @()) | + ForEach-Object { Convert-AgentInventoryHostToken $_ } | + Where-Object { $_ } + ) + if ($allowedInventoryHosts.Count -eq 0) { + return [pscustomobject]@{ ok = $false; reason = "typed_dispatch_inventory_scope_missing" } + } + $routeHostScope = Convert-AgentInventoryHostToken (Get-AgentField $typedRoute "host" "") + $alertHostText = [string](Get-AgentField $Alert "host" "") + $alertHostScope = Convert-AgentInventoryHostToken $alertHostText + if (-not $routeHostScope -or $routeHostScope -notin $allowedInventoryHosts) { + return [pscustomobject]@{ ok = $false; reason = "typed_dispatch_inventory_scope_mismatch" } + } + if ($alertHostText.Trim() -and (-not $alertHostScope -or $alertHostScope -ne $routeHostScope)) { + return [pscustomobject]@{ ok = $false; reason = "typed_dispatch_alert_host_mismatch" } + } + + $dispatchIdentity = Get-AgentField $awoooiMetadata "agent99DispatchIdentity" $null + if (-not $dispatchIdentity) { + return [pscustomobject]@{ ok = $false; reason = "controlled_dispatch_identity_missing" } + } + if ([string](Get-AgentField $dispatchIdentity "schema_version" "") -ne "agent99_controlled_dispatch_identity_v1") { + return [pscustomobject]@{ ok = $false; reason = "controlled_dispatch_identity_schema_invalid" } + } + foreach ($field in @( + "project_id", + "incident_id", + "source_fingerprint", + "route_id", + "execution_generation", + "run_id", + "trace_id", + "work_item_id", + "idempotency_key", + "single_flight_key", + "lock_owner", + "canonical_digest" + )) { + if (-not ([string](Get-AgentField $dispatchIdentity $field "")).Trim()) { + return [pscustomobject]@{ ok = $false; reason = "controlled_dispatch_identity_missing" } + } + } + if (-not $dispatchIdentity.PSObject.Properties["approval_id"]) { + return [pscustomobject]@{ ok = $false; reason = "controlled_dispatch_identity_missing" } + } + + foreach ($pair in @( + [pscustomobject]@{ identity = "run_id"; projection = "runId" }, + [pscustomobject]@{ identity = "trace_id"; projection = "traceId" }, + [pscustomobject]@{ identity = "work_item_id"; projection = "workItemId" }, + [pscustomobject]@{ identity = "idempotency_key"; projection = "idempotencyKey" }, + [pscustomobject]@{ identity = "execution_generation"; projection = "executionGeneration" } + )) { + if ( + [string](Get-AgentField $routing $pair.projection "") -ne + [string](Get-AgentField $dispatchIdentity $pair.identity "") + ) { + return [pscustomobject]@{ ok = $false; reason = "controlled_dispatch_identity_projection_mismatch" } + } + } + if ( + [string](Get-AgentField $awoooiMetadata "incidentId" "") -ne [string](Get-AgentField $dispatchIdentity "incident_id" "") -or + -not $awoooiMetadata.PSObject.Properties["approvalId"] -or + [string](Get-AgentField $awoooiMetadata "approvalId" "") -ne [string](Get-AgentField $dispatchIdentity "approval_id" "") + ) { + return [pscustomobject]@{ ok = $false; reason = "controlled_dispatch_identity_projection_mismatch" } + } + + return [pscustomobject]@{ + ok = $true + reason = "typed_dispatch_envelope_verified" + targetKind = $targetKind + canonicalAssetId = $canonicalAssetId + inventoryHost = $routeHostScope + } +} + function Test-AgentAlertSuppressTelegram { param([object]$Alert) if (Convert-AgentBool (Get-AgentField $Alert "suppressTelegram" $false)) { return $true } @@ -735,6 +919,24 @@ foreach ($file in $files) { continue } + $controlled = Get-AgentControlledApplyForAlert $alert $modeName + $typedDispatchDecision = Test-AgentTypedDispatchEnvelope $alert $modeName $controlled + if (-not $typedDispatchDecision.ok) { + $failedPath = Join-Path $FailedDir ("failed-" + (Convert-AgentSafeFileToken $alertId) + "-" + $file.Name) + Move-Item -Force $runningPath $failedPath + $failed += [pscustomobject]@{ + alertId = $alertId + file = $failedPath + ok = $false + reason = [string]$typedDispatchDecision.reason + mode = $modeName + controlledApply = $controlled + runtimeWritePerformed = $false + } + Write-AgentRelayQueueReceipt $alert $alertId "rejected" $false "" ([string]$typedDispatchDecision.reason) "typed_domain_rejected" + continue + } + $correlationKey = Get-AgentAlertCorrelationKey $alert $severity = [string](Get-AgentField $alert "severity" "unknown") $existingDedupe = @($dedupeEntries | Where-Object { [string]$_.key -eq $correlationKey } | Select-Object -First 1) @@ -782,7 +984,6 @@ foreach ($file in $files) { } $queueId = "sre-alert-" + (Get-Date -Format "yyyyMMdd-HHmmss") + "-" + ([guid]::NewGuid().ToString("N").Substring(0, 8)) - $controlled = Get-AgentControlledApplyForAlert $alert $modeName $instruction = New-AgentInstructionFromAlert $alert $alertId $modeName $source = [string](Get-AgentField $alert "source" "sre-war-room") $kind = [string](Get-AgentField $alert "kind" "unknown") @@ -812,7 +1013,7 @@ foreach ($file in $files) { $canonicalDigest = [string](Get-AgentField $dispatchIdentity "canonical_digest" "") $hasDispatchIdentity = [bool]($automationRunId -or $traceId -or $workItemId -or $idempotencyKey -or $incidentId -or $projectId -or $sourceFingerprint -or $dispatchRouteId -or $singleFlightKey) $dispatchIdentityIncomplete = [bool](-not $automationRunId -or -not $traceId -or -not $workItemId -or -not $idempotencyKey -or -not $incidentId -or -not $projectId -or -not $sourceFingerprint -or -not $dispatchRouteId -or -not $singleFlightKey -or -not $lockOwner -or -not $canonicalDigest -or -not $executionGeneration) - if (($controlled -and $modeName -eq "Recover" -and -not $hasDispatchIdentity) -or ($hasDispatchIdentity -and $dispatchIdentityIncomplete)) { + if (($controlled -and $modeName -in $remediationModes -and -not $hasDispatchIdentity) -or ($hasDispatchIdentity -and $dispatchIdentityIncomplete)) { $failedPath = Join-Path $FailedDir ("failed-" + (Convert-AgentSafeFileToken $alertId) + "-" + $file.Name) Move-Item -Force $runningPath $failedPath $failed += [pscustomobject]@{ diff --git a/apps/api/src/api/v1/webhooks.py b/apps/api/src/api/v1/webhooks.py index b910886e1..28b5d5eb4 100644 --- a/apps/api/src/api/v1/webhooks.py +++ b/apps/api/src/api/v1/webhooks.py @@ -436,6 +436,12 @@ async def _try_auto_repair_background( "namespace": namespace, "approval_id": approval_id, "execution_priority": execution_priority, + "source_fingerprint": resolved_source_fingerprint, + "source_occurrence_id": source_alert_id, + "source_recurrence_verified": bool( + resolved_source_fingerprint and source_alert_id + ), + "source_recurrence_source": "current_alert_webhook", }, project_id=str(getattr(incident, "project_id", None) or "awoooi"), ) @@ -1634,8 +1640,7 @@ async def receive_alert( if updated_approval: if ( - policy_promoted - and getattr(updated_approval, "incident_id", None) + getattr(updated_approval, "incident_id", None) and _controlled_ai_policy_allows(updated_approval.risk_level) and not is_heartbeat_alertname(alert.alert_type) ): @@ -3534,8 +3539,7 @@ async def alertmanager_webhook( updated_approval = await service.increment_hit_count(existing_approval.id) if updated_approval: if ( - policy_promoted - and getattr(updated_approval, "incident_id", None) + getattr(updated_approval, "incident_id", None) and _controlled_ai_policy_allows(updated_approval.risk_level) and not is_heartbeat_alertname(alertname) ): diff --git a/apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py b/apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py index c34e260c4..f23955fac 100644 --- a/apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py +++ b/apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py @@ -24,8 +24,8 @@ from src.core.config import settings from src.db.base import get_db_context from src.services.awooop_ansible_audit_service import ( build_ansible_decision_audit_payload, + preflight_ansible_candidate_generation, record_ansible_decision_audit, - verified_ansible_candidate_terminal_sql, ) from src.services.awooop_ansible_check_mode_service import ( preflight_failed_apply_retry_queue_once, @@ -47,6 +47,7 @@ _BACKFILL_REASON = ( ) _ERROR_BACKOFF_MIN_SECONDS = 15.0 _ERROR_BACKOFF_MAX_SECONDS = 30.0 +_CURRENT_RECURRENCE_MAX_AGE_SECONDS = 15 * 60 def _error_backoff_seconds() -> float: @@ -64,29 +65,35 @@ async def _fetch_missing_candidate_incidents( window_hours: int, scan_limit: int, ) -> list[dict[str, Any]]: - terminal_candidate_sql = verified_ansible_candidate_terminal_sql( - "existing" - ) async with get_db_context(project_id) as db: await db.execute(text("SET LOCAL statement_timeout = '5000ms'")) result = await db.execute( - text(f""" + text(""" SELECT - incident_id, - project_id, - status::text AS status, - severity::text AS severity, - alertname, - alert_category, - notification_type, - created_at, - updated_at, - resolved_at, - verification_result, - frequency_snapshot, - affected_services, - signals, - decision_chain, + incidents.incident_id, + incidents.project_id, + incidents.status::text AS status, + incidents.severity::text AS severity, + incidents.alertname, + incidents.alert_category, + incidents.notification_type, + incidents.created_at, + incidents.updated_at, + incidents.resolved_at, + incidents.verification_result, + incidents.frequency_snapshot, + incidents.affected_services, + incidents.signals, + incidents.decision_chain, + current_recurrence.source_fingerprint, + current_recurrence.source_occurrence_id, + current_recurrence.received_at AS source_received_at, + TRUE AS source_recurrence_verified, + 'awooop_conversation_event' AS source_recurrence_source, + latest_candidate.target_route_generation + AS latest_candidate_target_route_generation, + latest_candidate.source_occurrence_generation + AS latest_candidate_source_occurrence_generation, ( SELECT approval.id FROM approval_records approval @@ -99,45 +106,71 @@ async def _fetch_missing_candidate_incidents( LIMIT 1 ) AS approval_id FROM incidents - WHERE (project_id = :project_id OR project_id IS NULL) - AND created_at >= NOW() - (:window_hours * INTERVAL '1 hour') - AND resolved_at IS NULL - AND upper(coalesce(status::text, '')) NOT IN ('RESOLVED', 'CLOSED') - AND NOT EXISTS ( - SELECT 1 - FROM automation_operation_log existing - WHERE existing.operation_type = 'ansible_candidate_matched' - AND existing.created_at >= NOW() - (:window_hours * INTERVAL '1 hour') - AND existing.input ->> 'executor' = 'ansible' - AND existing.input ->> 'decision_path' = 'repair_candidate_controlled_queue' - AND coalesce(existing.incident_id::text, existing.input ->> 'incident_id') = incidents.incident_id::text - AND existing.op_id = ( - SELECT latest.op_id - FROM automation_operation_log latest - WHERE latest.operation_type - = 'ansible_candidate_matched' - AND latest.created_at >= NOW() - ( - :window_hours * INTERVAL '1 hour' - ) - AND latest.input ->> 'executor' = 'ansible' - AND latest.input ->> 'decision_path' - = 'repair_candidate_controlled_queue' - AND coalesce( - latest.incident_id::text, - latest.input ->> 'incident_id' - ) = incidents.incident_id::text - ORDER BY latest.created_at DESC, latest.op_id DESC - LIMIT 1 - ) - AND NOT ({terminal_candidate_sql}) - ) - ORDER BY created_at DESC + JOIN LATERAL ( + SELECT + recurrence.source_envelope #>> + '{log_correlation,fingerprint}' + AS source_fingerprint, + recurrence.provider_event_id AS source_occurrence_id, + recurrence.received_at + FROM awooop_conversation_event recurrence + WHERE recurrence.project_id = :project_id + AND recurrence.received_at >= NOW() - ( + :recurrence_max_age_seconds * INTERVAL '1 second' + ) + AND recurrence.source_envelope #> + '{source_refs,incident_ids}' + ? incidents.incident_id::text + AND NULLIF( + recurrence.source_envelope #>> + '{log_correlation,fingerprint}', + '' + ) IS NOT NULL + ORDER BY recurrence.received_at DESC + LIMIT 1 + ) current_recurrence ON TRUE + LEFT JOIN LATERAL ( + SELECT + candidate.input ->> 'target_route_generation' + AS target_route_generation, + candidate.input ->> 'source_occurrence_generation' + AS source_occurrence_generation + FROM automation_operation_log candidate + WHERE candidate.operation_type = 'ansible_candidate_matched' + AND candidate.input ->> 'executor' = 'ansible' + AND candidate.input ->> 'decision_path' + = 'repair_candidate_controlled_queue' + AND coalesce( + candidate.incident_id::text, + candidate.input ->> 'incident_id' + ) = incidents.incident_id::text + ORDER BY candidate.created_at DESC, candidate.op_id DESC + LIMIT 1 + ) latest_candidate ON TRUE + WHERE ( + incidents.project_id = :project_id + OR incidents.project_id IS NULL + ) + AND incidents.created_at >= NOW() - ( + :window_hours * INTERVAL '1 hour' + ) + AND incidents.resolved_at IS NULL + AND upper(coalesce(incidents.status::text, '')) + NOT IN ('RESOLVED', 'CLOSED') + AND NULLIF( + incidents.signals #>> '{0,labels,fingerprint}', + '' + ) = current_recurrence.source_fingerprint + ORDER BY current_recurrence.received_at DESC LIMIT :scan_limit """), { "project_id": project_id, "window_hours": max(1, window_hours), "scan_limit": max(1, scan_limit), + "recurrence_max_age_seconds": ( + _CURRENT_RECURRENCE_MAX_AGE_SECONDS + ), }, ) return [dict(row) for row in result.mappings().all()] @@ -150,6 +183,21 @@ def _build_backfill_proposal(incident: dict[str, Any]) -> dict[str, Any]: "action": "enqueue_allowlisted_ansible_check_mode", "alertname": incident.get("alertname"), "approval_id": str(incident.get("approval_id") or ""), + "source_fingerprint": str( + incident.get("source_fingerprint") or "" + ), + "source_occurrence_id": str( + incident.get("source_occurrence_id") or "" + ), + "source_received_at": str( + incident.get("source_received_at") or "" + ), + "source_recurrence_verified": ( + incident.get("source_recurrence_verified") is True + ), + "source_recurrence_source": str( + incident.get("source_recurrence_source") or "" + ), } @@ -325,6 +373,41 @@ async def enqueue_ai_decision_ansible_candidate( "active_blockers": ["no_allowlisted_ansible_candidate"], } + proposal_source = str(normalized_proposal.get("source") or "") + if proposal_source in { + "alert_webhook_controlled_router", + "truth_chain_candidate_backfill", + }: + generation_preflight = await preflight_ansible_candidate_generation( + incident=incident, + proposal_data=normalized_proposal, + decision_path=_BACKFILL_DECISION_PATH, + not_used_reason=( + "AI decision routed to the single-writer Ansible broker; " + "check-mode and diff are required before controlled apply" + ), + project_id=project_id, + ) + if generation_preflight.get("should_collect_evidence") is not True: + preflight_status = str( + generation_preflight.get("status") + or "generation_preflight_unavailable" + ) + existing_current = preflight_status == "existing_current_generation" + return { + "schema_version": "ai_decision_controlled_executor_handoff_v1", + "status": preflight_status, + "automation_run_id": generation_preflight.get( + "existing_candidate_op_id" + ), + "queued": existing_current, + "side_effect_performed": False, + "single_writer_executor": "awoooi-ansible-executor-broker", + "active_blockers": ( + [] if existing_current else [preflight_status] + ), + } + automation_run_id = str(uuid4()) try: snapshot = await evidence_collector( @@ -787,6 +870,8 @@ async def enqueue_missing_ansible_candidates_once( "no_catalog_candidate": 0, "pre_decision_evidence_ready": 0, "pre_decision_evidence_blocked": 0, + "existing_current_generation": 0, + "generation_reconciliation_blocked": 0, "repair_receipts_backfilled": 0, "failed_apply_retry_scanned": 0, "failed_apply_retry_replayed": 0, @@ -805,7 +890,11 @@ async def enqueue_missing_ansible_candidates_once( 1, window_hours or settings.AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WINDOW_HOURS, ) - scan_limit = min(100, max(25, bounded_limit * 5)) + # The batch write remains bounded by ``bounded_limit`` below, but the + # read-side scan must not let a small group of newer, non-catalog alerts + # starve an older exact typed candidate. One hundred rows is still a + # bounded query and matches the existing hard ceiling. + scan_limit = 100 stats: dict[str, Any] = { "skipped": False, "scanned": 0, @@ -814,6 +903,8 @@ async def enqueue_missing_ansible_candidates_once( "no_catalog_candidate": 0, "pre_decision_evidence_ready": 0, "pre_decision_evidence_blocked": 0, + "existing_current_generation": 0, + "generation_reconciliation_blocked": 0, "repair_receipts_backfilled": 0, "failed_apply_retry_scanned": 0, "failed_apply_retry_replayed": 0, @@ -890,6 +981,28 @@ async def enqueue_missing_ansible_candidates_once( if payload is None: stats["no_catalog_candidate"] += 1 continue + if incident.get("source_recurrence_verified") is True: + generation_preflight = ( + await preflight_ansible_candidate_generation( + incident=incident, + proposal_data=_build_backfill_proposal(incident), + decision_path=_BACKFILL_DECISION_PATH, + not_used_reason=_BACKFILL_REASON, + project_id=project_id, + ) + ) + if ( + generation_preflight.get("should_collect_evidence") + is not True + ): + if ( + generation_preflight.get("status") + == "existing_current_generation" + ): + stats["existing_current_generation"] += 1 + else: + stats["generation_reconciliation_blocked"] += 1 + continue automation_run_id = str(uuid4()) try: snapshot = await evidence_collector( diff --git a/apps/api/src/services/ai_agent_autonomous_runtime_control.py b/apps/api/src/services/ai_agent_autonomous_runtime_control.py index 294f3afe2..8e97dc9d3 100644 --- a/apps/api/src/services/ai_agent_autonomous_runtime_control.py +++ b/apps/api/src/services/ai_agent_autonomous_runtime_control.py @@ -8265,6 +8265,12 @@ _RUNTIME_OPERATION_CHAIN_SQL = """ CAST(NULLIF(:operation_chain_ref_2, '') AS uuid), CAST(NULLIF(:operation_chain_ref_3, '') AS uuid) ) + OR ( + operation_type = 'ansible_learning_writeback_recorded' + AND parent_op_id = CAST( + NULLIF(:operation_chain_ref_1, '') AS uuid + ) + ) ORDER BY created_at DESC """ @@ -8326,6 +8332,10 @@ _RUNTIME_OPERATION_CHAIN_DIRECT_SQL = """ NULLIF($2, '')::uuid, NULLIF($3, '')::uuid ) + OR ( + operation_type = 'ansible_learning_writeback_recorded' + AND parent_op_id = NULLIF($1, '')::uuid + ) ORDER BY created_at DESC """ diff --git a/apps/api/src/services/awooop_ansible_audit_service.py b/apps/api/src/services/awooop_ansible_audit_service.py index 1914d71e8..42dc7d31b 100644 --- a/apps/api/src/services/awooop_ansible_audit_service.py +++ b/apps/api/src/services/awooop_ansible_audit_service.py @@ -7,6 +7,7 @@ allowlisted playbooks with check-mode support can move from dry-run to apply. from __future__ import annotations +import hashlib import json import os import re @@ -24,6 +25,108 @@ from src.services.controlled_alert_target_router import ( logger = structlog.get_logger(__name__) +_TYPED_ROUTE_GENERATION_PREFIX = "typed-route-v2" + + +def _source_fingerprint(incident: dict[str, Any], proposal: dict[str, Any]) -> str: + """Return the current alert fingerprint without inventing one.""" + + explicit = str(proposal.get("source_fingerprint") or "").strip() + if explicit: + return explicit + for signal in incident.get("signals") or []: + if not isinstance(signal, dict): + continue + labels = signal.get("labels") + if not isinstance(labels, dict): + continue + fingerprint = str( + labels.get("fingerprint") or labels.get("alert_fingerprint") or "" + ).strip() + if fingerprint: + return fingerprint + return "" + + +def _source_recurrence_receipt( + incident: dict[str, Any], + proposal: dict[str, Any], +) -> dict[str, Any]: + """Build a fail-closed receipt proving this is not a historical-only replay.""" + + fingerprint = _source_fingerprint(incident, proposal) + proposal_source = str(proposal.get("source") or "").strip() + occurrence_id = str(proposal.get("source_occurrence_id") or "").strip() + verified = bool( + fingerprint + and occurrence_id + and proposal.get("source_recurrence_verified") is True + ) + return { + "schema_version": "alert_source_recurrence_receipt_v1", + "verified": verified, + "fingerprint": fingerprint or None, + "occurrence_id": occurrence_id or None, + "observed_at": str(proposal.get("source_received_at") or "") or None, + "source": ( + "current_alert_webhook" + if proposal_source == "alert_webhook_controlled_router" + else str(proposal.get("source_recurrence_source") or "") or None + ), + } + + +def build_typed_target_route_generation( + *, + typed_target_route: dict[str, Any], + executor_candidates: list[dict[str, Any]], +) -> str: + """Hash only semantic target scope, not deploy SHA or volatile evidence.""" + + identity = { + "schema_version": str(typed_target_route.get("schema_version") or ""), + "resolution_status": str( + typed_target_route.get("resolution_status") or "" + ), + "canonical_asset_id": str( + typed_target_route.get("canonical_asset_id") or "" + ), + "typed_domain": str(typed_target_route.get("target_kind") or ""), + "executor": str(typed_target_route.get("executor") or ""), + "cross_domain_fallback_allowed": typed_target_route.get( + "cross_domain_fallback_allowed" + ), + "candidates": sorted( + ( + { + "catalog_id": str(row.get("catalog_id") or ""), + "catalog_revision": str(row.get("catalog_revision") or ""), + "canonical_asset_id": str( + row.get("canonical_asset_id") or "" + ), + "typed_domain": str(row.get("typed_domain") or ""), + "inventory_hosts": sorted( + str(host) for host in row.get("inventory_hosts") or [] + ), + } + for row in executor_candidates + ), + key=lambda row: ( + row["catalog_id"], + row["catalog_revision"], + row["inventory_hosts"], + ), + ), + } + canonical = json.dumps( + identity, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ) + digest = hashlib.sha256(canonical.encode("utf-8")).hexdigest()[:24] + return f"{_TYPED_ROUTE_GENERATION_PREFIX}:{digest}" + def _automation_operation_log_incident_id(value: str) -> int | None: # ADR-090 keeps external INC-* ids in input JSON; the DB column is BIGINT. @@ -1273,8 +1376,49 @@ def build_ansible_decision_audit_payload( if not isinstance(execution_priority, int) or isinstance(execution_priority, bool): execution_priority = 100 execution_priority = max(0, min(execution_priority, 100)) - idempotency_key = f"ansible-candidate:{project_id}:{incident_id}" typed_target_route = hints["typed_target_route"] + executor_candidates = [ + { + "catalog_id": row["catalog_id"], + "playbook_path": row["playbook_path"], + "check_mode_playbook_path": row.get("check_mode_playbook_path"), + "inventory_hosts": row["inventory_hosts"], + "supports_check_mode": row["supports_check_mode"], + "auto_apply_enabled": row["auto_apply_enabled"], + "approval_required": row["approval_required"], + "risk_level": row["risk_level"], + "catalog_revision": row.get("catalog_revision"), + "canonical_asset_id": row.get("canonical_asset_id"), + "typed_domain": row.get("typed_domain"), + "target_scope_validated": row.get("target_scope_validated"), + "match_score": row["match_score"], + "matched_keywords": row["matched_keywords"], + "semantic_match": row["semantic_match"], + "matched_activation_keywords": row[ + "matched_activation_keywords" + ], + } + for row in candidates[:5] + ] + target_route_generation = build_typed_target_route_generation( + typed_target_route=typed_target_route, + executor_candidates=executor_candidates, + ) + source_recurrence = _source_recurrence_receipt( + incident_payload, + proposal_data, + ) + occurrence_identity = "|".join( + str(source_recurrence.get(key) or "") + for key in ("fingerprint", "occurrence_id", "observed_at") + ) + occurrence_generation = hashlib.sha256( + occurrence_identity.encode("utf-8") + ).hexdigest()[:20] + idempotency_key = ( + f"ansible-candidate:{project_id}:{incident_id}:" + f"{target_route_generation}:{occurrence_generation}" + ) input_payload = { "incident_id": incident_id, "project_id": project_id, @@ -1293,6 +1437,10 @@ def build_ansible_decision_audit_payload( "execution_priority": execution_priority, "decision_path": decision_path, "idempotency_key": idempotency_key, + "candidate_lane_lock_key": f"ansible-candidate:{project_id}:{incident_id}", + "target_route_generation": target_route_generation, + "source_occurrence_generation": occurrence_generation, + "source_recurrence": source_recurrence, "router_source_sha": os.getenv("AWOOOI_BUILD_COMMIT_SHA", "") .strip() .lower(), @@ -1302,29 +1450,7 @@ def build_ansible_decision_audit_payload( "candidate_catalog_schema": hints["match_mode"], "typed_target_route": typed_target_route, "cross_domain_fallback_allowed": False, - "executor_candidates": [ - { - "catalog_id": row["catalog_id"], - "playbook_path": row["playbook_path"], - "check_mode_playbook_path": row.get("check_mode_playbook_path"), - "inventory_hosts": row["inventory_hosts"], - "supports_check_mode": row["supports_check_mode"], - "auto_apply_enabled": row["auto_apply_enabled"], - "approval_required": row["approval_required"], - "risk_level": row["risk_level"], - "catalog_revision": row.get("catalog_revision"), - "canonical_asset_id": row.get("canonical_asset_id"), - "typed_domain": row.get("typed_domain"), - "target_scope_validated": row.get("target_scope_validated"), - "match_score": row["match_score"], - "matched_keywords": row["matched_keywords"], - "semantic_match": row["semantic_match"], - "matched_activation_keywords": row[ - "matched_activation_keywords" - ], - } - for row in candidates[:5] - ], + "executor_candidates": executor_candidates, "proposal_source": proposal_data.get("source", ""), "approval_id": str(proposal_data.get("approval_id") or ""), "proposal_risk_level": proposal_risk_level, @@ -1385,6 +1511,142 @@ def build_ansible_decision_audit_payload( } +async def preflight_ansible_candidate_generation( + *, + incident: Any, + proposal_data: dict[str, Any], + decision_path: str, + not_used_reason: str, + project_id: str = "awoooi", +) -> dict[str, Any]: + """Avoid expensive evidence collection for an already-current generation.""" + + payload = build_ansible_decision_audit_payload( + incident=incident, + proposal_data=proposal_data, + decision_path=decision_path, + not_used_reason=not_used_reason, + ) + if payload is None: + return { + "status": "no_allowlisted_ansible_candidate", + "should_collect_evidence": False, + } + incident_id = str(payload["input"].get("incident_id") or "") + generation = str( + payload["input"].get("target_route_generation") or "" + ) + occurrence = str( + payload["input"].get("source_occurrence_generation") or "" + ) + terminal_sql = verified_ansible_candidate_terminal_sql("candidate") + try: + async with get_db_context(project_id) as db: + result = await db.execute( + text(f""" + SELECT + candidate.op_id::text AS op_id, + candidate.input, + ({terminal_sql}) AS is_terminal, + EXISTS ( + SELECT 1 + FROM automation_operation_log child + WHERE child.parent_op_id = candidate.op_id + AND child.status IN ('pending', 'running') + ) OR EXISTS ( + SELECT 1 + FROM automation_operation_log check_mode + JOIN automation_operation_log apply + ON apply.parent_op_id = check_mode.op_id + AND apply.operation_type = 'ansible_apply_executed' + AND apply.status IN ('pending', 'running') + WHERE check_mode.parent_op_id = candidate.op_id + AND check_mode.operation_type + = 'ansible_check_mode_executed' + ) AS has_pending_child, + EXISTS ( + SELECT 1 + FROM automation_operation_log check_mode + JOIN automation_operation_log apply + ON apply.parent_op_id = check_mode.op_id + AND apply.operation_type = 'ansible_apply_executed' + WHERE check_mode.parent_op_id = candidate.op_id + AND check_mode.operation_type + = 'ansible_check_mode_executed' + ) AS has_apply + FROM automation_operation_log candidate + WHERE candidate.operation_type = 'ansible_candidate_matched' + AND coalesce( + candidate.incident_id::text, + candidate.input ->> 'incident_id' + ) = :incident_id + AND candidate.input ->> 'executor' = 'ansible' + AND candidate.input ->> 'decision_path' + = 'repair_candidate_controlled_queue' + ORDER BY candidate.created_at DESC, candidate.op_id DESC + LIMIT 1 + """), + {"incident_id": incident_id}, + ) + latest = result.mappings().first() + except Exception as exc: + logger.warning( + "ansible_candidate_generation_preflight_failed", + incident_id=incident_id, + error_type=type(exc).__name__, + ) + return { + "status": "generation_preflight_unavailable", + "should_collect_evidence": False, + } + + if latest is None: + return { + "status": "missing_candidate_generation", + "should_collect_evidence": True, + "target_route_generation": generation, + } + latest_input = _json_object(dict(latest), "input") + same_generation = str( + latest_input.get("target_route_generation") or "" + ) == generation + same_occurrence = str( + latest_input.get("source_occurrence_generation") or "" + ) == occurrence + is_terminal = latest.get("is_terminal") is True + if same_generation and (not is_terminal or same_occurrence): + return { + "status": "existing_current_generation", + "should_collect_evidence": False, + "target_route_generation": generation, + "existing_candidate_op_id": latest.get("op_id"), + } + if not is_terminal and latest.get("has_apply") is True: + return { + "status": "generation_reconciliation_blocked_has_apply", + "should_collect_evidence": False, + "target_route_generation": generation, + "existing_candidate_op_id": latest.get("op_id"), + } + if not is_terminal and latest.get("has_pending_child") is True: + return { + "status": "generation_reconciliation_deferred_pending_child", + "should_collect_evidence": False, + "target_route_generation": generation, + "existing_candidate_op_id": latest.get("op_id"), + } + return { + "status": ( + "fresh_occurrence_same_generation" + if same_generation + else "generation_reconciliation_required" + ), + "should_collect_evidence": True, + "target_route_generation": generation, + "existing_candidate_op_id": latest.get("op_id"), + } + + async def record_ansible_decision_audit( *, incident: Any, @@ -1411,9 +1673,11 @@ async def record_ansible_decision_audit( except (TypeError, ValueError): candidate_op_id = str(uuid4()) payload["input"]["automation_run_id"] = candidate_op_id - terminal_candidate_sql = verified_ansible_candidate_terminal_sql( - "candidate" + terminal_candidate_sql = verified_ansible_candidate_terminal_sql("candidate") + target_route_generation = str( + payload["input"].get("target_route_generation") or "" ) + source_recurrence = payload["input"].get("source_recurrence") or {} try: async with get_db_context(str(project_id)) as db: await db.execute( @@ -1422,11 +1686,44 @@ async def record_ansible_decision_audit( hashtextextended(:idempotency_key, 0) ) """), - {"idempotency_key": payload["input"]["idempotency_key"]}, + { + "idempotency_key": payload["input"][ + "candidate_lane_lock_key" + ] + }, ) - existing = await db.execute( + latest_result = await db.execute( text(f""" - SELECT candidate.op_id + SELECT + candidate.op_id::text AS op_id, + candidate.input, + ({terminal_candidate_sql}) AS is_terminal, + EXISTS ( + SELECT 1 + FROM automation_operation_log child + WHERE child.parent_op_id = candidate.op_id + AND child.status IN ('pending', 'running') + ) OR EXISTS ( + SELECT 1 + FROM automation_operation_log check_mode + JOIN automation_operation_log apply + ON apply.parent_op_id = check_mode.op_id + AND apply.operation_type = 'ansible_apply_executed' + AND apply.status IN ('pending', 'running') + WHERE check_mode.parent_op_id = candidate.op_id + AND check_mode.operation_type + = 'ansible_check_mode_executed' + ) AS has_pending_child, + EXISTS ( + SELECT 1 + FROM automation_operation_log check_mode + JOIN automation_operation_log apply + ON apply.parent_op_id = check_mode.op_id + AND apply.operation_type = 'ansible_apply_executed' + WHERE check_mode.parent_op_id = candidate.op_id + AND check_mode.operation_type + = 'ansible_check_mode_executed' + ) AS has_apply FROM automation_operation_log candidate WHERE candidate.operation_type = 'ansible_candidate_matched' AND coalesce( @@ -1436,28 +1733,124 @@ async def record_ansible_decision_audit( AND candidate.input ->> 'executor' = 'ansible' AND candidate.input ->> 'decision_path' = 'repair_candidate_controlled_queue' - AND candidate.op_id = ( - SELECT latest.op_id - FROM automation_operation_log latest - WHERE latest.operation_type - = 'ansible_candidate_matched' - AND coalesce( - latest.incident_id::text, - latest.input ->> 'incident_id' - ) = :incident_id - AND latest.input ->> 'executor' = 'ansible' - AND latest.input ->> 'decision_path' - = 'repair_candidate_controlled_queue' - ORDER BY latest.created_at DESC, latest.op_id DESC - LIMIT 1 - ) - AND NOT ({terminal_candidate_sql}) + ORDER BY candidate.created_at DESC, candidate.op_id DESC LIMIT 1 """), {"incident_id": incident_id}, ) - if existing.scalar() is not None: - return False + latest = latest_result.mappings().first() + if latest is not None: + latest_input = _json_object(dict(latest), "input") + latest_generation = str( + latest_input.get("target_route_generation") or "" + ) + latest_is_terminal = latest.get("is_terminal") is True + same_generation = latest_generation == target_route_generation + same_occurrence = str( + latest_input.get("source_occurrence_generation") or "" + ) == str( + payload["input"].get("source_occurrence_generation") or "" + ) + if same_generation and ( + not latest_is_terminal or same_occurrence + ): + return False + if source_recurrence.get("verified") is not True: + logger.warning( + "ansible_typed_route_supersession_blocked", + incident_id=incident_id, + reason="current_source_recurrence_not_verified", + previous_candidate_op_id=latest.get("op_id"), + ) + return False + if not latest_is_terminal and not same_generation: + if latest.get("has_apply") is True: + logger.warning( + "ansible_typed_route_supersession_blocked", + incident_id=incident_id, + reason="previous_candidate_has_apply", + previous_candidate_op_id=latest.get("op_id"), + ) + return False + if latest.get("has_pending_child") is True: + logger.info( + "ansible_typed_route_supersession_deferred", + incident_id=incident_id, + reason="previous_candidate_has_pending_child", + previous_candidate_op_id=latest.get("op_id"), + ) + return False + + await db.execute( + text(""" + INSERT INTO automation_operation_log ( + operation_type, actor, status, incident_id, + input, output, dry_run_result, + parent_op_id, tags + ) VALUES ( + 'ansible_execution_skipped', + 'typed_route_reconciler', + 'dry_run', + :incident_db_id, + CAST(:input AS jsonb), + CAST(:output AS jsonb), + CAST(:dry_run_result AS jsonb), + CAST(:parent_op_id AS uuid), + :tags + ) + """), + { + "incident_db_id": _automation_operation_log_incident_id( + incident_id + ), + "input": json.dumps( + { + "incident_id": incident_id, + "automation_run_id": str( + latest_input.get("automation_run_id") + or latest.get("op_id") + or "" + ), + "execution_mode": "typed_route_supersession", + "previous_target_route_generation": ( + latest_generation or None + ), + "superseded_by_target_route_generation": ( + target_route_generation + ), + "superseded_by_candidate_op_id": candidate_op_id, + "source_recurrence": source_recurrence, + "cross_domain_fallback_allowed": False, + }, + ensure_ascii=False, + ), + "output": json.dumps( + { + "reason": "typed_route_superseded", + "terminal_disposition": ( + "typed_route_superseded_no_write" + ), + "runtime_apply_executed": False, + }, + ensure_ascii=False, + ), + "dry_run_result": json.dumps( + { + "skipped": True, + "apply_executed": False, + "runtime_apply_executed": False, + "supersession_recorded": True, + }, + ensure_ascii=False, + ), + "parent_op_id": str(latest.get("op_id") or ""), + "tags": [ + "ansible", + "no_write", + "typed_route_superseded", + ], + }, + ) inserted = await db.execute( text(""" INSERT INTO automation_operation_log ( diff --git a/apps/api/tests/test_agent99_alert_dispatch_order_contract.py b/apps/api/tests/test_agent99_alert_dispatch_order_contract.py index 9d9fa1c56..c81da8324 100644 --- a/apps/api/tests/test_agent99_alert_dispatch_order_contract.py +++ b/apps/api/tests/test_agent99_alert_dispatch_order_contract.py @@ -109,6 +109,132 @@ def test_agent99_runtime_has_local_correlation_dedupe_and_lock() -> None: assert evidence_ref in control_plane +def test_agent99_mutating_inbox_requires_complete_typed_dispatch_envelope() -> None: + source = SRE_INBOX.read_text(encoding="utf-8") + guard = source[ + source.index("function Test-AgentTypedDispatchEnvelope") : source.index( + "function Test-AgentAlertSuppressTelegram" + ) + ] + + assert '"agent99_alert_route_v1"' in guard + assert '"typed_domain_target_route_v2"' in guard + assert '"resolved"' in guard + assert '"typedDispatchAllowed"' in guard + assert '"crossDomainFallbackAllowed"' in guard + assert '"cross_domain_fallback_allowed"' in guard + assert '"agent99_host_operations_bridge_v1"' in guard + assert '"dispatch_allowed"' in guard + assert '"controlled_apply_allowed"' in guard + assert '"control_plane_recovery" = @("Recover")' in guard + assert '"windows_vmware" = @("Recover", "StartVMs")' in guard + assert '$executor -ne "Agent99"' in guard + assert '"canonical_asset_id"' in guard + assert '"allowed_inventory_hosts"' in guard + assert '"canonicalAssetId"' in guard + assert '"typedDomain"' in guard + assert '"assetResolutionStatus"' in guard + assert '"agent99_controlled_dispatch_identity_v1"' in guard + for field in ( + "project_id", + "incident_id", + "source_fingerprint", + "route_id", + "execution_generation", + "run_id", + "trace_id", + "work_item_id", + "idempotency_key", + "single_flight_key", + "lock_owner", + "canonical_digest", + "approval_id", + ): + assert f'"{field}"' in guard + for projection in ( + "runId", + "traceId", + "workItemId", + "idempotencyKey", + "executionGeneration", + "incidentId", + "approvalId", + ): + assert f'"{projection}"' in guard + assert 'reason = "controlled_dispatch_identity_projection_mismatch"' in guard + + +def test_agent99_typed_guard_rejects_before_dedupe_or_queue_write() -> None: + source = SRE_INBOX.read_text(encoding="utf-8") + processing = source[source.index("foreach ($file in $files)") :] + guard_at = processing.index( + "$typedDispatchDecision = Test-AgentTypedDispatchEnvelope" + ) + reject_at = processing.index( + 'Write-AgentRelayQueueReceipt $alert $alertId "rejected"', guard_at + ) + dedupe_at = processing.index("$correlationKey = Get-AgentAlertCorrelationKey") + queue_at = processing.index('$queueId = "sre-alert-"') + + assert guard_at < reject_at < dedupe_at < queue_at + assert '"typed_domain_rejected"' in processing[guard_at:dedupe_at] + assert "runtimeWritePerformed = $false" in processing[guard_at:dedupe_at] + assert ( + '$controlled -and $modeName -in $remediationModes' + in processing + ) + + +def test_agent99_read_only_status_does_not_require_typed_dispatch_envelope() -> None: + source = SRE_INBOX.read_text(encoding="utf-8") + guard = source[ + source.index("function Test-AgentTypedDispatchEnvelope") : source.index( + "function Test-AgentAlertSuppressTelegram" + ) + ] + early_return_at = guard.index( + "if ($ModeName -notin $remediationModes)" + ) + controlled_apply_reject_at = guard.index("if (-not $ControlledApply)") + route_read_at = guard.index('$routing = Get-AgentField $Alert "routing"') + + assert early_return_at < controlled_apply_reject_at < route_read_at + assert 'reason = "no_mutating_dispatch"' in guard[:controlled_apply_reject_at] + assert ( + 'reason = "mutating_mode_requires_controlled_apply"' + in guard[controlled_apply_reject_at:route_read_at] + ) + assert ( + '$remediationModes = @("Recover", "StartVMs", "HarborRepair", ' + '"AwoooRepair", "Perf", "LoadShed")' + in source + ) + + +def test_agent99_generic_cold_start_waits_for_explicit_inventory_scope() -> None: + router = (ROOT / "apps/api/src/services/controlled_alert_target_router.py").read_text( + encoding="utf-8" + ) + cold_start_route = router[ + router.index("if legacy_recovery is not None:") : router.index( + "if compact_alert in {", + router.index("if legacy_recovery is not None:"), + ) + ] + inbox = SRE_INBOX.read_text(encoding="utf-8") + guard = inbox[ + inbox.index("function Test-AgentTypedDispatchEnvelope") : inbox.index( + "function Test-AgentAlertSuppressTelegram" + ) + ] + + assert 'canonical_asset_id="control-plane:cold-start-gate"' in cold_start_route + assert "allowed_inventory_hosts=" not in cold_start_route + assert 'reason = "typed_dispatch_inventory_scope_missing"' in guard + assert '"host-guest:host_112"' in router + assert 'allowed_inventory_hosts=["host_112"]' in router + + def test_running_claim_is_reconcile_only_and_pending_retry_is_not_before() -> None: source = ( ROOT diff --git a/apps/api/tests/test_ai_agent_autonomous_runtime_control.py b/apps/api/tests/test_ai_agent_autonomous_runtime_control.py index 6d3b970d1..6b2e00233 100644 --- a/apps/api/tests/test_ai_agent_autonomous_runtime_control.py +++ b/apps/api/tests/test_ai_agent_autonomous_runtime_control.py @@ -270,8 +270,19 @@ class _BudgetRuntimeDbContext: class _PrimaryAnchorRuntimeDb: - def __init__(self): + def __init__( + self, + *, + fail_operation_latest: bool = False, + include_learning_child: bool = False, + ): self.statements: list[str] = [] + self.fail_operation_latest = fail_operation_latest + self.include_learning_child = include_learning_child + self.rollback_count = 0 + + async def rollback(self) -> None: + self.rollback_count += 1 async def execute(self, statement, *_args, **_kwargs): sql = str(statement) @@ -292,9 +303,11 @@ class _PrimaryAnchorRuntimeDb: } ]) if "WITH latest_apply_chain AS" in sql: + if self.fail_operation_latest: + raise RuntimeError("operation_latest_query_failed") return _FakeMappingResult(_stdin_boundary_replay_rows()) if "CAST(NULLIF(:operation_chain_ref_1" in sql: - return _FakeMappingResult([ + rows = [ { "op_id": "stable-apply-op", "parent_op_id": "stable-check-op", @@ -317,7 +330,18 @@ class _PrimaryAnchorRuntimeDb: "status": "dry_run", "automation_run_id": "stable-candidate-op", }, - ]) + ] + if self.include_learning_child: + rows.append( + { + "op_id": "stable-learning-op", + "parent_op_id": "stable-apply-op", + "operation_type": "ansible_learning_writeback_recorded", + "status": "success", + "automation_run_id": "stable-candidate-op", + } + ) + return _FakeMappingResult(rows) return _FakeMappingResult([]) @@ -3438,6 +3462,7 @@ def test_runtime_operation_latest_queries_prioritize_latest_apply_chain() -> Non def test_runtime_primary_apply_anchor_is_global_and_chain_lookup_uses_uuid_index(): anchor_sql = runtime_control_module._RUNTIME_PRIMARY_APPLY_CHAIN_ANCHOR_SQL chain_sql = runtime_control_module._RUNTIME_OPERATION_CHAIN_SQL + direct_chain_sql = runtime_control_module._RUNTIME_OPERATION_CHAIN_DIRECT_SQL assert "operation_type = 'ansible_apply_executed'" in anchor_sql assert "AND status IN ('success', 'failed')" in anchor_sql @@ -3446,6 +3471,58 @@ def test_runtime_primary_apply_anchor_is_global_and_chain_lookup_uses_uuid_index assert "WHERE op_id IN" in chain_sql assert "CAST(NULLIF(:operation_chain_ref_1, '') AS uuid)" in chain_sql assert "WHERE op_id::text IN" not in chain_sql + for sql in (chain_sql, direct_chain_sql): + assert "operation_type = 'ansible_learning_writeback_recorded'" in sql + assert "AND parent_op_id = CAST(" in chain_sql + assert "NULLIF(:operation_chain_ref_1, '') AS uuid" in chain_sql + assert "AND parent_op_id = NULLIF($1, '')::uuid" in direct_chain_sql + + +@pytest.mark.asyncio +async def test_runtime_loader_recovers_learning_child_when_latest_query_fails( + monkeypatch, +): + fake_db = _PrimaryAnchorRuntimeDb( + fail_operation_latest=True, + include_learning_child=True, + ) + + async def _fake_consumer_readback(*, project_id: str): + assert project_id == "awoooi" + return _log_controlled_writeback_consumer_readback() + + monkeypatch.setattr( + runtime_control_module, + "get_db_context", + lambda project_id: _PrimaryAnchorRuntimeDbContext(fake_db), + ) + monkeypatch.setattr( + runtime_control_module, + "_load_log_controlled_writeback_consumer_readback", + _fake_consumer_readback, + ) + + readback = await ( + runtime_control_module._load_ai_agent_autonomous_runtime_receipt_readback_uncached() + ) + + assert readback["db_read_status"] == "partial" + assert { + (failure["query_name"], failure["error_type"]) + for failure in readback["partial_query_failures"] + } >= {("operation_latest", "RuntimeError")} + assert fake_db.rollback_count == 1 + + learning = readback["canonical_learning_runtime"] + assert learning["learning_operation_id"] == "stable-learning-op" + assert learning["controls"]["same_run_learning_operation"] is True + assert learning["controls"]["learning_operation_success"] is True + assert learning["controls"]["learning_parent_apply_matches"] is True + assert not { + "same_run_learning_operation_not_verified", + "learning_operation_success_not_verified", + "learning_parent_apply_matches_not_verified", + } & set(learning["active_blockers"]) @pytest.mark.asyncio diff --git a/apps/api/tests/test_awooop_ansible_candidate_backfill_job.py b/apps/api/tests/test_awooop_ansible_candidate_backfill_job.py index a1b9f6eed..b4c0c951c 100644 --- a/apps/api/tests/test_awooop_ansible_candidate_backfill_job.py +++ b/apps/api/tests/test_awooop_ansible_candidate_backfill_job.py @@ -293,7 +293,7 @@ async def test_backfill_skips_when_worker_disabled(monkeypatch: pytest.MonkeyPat assert result["queued"] == 0 -def test_backfill_query_excludes_existing_ansible_candidate_rows() -> None: +def test_backfill_query_reconciles_current_same_fingerprint_recurrence() -> None: from src.jobs import awooop_ansible_candidate_backfill_job as job source = MagicMock(wraps=job._fetch_missing_candidate_incidents) @@ -301,9 +301,53 @@ def test_backfill_query_excludes_existing_ansible_candidate_rows() -> None: joined = "\n".join(str(item) for item in query_source) assert source is not None - assert "NOT EXISTS" in joined assert "ansible_candidate_matched" in joined assert "resolved_at IS NULL" in joined + assert "awooop_conversation_event" in joined + assert "source_fingerprint" in joined + assert "source_occurrence_id" in joined + assert "target_route_generation" in joined + assert "source_recurrence_verified" in joined + + +@pytest.mark.asyncio +async def test_backfill_scans_bounded_hundred_rows_without_expanding_write_batch( + monkeypatch: pytest.MonkeyPatch, +) -> None: + from src.jobs import awooop_ansible_candidate_backfill_job as job + + observed: dict[str, int] = {} + + async def fake_fetch_missing_candidate_incidents( + *, project_id: str, window_hours: int, scan_limit: int + ) -> list[dict]: + assert project_id == "awoooi" + assert window_hours >= 1 + observed["scan_limit"] = scan_limit + return [] + + async def fake_retry_replayer(**_kwargs): + return {} + + monkeypatch.setattr( + job, + "_fetch_missing_candidate_incidents", + fake_fetch_missing_candidate_incidents, + ) + monkeypatch.setattr( + job.settings, + "ENABLE_AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WORKER", + True, + ) + + result = await job.enqueue_missing_ansible_candidates_once( + project_id="awoooi", + limit=1, + retry_replayer=fake_retry_replayer, + ) + + assert observed["scan_limit"] == 100 + assert result["queued"] == 0 def test_backfill_worker_error_backoff_is_short_and_bounded() -> None: diff --git a/apps/api/tests/test_sre_typed_domain_router.py b/apps/api/tests/test_sre_typed_domain_router.py index 4234c6361..cc7089374 100644 --- a/apps/api/tests/test_sre_typed_domain_router.py +++ b/apps/api/tests/test_sre_typed_domain_router.py @@ -3,8 +3,11 @@ from __future__ import annotations # ruff: noqa: E402, I001 import hashlib +import json import os +from contextlib import asynccontextmanager from pathlib import Path +from types import SimpleNamespace import pytest import yaml @@ -15,6 +18,7 @@ from src.services.awooop_ansible_audit_service import ( # noqa: E402 _catalog_hints, build_ansible_decision_audit_payload, get_ansible_catalog_item, + record_ansible_decision_audit, ) from src.services.awooop_ansible_check_mode_service import ( # noqa: E402 build_ansible_check_mode_claim_input, @@ -26,6 +30,10 @@ from src.services.agent99_sre_bridge import build_agent99_sre_alert # noqa: E40 from src.services.agent99_sre_bridge import ( # noqa: E402 bridge_alertmanager_to_agent99, ) +from src.services.agent99_controlled_dispatch_ledger import ( # noqa: E402 + attach_agent99_dispatch_identity, + build_agent99_dispatch_identity, +) from src.services.controlled_alert_target_router import ( # noqa: E402 resolve_typed_alert_target, ) @@ -69,6 +77,78 @@ def _sentry_incident() -> dict: } +def _sentry_runtime_incident() -> SimpleNamespace: + return SimpleNamespace( + incident_id="INC-20260714-471307", + project_id="awoooi", + alertname="DockerContainerUnhealthy", + alert_category="infrastructure", + notification_type="TYPE-3", + severity=SimpleNamespace(value="medium"), + affected_services=[SENTRY_CONTAINER], + signals=[ + SimpleNamespace( + alert_name="DockerContainerUnhealthy", + labels={ + "container_name": SENTRY_CONTAINER, + "namespace": "default", + "host": "110", + "fingerprint": "sentry-profiling-fingerprint", + }, + annotations={}, + ) + ], + ) + + +def _sentry_runtime_proposal(occurrence_id: str) -> dict: + return { + "source": "alert_webhook_controlled_router", + "risk_level": "medium", + "action": SENTRY_CONTAINER, + "source_fingerprint": "sentry-profiling-fingerprint", + "source_occurrence_id": occurrence_id, + "source_recurrence_verified": True, + "source_recurrence_source": "current_alert_webhook", + } + + +class _GenerationResult: + def __init__(self, *, row=None, scalar=None): + self._row = row + self._scalar = scalar + + def mappings(self): + return self + + def first(self): + return self._row + + def scalar(self): + return self._scalar + + +class _GenerationDb: + def __init__(self, latest: dict | None): + self.latest = latest + self.statements: list[str] = [] + self.parameters: list[dict] = [] + + async def execute(self, statement, parameters=None): + sql = str(statement) + self.statements.append(sql) + self.parameters.append(parameters or {}) + if "pg_advisory_xact_lock" in sql: + return _GenerationResult() + if "SELECT" in sql and "candidate.op_id::text AS op_id" in sql: + return _GenerationResult(row=self.latest) + if "typed_route_reconciler" in sql: + return _GenerationResult() + if "INSERT INTO automation_operation_log" in sql: + return _GenerationResult(scalar=parameters.get("op_id")) + raise AssertionError(f"unexpected SQL: {sql}") + + def test_unknown_registry_identity_fails_closed_without_restart_fallback() -> None: registry = ServiceRegistryClient(REGISTRY) @@ -198,6 +278,10 @@ def test_agent99_payload_carries_same_typed_host_operations_contract() -> None: assert payload["awoooi"]["canonicalAssetId"] == ( "control-plane:cold-start-gate" ) + # Generic cold-start remains no-write until the router binds an exact + # inventory scope; the Windows inbox rejects an empty scope fail-closed. + assert route["host"] is None + assert route["allowed_inventory_hosts"] == [] assert bridge["dispatch_identity"] == ( "agent99_controlled_dispatch_identity_v1" ) @@ -233,6 +317,49 @@ def test_host112_guest_recovery_is_an_exact_agent99_domain_route() -> None: assert payload["routing"]["typedDispatchAllowed"] is True assert payload["routing"]["crossDomainFallbackAllowed"] is False + identity = build_agent99_dispatch_identity( + project_id="awoooi", + incident_id="INC-HOST112-GUEST", + source_fingerprint="host112-guest-readiness", + route_id=route["route_id"], + work_item_id="AIA-SRE-HOST112-GUEST", + ) + enriched = attach_agent99_dispatch_identity(payload, identity) + dispatch_identity = enriched["awoooi"]["agent99DispatchIdentity"] + + assert enriched["host"] == "112" + assert route["host"] == "192.168.0.112" + assert route["allowed_inventory_hosts"] == ["host_112"] + assert dispatch_identity["schema_version"] == ( + "agent99_controlled_dispatch_identity_v1" + ) + for field in ( + "project_id", + "incident_id", + "source_fingerprint", + "route_id", + "execution_generation", + "run_id", + "trace_id", + "work_item_id", + "idempotency_key", + "single_flight_key", + "lock_owner", + "canonical_digest", + ): + assert dispatch_identity[field] + assert "approval_id" in dispatch_identity + for identity_field, projection_field in ( + ("run_id", "runId"), + ("trace_id", "traceId"), + ("work_item_id", "workItemId"), + ("idempotency_key", "idempotencyKey"), + ("execution_generation", "executionGeneration"), + ): + assert enriched["routing"][projection_field] == dispatch_identity[identity_field] + assert enriched["awoooi"]["incidentId"] == dispatch_identity["incident_id"] + assert enriched["awoooi"]["approvalId"] == dispatch_identity["approval_id"] + def test_unresolved_vmware_target_cannot_default_to_host99() -> None: route = resolve_typed_alert_target( @@ -363,6 +490,180 @@ def test_sentry_candidate_reaches_bounded_check_mode_with_exact_scope() -> None: assert claim["controlled_apply_allowed"] is True +@pytest.mark.asyncio +async def test_stale_broad_candidate_is_no_write_superseded_before_exact_candidate( + monkeypatch: pytest.MonkeyPatch, +) -> None: + old_candidate_id = "00000000-0000-0000-0000-000000471306" + new_candidate_id = "00000000-0000-0000-0000-000000471307" + db = _GenerationDb( + { + "op_id": old_candidate_id, + "input": { + "automation_run_id": old_candidate_id, + "candidate_catalog_schema": "static_catalog_keyword_hint_v1", + "executor_candidates": [ + { + "catalog_id": "ansible:188-ai-web", + "inventory_hosts": ["host_188"], + } + ], + }, + "is_terminal": False, + "has_pending_child": False, + "has_apply": False, + } + ) + + @asynccontextmanager + async def fake_db_context(_project_id: str = "awoooi"): + yield db + + monkeypatch.setattr( + "src.services.awooop_ansible_audit_service.get_db_context", + fake_db_context, + ) + + inserted = await record_ansible_decision_audit( + incident=_sentry_runtime_incident(), + proposal_data=_sentry_runtime_proposal("alert-current-1"), + decision_path="repair_candidate_controlled_queue", + not_used_reason="typed route reconciliation", + automation_run_id=new_candidate_id, + ) + + assert inserted is True + supersede_index = next( + index + for index, sql in enumerate(db.statements) + if "typed_route_reconciler" in sql + ) + candidate_index = next( + index + for index, sql in enumerate(db.statements) + if "'decision_manager'" in sql + ) + assert supersede_index < candidate_index + supersede = db.parameters[supersede_index] + assert supersede["parent_op_id"] == old_candidate_id + assert json.loads(supersede["dry_run_result"]) == { + "skipped": True, + "apply_executed": False, + "runtime_apply_executed": False, + "supersession_recorded": True, + } + candidate_input = json.loads(db.parameters[candidate_index]["input"]) + assert candidate_input["candidate_catalog_schema"] == "typed_domain_router_v2" + assert candidate_input["target_route_generation"].startswith("typed-route-v2:") + assert candidate_input["source_recurrence"]["verified"] is True + assert candidate_input["target_selector"]["catalog_ids"] == [SENTRY_CATALOG] + assert candidate_input["target_selector"]["inventory_hosts"] == ["host_110"] + + +@pytest.mark.asyncio +async def test_same_nonterminal_generation_dedupes_without_new_write( + monkeypatch: pytest.MonkeyPatch, +) -> None: + incident = _sentry_runtime_incident() + proposal = _sentry_runtime_proposal("alert-current-1") + payload = build_ansible_decision_audit_payload( + incident=incident, + proposal_data=proposal, + decision_path="repair_candidate_controlled_queue", + not_used_reason="test", + ) + assert payload is not None + db = _GenerationDb( + { + "op_id": "00000000-0000-0000-0000-000000471306", + "input": payload["input"], + "is_terminal": False, + "has_pending_child": False, + "has_apply": False, + } + ) + + @asynccontextmanager + async def fake_db_context(_project_id: str = "awoooi"): + yield db + + monkeypatch.setattr( + "src.services.awooop_ansible_audit_service.get_db_context", + fake_db_context, + ) + + inserted = await record_ansible_decision_audit( + incident=incident, + proposal_data=proposal, + decision_path="repair_candidate_controlled_queue", + not_used_reason="test", + ) + + assert inserted is False + assert not any("typed_route_reconciler" in sql for sql in db.statements) + assert not any("'decision_manager'" in sql for sql in db.statements) + + +@pytest.mark.asyncio +async def test_same_terminal_generation_accepts_one_fresh_occurrence( + monkeypatch: pytest.MonkeyPatch, +) -> None: + incident = _sentry_runtime_incident() + old_payload = build_ansible_decision_audit_payload( + incident=incident, + proposal_data=_sentry_runtime_proposal("alert-old"), + decision_path="repair_candidate_controlled_queue", + not_used_reason="test", + ) + assert old_payload is not None + db = _GenerationDb( + { + "op_id": "00000000-0000-0000-0000-000000471306", + "input": old_payload["input"], + "is_terminal": True, + "has_pending_child": False, + "has_apply": False, + } + ) + + @asynccontextmanager + async def fake_db_context(_project_id: str = "awoooi"): + yield db + + monkeypatch.setattr( + "src.services.awooop_ansible_audit_service.get_db_context", + fake_db_context, + ) + + inserted = await record_ansible_decision_audit( + incident=incident, + proposal_data=_sentry_runtime_proposal("alert-new"), + decision_path="repair_candidate_controlled_queue", + not_used_reason="test", + automation_run_id="00000000-0000-0000-0000-000000471307", + ) + + assert inserted is True + assert not any("typed_route_reconciler" in sql for sql in db.statements) + assert sum("'decision_manager'" in sql for sql in db.statements) == 1 + + +def test_webhook_recurrence_receipt_requires_explicit_occurrence_identity() -> None: + proposal = _sentry_runtime_proposal("") + payload = build_ansible_decision_audit_payload( + incident=_sentry_runtime_incident(), + proposal_data=proposal, + decision_path="repair_candidate_controlled_queue", + not_used_reason="test", + ) + + assert payload is not None + recurrence = payload["input"]["source_recurrence"] + assert recurrence["fingerprint"] == "sentry-profiling-fingerprint" + assert recurrence["occurrence_id"] is None + assert recurrence["verified"] is False + + def test_sentry_playbook_and_post_verifier_are_exact_container_bounded() -> None: payload = yaml.safe_load(PLAYBOOK.read_text(encoding="utf-8")) source = PLAYBOOK.read_text(encoding="utf-8") @@ -388,6 +689,24 @@ def test_sentry_playbook_and_post_verifier_are_exact_container_bounded() -> None assert forbidden not in source.lower() +def test_converged_webhook_recurrence_reenters_generation_preflight() -> None: + source = (ROOT / "apps/api/src/api/v1/webhooks.py").read_text( + encoding="utf-8" + ) + + assert '"source_fingerprint": resolved_source_fingerprint' in source + assert '"source_occurrence_id": source_alert_id' in source + assert '"source_recurrence_verified": bool(' in source + assert source.count( + 'getattr(updated_approval, "incident_id", None)\n' + " and _controlled_ai_policy_allows" + ) >= 2 + assert source.count( + "policy_promoted\n" + ' and getattr(updated_approval, "incident_id", None)' + ) == 0 + + def test_runtime_registry_is_exact_generated_mirror() -> None: source_text = REGISTRY.read_text(encoding="utf-8") configmap = yaml.safe_load(CONFIGMAP.read_text(encoding="utf-8")) From f202a0b454dd22a699758faa753644bf706fd627 Mon Sep 17 00:00:00 2001 From: AWOOOI CD Date: Wed, 15 Jul 2026 13:34:56 +0800 Subject: [PATCH 4/5] chore(cd): deploy 89aee21 [skip ci] --- k8s/awoooi-prod/06-deployment-api.yaml | 4 ++-- k8s/awoooi-prod/08-deployment-ansible-executor-broker.yaml | 2 +- k8s/awoooi-prod/08-deployment-worker.yaml | 2 +- k8s/awoooi-prod/kustomization.yaml | 4 ++-- 4 files changed, 6 insertions(+), 6 deletions(-) diff --git a/k8s/awoooi-prod/06-deployment-api.yaml b/k8s/awoooi-prod/06-deployment-api.yaml index cb1747741..f72ed628a 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: "4dc5b51d0f1bb1d6c628ba747ab16d5c6bc1e9bf" + value: "89aee2157081d196d5eb067419200141d35cc57d" - 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: "4dc5b51d0f1bb1d6c628ba747ab16d5c6bc1e9bf" + value: "89aee2157081d196d5eb067419200141d35cc57d" - 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 d1a0e850d..cdd9e613e 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: "4dc5b51d0f1bb1d6c628ba747ab16d5c6bc1e9bf" + value: "89aee2157081d196d5eb067419200141d35cc57d" - 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 3694f5625..e10060365 100644 --- a/k8s/awoooi-prod/08-deployment-worker.yaml +++ b/k8s/awoooi-prod/08-deployment-worker.yaml @@ -181,7 +181,7 @@ spec: - name: DATABASE_BOOTSTRAP_DDL_ENABLED value: "false" - name: AWOOOI_BUILD_COMMIT_SHA - value: "4dc5b51d0f1bb1d6c628ba747ab16d5c6bc1e9bf" + value: "89aee2157081d196d5eb067419200141d35cc57d" - 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 ec9b74e56..7f9a0d1b8 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:70ee43cd6e0d4eb8ee972c109dc244011d9126c0e602354d7b1fa6d514852079 +- digest: sha256:0552ac93bc3f2ee76917f7754a10bd9b7bf78243ae96263ae7ecbbd075307eb0 name: 192.168.0.110:5000/library/api:IMAGE_TAG_PLACEHOLDER newName: 192.168.0.110:5000/awoooi/api -- digest: sha256:3931889bc1f682b088ef3e0beb87de6ac67c2451031e550de127007d4af97246 +- digest: sha256:c9a8534df33ef42699c9b7ff29b3fbd0e27dcfc12378d183c7ff93bb5d3cc732 name: 192.168.0.110:5000/library/web:IMAGE_TAG_PLACEHOLDER newName: 192.168.0.110:5000/awoooi/web From ddea039cd6bfd7d6605a4a018b21bd72006bc0f4 Mon Sep 17 00:00:00 2001 From: ogt Date: Wed, 15 Jul 2026 13:40:23 +0800 Subject: [PATCH 5/5] test(aiops): preserve verified generation contract --- .../awooop_ansible_candidate_backfill_job.py | 18 ++++++++++++++---- .../api/tests/test_ansible_verified_closure.py | 13 ++++++++++--- 2 files changed, 24 insertions(+), 7 deletions(-) diff --git a/apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py b/apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py index f23955fac..e89e2d5a2 100644 --- a/apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py +++ b/apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py @@ -26,6 +26,7 @@ from src.services.awooop_ansible_audit_service import ( build_ansible_decision_audit_payload, preflight_ansible_candidate_generation, record_ansible_decision_audit, + verified_ansible_candidate_terminal_sql, ) from src.services.awooop_ansible_check_mode_service import ( preflight_failed_apply_retry_queue_once, @@ -65,10 +66,12 @@ async def _fetch_missing_candidate_incidents( window_hours: int, scan_limit: int, ) -> list[dict[str, Any]]: + terminal_candidate_sql = verified_ansible_candidate_terminal_sql( + "candidate" + ) async with get_db_context(project_id) as db: await db.execute(text("SET LOCAL statement_timeout = '5000ms'")) - result = await db.execute( - text(""" + query_sql = """ SELECT incidents.incident_id, incidents.project_id, @@ -94,6 +97,8 @@ async def _fetch_missing_candidate_incidents( AS latest_candidate_target_route_generation, latest_candidate.source_occurrence_generation AS latest_candidate_source_occurrence_generation, + latest_candidate.is_terminal + AS latest_candidate_is_terminal, ( SELECT approval.id FROM approval_records approval @@ -134,7 +139,8 @@ async def _fetch_missing_candidate_incidents( candidate.input ->> 'target_route_generation' AS target_route_generation, candidate.input ->> 'source_occurrence_generation' - AS source_occurrence_generation + AS source_occurrence_generation, + ({terminal_candidate_sql}) AS is_terminal FROM automation_operation_log candidate WHERE candidate.operation_type = 'ansible_candidate_matched' AND candidate.input ->> 'executor' = 'ansible' @@ -163,7 +169,11 @@ async def _fetch_missing_candidate_incidents( ) = current_recurrence.source_fingerprint ORDER BY current_recurrence.received_at DESC LIMIT :scan_limit - """), + """.replace( + "{terminal_candidate_sql}", terminal_candidate_sql + ) + result = await db.execute( + text(query_sql), { "project_id": project_id, "window_hours": max(1, window_hours), diff --git a/apps/api/tests/test_ansible_verified_closure.py b/apps/api/tests/test_ansible_verified_closure.py index 966887f36..9770392c3 100644 --- a/apps/api/tests/test_ansible_verified_closure.py +++ b/apps/api/tests/test_ansible_verified_closure.py @@ -618,9 +618,11 @@ def test_terminal_candidate_contract_requires_verified_four_node_chain() -> None assert "repository_readback_verified" in predicate assert "{detail,execution_success}" in predicate assert "{detail,telegram_receipt_acknowledged}" in predicate - assert "ORDER BY latest.created_at DESC, latest.op_id DESC" in ( + assert "ORDER BY candidate.created_at DESC, candidate.op_id DESC" in ( recorder_source ) + assert "target_route_generation" in recorder_source + assert "candidate_lane_lock_key" in recorder_source @pytest.mark.asyncio @@ -1169,8 +1171,13 @@ def test_backfill_preserves_approval_and_replaces_terminal_skipped_candidate() - assert proposal["approval_id"] == "00000000-0000-0000-0000-000000000103" assert "approval_records" in query_source assert "verified_ansible_candidate_terminal_sql" in query_source - assert "AND NOT ({terminal_candidate_sql})" in query_source - assert "ORDER BY latest.created_at DESC, latest.op_id DESC" in query_source + assert "current_recurrence" in query_source + assert "target_route_generation" in query_source + assert "latest_candidate_is_terminal" in query_source + assert "preflight_ansible_candidate_generation" in inspect.getsource( + candidate_job.enqueue_missing_ansible_candidates_once + ) + assert "ORDER BY candidate.created_at DESC, candidate.op_id DESC" in query_source def test_retry_projection_only_repairs_latest_candidate_run() -> None: