From ed3ed9d8e6bb9fe8ebc2ff214263ce125f16da0c Mon Sep 17 00:00:00 2001 From: AWOOOI CD Date: Thu, 9 Jul 2026 22:30:20 +0800 Subject: [PATCH 01/16] chore(cd): deploy 7ee0a17 [skip ci] --- k8s/awoooi-prod/06-deployment-api.yaml | 4 ++-- k8s/awoooi-prod/kustomization.yaml | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/k8s/awoooi-prod/06-deployment-api.yaml b/k8s/awoooi-prod/06-deployment-api.yaml index 6e749302e..14a84b213 100644 --- a/k8s/awoooi-prod/06-deployment-api.yaml +++ b/k8s/awoooi-prod/06-deployment-api.yaml @@ -84,12 +84,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: "cfeaff2ef1d0ee5484bf08a7d9308ca3cc61967f" + value: "7ee0a17ab7b6842dc8eb2104350cbab300d05243" - 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: "cfeaff2ef1d0ee5484bf08a7d9308ca3cc61967f" + value: "7ee0a17ab7b6842dc8eb2104350cbab300d05243" - 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/kustomization.yaml b/k8s/awoooi-prod/kustomization.yaml index b57b8ecf9..421dbb644 100644 --- a/k8s/awoooi-prod/kustomization.yaml +++ b/k8s/awoooi-prod/kustomization.yaml @@ -41,7 +41,7 @@ resources: images: - name: 192.168.0.110:5000/library/api:IMAGE_TAG_PLACEHOLDER newName: 192.168.0.110:5000/awoooi/api - newTag: cfeaff2ef1d0ee5484bf08a7d9308ca3cc61967f + newTag: 7ee0a17ab7b6842dc8eb2104350cbab300d05243 - name: 192.168.0.110:5000/library/web:IMAGE_TAG_PLACEHOLDER newName: 192.168.0.110:5000/awoooi/web - newTag: cfeaff2ef1d0ee5484bf08a7d9308ca3cc61967f + newTag: 7ee0a17ab7b6842dc8eb2104350cbab300d05243 From dced1be755e3f7f51a8462d321109377a02fe473 Mon Sep 17 00:00:00 2001 From: ogt Date: Thu, 9 Jul 2026 22:31:36 +0800 Subject: [PATCH 02/16] fix(agents): fail-soft AI alert card readback --- apps/api/src/api/v1/platform/operator_runs.py | 12 +- .../src/services/platform_operator_service.py | 171 ++++++++++++++---- ...gram_alert_monitoring_coverage_readback.py | 17 +- .../test_awooop_operator_timeline_labels.py | 81 ++++++++- ..._alert_monitoring_coverage_readback_api.py | 33 ++++ 5 files changed, 273 insertions(+), 41 deletions(-) diff --git a/apps/api/src/api/v1/platform/operator_runs.py b/apps/api/src/api/v1/platform/operator_runs.py index 22a3bf6b7..a9991cdfb 100644 --- a/apps/api/src/api/v1/platform/operator_runs.py +++ b/apps/api/src/api/v1/platform/operator_runs.py @@ -36,7 +36,7 @@ from src.services.platform_operator_service import ( get_run_detail as get_run_detail_svc, ) from src.services.platform_operator_service import ( - list_cicd_events as list_cicd_events_svc, + list_ai_alert_card_delivery_readback as list_ai_alert_card_delivery_readback_svc, ) from src.services.platform_operator_service import ( list_approvals as list_approvals_svc, @@ -45,7 +45,7 @@ from src.services.platform_operator_service import ( list_callback_replies as list_callback_replies_svc, ) from src.services.platform_operator_service import ( - list_ai_alert_card_delivery_readback as list_ai_alert_card_delivery_readback_svc, + list_cicd_events as list_cicd_events_svc, ) from src.services.platform_operator_service import ( list_runs as list_runs_svc, @@ -156,6 +156,10 @@ class AiAlertCardDeliverySummary(BaseModel): event_type: str | None = None lane: str | None = None status: str + db_read_status: str = "ok" + source_unavailable_reason: str | None = None + source_unavailable_next_action: str = "none" + controlled_repair_queue_present: bool = False total: int sent_total: int failed_total: int @@ -582,7 +586,7 @@ async def get_run_detail( ) async def get_awooop_status_chain( project_id: str | None = Query(None, description="租戶 ID（可選）"), - incident_id: list[str] | None = Query( + incident_id: list[str] | None = Query( # noqa: B008 None, description="Incident ID，可重複傳入以合併同一工作項的多個事件", ), @@ -631,7 +635,7 @@ async def list_approvals( async def decide_approval( run_id: str, body: DecideApprovalRequest, - operator: AwoooPOperatorPrincipal = Depends(verify_awooop_operator), + operator: AwoooPOperatorPrincipal = Depends(verify_awooop_operator), # noqa: B008 ) -> dict[str, Any]: return await decide_approval_svc( run_id=run_id, diff --git a/apps/api/src/services/platform_operator_service.py b/apps/api/src/services/platform_operator_service.py index 6d89ff2e3..bfefab1ed 100644 --- a/apps/api/src/services/platform_operator_service.py +++ b/apps/api/src/services/platform_operator_service.py @@ -39,6 +39,12 @@ from src.db.awooop_models import ( ) from src.db.base import get_db_context from src.db.models import ApprovalRecord, IncidentRecord, MCPAuditLog +from src.services.ai_agent_result_capture_owner_release_approval_gate import ( + load_latest_ai_agent_result_capture_owner_release_approval_gate, +) +from src.services.ai_agent_result_capture_release_verifier_preflight_gate import ( + load_latest_ai_agent_result_capture_release_verifier_preflight_gate, +) from src.services.audit_sink import write_audit from src.services.awooop_ansible_audit_service import summarize_ansible_execution from src.services.awooop_approval_token import issue_approval_token, record_approval @@ -47,12 +53,6 @@ from src.services.awooop_truth_chain_service import ( _summarize_mcp, fetch_truth_chain, ) -from src.services.ai_agent_result_capture_owner_release_approval_gate import ( - load_latest_ai_agent_result_capture_owner_release_approval_gate, -) -from src.services.ai_agent_result_capture_release_verifier_preflight_gate import ( - load_latest_ai_agent_result_capture_release_verifier_preflight_gate, -) from src.services.governance_km_stale_review_service import ( query_km_stale_owner_review_completion_queue, ) @@ -1350,11 +1350,23 @@ async def list_ai_alert_card_delivery_readback( "per_page": normalized_per_page, } if not refresh: - cached_response = await get_cached_operator_summary_async( - "ai_alert_card_delivery_readback", - cache_key, - ttl_seconds=_AI_ALERT_CARD_CACHE_TTL_SECONDS, - ) + try: + cached_response = await get_cached_operator_summary_async( + "ai_alert_card_delivery_readback", + cache_key, + ttl_seconds=_AI_ALERT_CARD_CACHE_TTL_SECONDS, + ) + except Exception as exc: # pragma: no cover - cache pressure + logger.warning( + "operator_ai_alert_card_delivery_readback_cache_unavailable", + project_id=normalized_project_id, + event_type=normalized_event_type, + lane=normalized_lane, + page=normalized_page, + per_page=normalized_per_page, + error_type=type(exc).__name__, + ) + cached_response = None if cached_response is not None and "learning_registry" in cached_response: logger.info( "operator_ai_alert_card_delivery_readback_cache_hit", @@ -1370,7 +1382,7 @@ async def list_ai_alert_card_delivery_readback( where_clauses = [ "m.project_id = :project_id", "m.channel_type = 'telegram'", - "m.source_envelope ? 'ai_automation_alert_card'", + "COALESCE(m.source_envelope::jsonb, '{}'::jsonb) ? 'ai_automation_alert_card'", ] params: dict[str, Any] = { "project_id": normalized_project_id, @@ -1379,12 +1391,14 @@ async def list_ai_alert_card_delivery_readback( } if normalized_event_type: where_clauses.append( - "m.source_envelope #>> '{ai_automation_alert_card,event_type}' = :event_type" + "COALESCE(m.source_envelope::jsonb, '{}'::jsonb) #>> " + "'{ai_automation_alert_card,event_type}' = :event_type" ) params["event_type"] = normalized_event_type if normalized_lane: where_clauses.append( - "m.source_envelope #>> '{ai_automation_alert_card,lane}' = :lane" + "COALESCE(m.source_envelope::jsonb, '{}'::jsonb) #>> " + "'{ai_automation_alert_card,lane}' = :lane" ) params["lane"] = normalized_lane @@ -1398,31 +1412,31 @@ async def list_ai_alert_card_delivery_readback( COUNT(*) FILTER (WHERE m.send_status = 'shadow') AS shadow_total, COUNT(*) FILTER ( WHERE COALESCE( - m.source_envelope #>> + COALESCE(m.source_envelope::jsonb, '{{}}'::jsonb) #>> '{{ai_automation_alert_card,delivery_receipt_readback_required}}', '' ) = 'true' ) AS delivery_receipt_required_total, COUNT(*) FILTER ( WHERE COALESCE( - m.source_envelope #>> + COALESCE(m.source_envelope::jsonb, '{{}}'::jsonb) #>> '{{ai_automation_alert_card,runtime_write_gate_count}}', '0' ) <> '0' ) AS runtime_write_gate_open_count, COUNT(*) FILTER ( WHERE COALESCE( - m.source_envelope #>> + COALESCE(m.source_envelope::jsonb, '{{}}'::jsonb) #>> '{{ai_automation_alert_card,delivery_receipt_readback_required}}', '' ) = 'true' AND COALESCE( - m.source_envelope #>> + COALESCE(m.source_envelope::jsonb, '{{}}'::jsonb) #>> '{{ai_automation_alert_card,event_type}}', '' ) <> '' AND COALESCE( - m.source_envelope #>> + COALESCE(m.source_envelope::jsonb, '{{}}'::jsonb) #>> '{{ai_automation_alert_card,lane}}', '' ) <> '' @@ -1445,8 +1459,10 @@ async def list_ai_alert_card_delivery_readback( m.queued_at, m.sent_at, m.triggered_by_state, - m.source_envelope -> 'ai_automation_alert_card' AS alert_card, - m.source_envelope -> 'source_refs' AS source_refs, + COALESCE(m.source_envelope::jsonb, '{{}}'::jsonb) + -> 'ai_automation_alert_card' AS alert_card, + COALESCE(m.source_envelope::jsonb, '{{}}'::jsonb) + -> 'source_refs' AS source_refs, r.agent_id, r.state AS run_state, r.created_at AS run_created_at @@ -1459,11 +1475,30 @@ async def list_ai_alert_card_delivery_readback( LIMIT :limit OFFSET :offset """) - async with get_db_context(normalized_project_id) as db: - summary_result = await db.execute(summary_sql, params) - summary_row = summary_result.mappings().first() or {} - rows_result = await db.execute(list_sql, params) - rows = list(rows_result.mappings().all()) + try: + async with get_db_context(normalized_project_id) as db: + summary_result = await db.execute(summary_sql, params) + summary_row = summary_result.mappings().first() or {} + rows_result = await db.execute(list_sql, params) + rows = list(rows_result.mappings().all()) + except Exception as exc: # pragma: no cover - production DB drift/pressure + logger.warning( + "operator_ai_alert_card_delivery_readback_source_unavailable", + project_id=normalized_project_id, + event_type=normalized_event_type, + lane=normalized_lane, + page=normalized_page, + per_page=normalized_per_page, + error_type=type(exc).__name__, + ) + return _ai_alert_card_delivery_source_unavailable_response( + project_id=normalized_project_id, + event_type=normalized_event_type or None, + lane=normalized_lane or None, + page=normalized_page, + per_page=normalized_per_page, + error_type=type(exc).__name__, + ) summary = _ai_alert_card_delivery_summary_from_row( summary_row, @@ -1503,12 +1538,86 @@ async def list_ai_alert_card_delivery_readback( cache_status="miss", cache_ttl_seconds=_AI_ALERT_CARD_CACHE_TTL_SECONDS, ) - return await store_operator_summary_async( - "ai_alert_card_delivery_readback", - cache_key, - response, - ttl_seconds=_AI_ALERT_CARD_CACHE_TTL_SECONDS, + try: + return await store_operator_summary_async( + "ai_alert_card_delivery_readback", + cache_key, + response, + ttl_seconds=_AI_ALERT_CARD_CACHE_TTL_SECONDS, + ) + except Exception as exc: # pragma: no cover - cache pressure + logger.warning( + "operator_ai_alert_card_delivery_readback_cache_store_unavailable", + project_id=normalized_project_id, + event_type=normalized_event_type, + lane=normalized_lane, + page=normalized_page, + per_page=normalized_per_page, + error_type=type(exc).__name__, + ) + response["summary"]["cache_write_status"] = "unavailable" + return response + + +def _ai_alert_card_delivery_source_unavailable_response( + *, + project_id: str, + event_type: str | None, + lane: str | None, + page: int, + per_page: int, + error_type: str, +) -> dict[str, Any]: + summary = _ai_alert_card_delivery_summary_from_row( + {}, + project_id=project_id, + event_type=event_type, + lane=lane, ) + learning_registry = _ai_alert_card_learning_registry_readback( + items=[], + project_id=project_id, + ) + for target in learning_registry["targets"]: + target["status"] = "blocked_by_ai_alert_card_delivery_readback_unavailable" + target["next_action"] = "repair_awooop_ai_alert_card_delivery_readback" + + summary.update({ + "status": "source_unavailable_ai_controlled_repair_required", + "db_read_status": "unavailable", + "source_unavailable_reason": error_type, + "source_unavailable_next_action": ( + "repair_awooop_ai_alert_card_delivery_readback" + ), + "controlled_repair_queue_present": True, + "runtime_write_allowed": False, + "learning_writeback_next_action": ( + "repair_awooop_ai_alert_card_delivery_readback" + ), + "learning_registry_target_count": learning_registry["target_count"], + "learning_registry_ready_target_count": 0, + "learning_registry_missing_target_count": learning_registry["target_count"], + "learning_registry_next_action": ( + "repair_awooop_ai_alert_card_delivery_readback_then_consume_learning_registry" + ), + }) + learning_registry.update({ + "status": "source_unavailable_ai_controlled_repair_required", + "ready_target_count": 0, + "missing_target_count": learning_registry["target_count"], + "ready_receipt_ref_count": 0, + "next_action": ( + "repair_awooop_ai_alert_card_delivery_readback_then_consume_learning_registry" + ), + }) + return { + "items": [], + "total": 0, + "page": page, + "per_page": per_page, + "summary": summary, + "learning_registry": learning_registry, + } def _ai_alert_card_delivery_summary_from_row( diff --git a/apps/api/src/services/telegram_alert_monitoring_coverage_readback.py b/apps/api/src/services/telegram_alert_monitoring_coverage_readback.py index d073ac2a1..c89ad0c1d 100644 --- a/apps/api/src/services/telegram_alert_monitoring_coverage_readback.py +++ b/apps/api/src/services/telegram_alert_monitoring_coverage_readback.py @@ -455,10 +455,19 @@ async def _load_ai_alert_card_delivery(*, project_id: str) -> dict[str, Any]: timeout=_LIVE_READBACK_TIMEOUT_SECONDS, ) summary = _dict(payload.get("summary")) + learning_registry = _dict(payload.get("learning_registry")) + db_read_status = str(summary.get("db_read_status") or "ok") + source_status = str(summary.get("status") or "") return { - "status": "ok", + "status": ( + "unavailable" + if db_read_status == "unavailable" + or source_status.startswith("source_unavailable") + else "ok" + ), "summary": summary, "total": _int(payload.get("total")), + "learning_registry": learning_registry, } except Exception as exc: # pragma: no cover - live DB pressure logger.warning( @@ -524,10 +533,12 @@ async def _load_runtime_log_readback_once(*, project_id: str) -> dict[str, Any]: COUNT(*) AS outbound_telegram_total, COUNT(*) FILTER (WHERE send_status = 'sent') AS outbound_sent_total, COUNT(*) FILTER ( - WHERE source_envelope ? 'ai_automation_alert_card' + WHERE COALESCE(source_envelope::jsonb, '{}'::jsonb) + ? 'ai_automation_alert_card' ) AS outbound_ai_alert_card_total, COUNT(*) FILTER ( - WHERE source_envelope ? 'source_refs' + WHERE COALESCE(source_envelope::jsonb, '{}'::jsonb) + ? 'source_refs' ) AS outbound_source_refs_total, MAX(COALESCE(sent_at, queued_at)) AS latest_outbound_at FROM awooop_outbound_message diff --git a/apps/api/tests/test_awooop_operator_timeline_labels.py b/apps/api/tests/test_awooop_operator_timeline_labels.py index 3abec9da5..7f4a8f407 100644 --- a/apps/api/tests/test_awooop_operator_timeline_labels.py +++ b/apps/api/tests/test_awooop_operator_timeline_labels.py @@ -21,6 +21,8 @@ from src.services.ollama_failover_manager import OllamaEndpoint, OllamaRoutingRe from src.services.ollama_health_monitor import HealthReport, HealthStatus from src.services.platform_operator_service import ( _RUN_CONTEXT_QUERY_CHUNK_SIZE, + _ai_alert_card_delivery_item, + _ai_alert_card_delivery_summary_from_row, _ai_route_health_map, _ai_route_lane_state, _ai_route_policy_order, @@ -32,8 +34,6 @@ from src.services.platform_operator_service import ( _cicd_duration_seconds, _cicd_event_item_from_row, _collect_run_incident_ids, - _ai_alert_card_delivery_item, - _ai_alert_card_delivery_summary_from_row, _is_source_correlation_applied_link, _iter_run_context_batches, _legacy_mcp_timeline_status, @@ -46,8 +46,8 @@ from src.services.platform_operator_service import ( _recent_event_source_summary, _remediation_summary_matches_incident_id, _remediation_summary_matches_status, - _repair_candidate_projection_from_metadata, _remediation_timeline_summary, + _repair_candidate_projection_from_metadata, _run_callback_reply_summary, _run_remediation_list_summary, _score_source_correlation_event, @@ -1253,6 +1253,81 @@ def test_list_ai_alert_cards_response_preserves_delivery_metadata() -> None: assert dumped["summary"]["production_write_count"] == 0 +def test_ai_alert_card_source_unavailable_response_exposes_ai_repair_queue() -> None: + response = ListAiAlertCardsResponse.model_validate( + platform_operator_service._ai_alert_card_delivery_source_unavailable_response( + project_id="awoooi", + event_type=None, + lane=None, + page=1, + per_page=6, + error_type="ProgrammingError", + ) + ) + + dumped = response.model_dump(mode="json") + assert dumped["total"] == 0 + assert dumped["summary"]["status"] == ( + "source_unavailable_ai_controlled_repair_required" + ) + assert dumped["summary"]["db_read_status"] == "unavailable" + assert dumped["summary"]["controlled_repair_queue_present"] is True + assert dumped["summary"]["learning_registry_target_count"] == 6 + assert dumped["summary"]["learning_registry_missing_target_count"] == 6 + assert dumped["summary"]["source_unavailable_next_action"] == ( + "repair_awooop_ai_alert_card_delivery_readback" + ) + assert dumped["learning_registry"]["status"] == ( + "source_unavailable_ai_controlled_repair_required" + ) + assert dumped["learning_registry"]["missing_target_count"] == 6 + assert {target["target"] for target in dumped["learning_registry"]["targets"]} == { + "km", + "playbook", + "rag", + "mcp", + "verifier", + "ai_agent", + } + assert all( + target["status"] + == "blocked_by_ai_alert_card_delivery_readback_unavailable" + for target in dumped["learning_registry"]["targets"] + ) + assert dumped["learning_registry"]["operation_boundaries"]["secret_value_read"] is False + assert dumped["learning_registry"]["operation_boundaries"]["github_api_used"] is False + + +@pytest.mark.asyncio +async def test_list_ai_alert_card_delivery_readback_failsofts_db_unavailable( + monkeypatch: pytest.MonkeyPatch, +) -> None: + class FailingDbContext: + async def __aenter__(self): + raise RuntimeError("db unavailable") + + async def __aexit__(self, exc_type, exc, tb): + return False + + monkeypatch.setattr( + platform_operator_service, + "get_db_context", + lambda _project_id: FailingDbContext(), + ) + + payload = await platform_operator_service.list_ai_alert_card_delivery_readback( + project_id="awoooi", + page=1, + per_page=6, + refresh=True, + ) + + assert payload["total"] == 0 + assert payload["summary"]["db_read_status"] == "unavailable" + assert payload["summary"]["controlled_repair_queue_present"] is True + assert payload["learning_registry"]["missing_target_count"] == 6 + + def test_list_callback_replies_keeps_audit_summary_separate_from_km_summary() -> None: source = inspect.getsource(platform_operator_service.list_callback_replies) diff --git a/apps/api/tests/test_telegram_alert_monitoring_coverage_readback_api.py b/apps/api/tests/test_telegram_alert_monitoring_coverage_readback_api.py index d986e95b7..21d550106 100644 --- a/apps/api/tests/test_telegram_alert_monitoring_coverage_readback_api.py +++ b/apps/api/tests/test_telegram_alert_monitoring_coverage_readback_api.py @@ -9,6 +9,7 @@ from fastapi.testclient import TestClient from src.api.v1 import agents from src.api.v1.agents import router +from src.services import platform_operator_service from src.services import telegram_alert_monitoring_coverage_readback as coverage_service from src.services.telegram_alert_monitoring_coverage_readback import ( _inspect_source_contract, @@ -353,6 +354,38 @@ async def test_telegram_alert_monitoring_live_readbacks_are_bounded( assert runtime_log["event_type_counts_7d"] == {} +@pytest.mark.asyncio +async def test_telegram_alert_monitoring_marks_ai_alert_cards_source_unavailable( + monkeypatch: pytest.MonkeyPatch, +): + async def source_unavailable_ai_alert_cards(**_kwargs): + return ( + platform_operator_service + ._ai_alert_card_delivery_source_unavailable_response( + project_id="awoooi", + event_type=None, + lane=None, + page=1, + per_page=6, + error_type="ProgrammingError", + ) + ) + + monkeypatch.setattr( + coverage_service, + "list_ai_alert_card_delivery_readback", + source_unavailable_ai_alert_cards, + ) + + ai_cards = await coverage_service._load_ai_alert_card_delivery(project_id="awoooi") + + assert ai_cards["status"] == "unavailable" + assert ai_cards["summary"]["db_read_status"] == "unavailable" + assert ai_cards["learning_registry"]["status"] == ( + "source_unavailable_ai_controlled_repair_required" + ) + + def test_telegram_alert_monitoring_coverage_source_contract(tmp_path): (tmp_path / "apps/api/src/api/v1").mkdir(parents=True) (tmp_path / "apps/api/src/repositories").mkdir(parents=True) From 13faee8fd48a5687e167bc9e7c570b458668db3a Mon Sep 17 00:00:00 2001 From: ogt Date: Thu, 9 Jul 2026 22:36:09 +0800 Subject: [PATCH 03/16] fix(iwooos): count alertmanager receipts in security readback --- apps/api/src/api/v1/iwooos.py | 26 +++++++++- .../iwooos_security_operating_system.py | 45 +++++++++++++++-- .../test_iwooos_security_operating_system.py | 48 +++++++++++++++++++ 3 files changed, 115 insertions(+), 4 deletions(-) diff --git a/apps/api/src/api/v1/iwooos.py b/apps/api/src/api/v1/iwooos.py index ed92e61b5..04405fbd8 100644 --- a/apps/api/src/api/v1/iwooos.py +++ b/apps/api/src/api/v1/iwooos.py @@ -74,6 +74,7 @@ from src.services.iwooos_wazuh_runtime_gate_owner_review_readback import ( from src.services.iwooos_wazuh_runtime_gate_owner_review_readback import ( validate_iwooos_wazuh_runtime_gate_owner_review_packet as validate_wazuh_runtime_gate_owner_review_packet_payload, ) +from src.services.platform_operator_service import list_recent_channel_events from src.services.public_redaction import redact_public_lan_topology router = APIRouter(tags=["IwoooS Security"]) @@ -84,6 +85,25 @@ async def _wazuh_readonly_status() -> JSONResponse: return JSONResponse(status_code=result.http_status, content=result.payload) +async def _alertmanager_receipt_readback() -> dict[str, Any]: + try: + payload = await list_recent_channel_events( + project_id="awoooi", + channel_type=None, + provider_prefix="alertmanager", + limit=20, + ) + return {**payload, "source_status": payload.get("source_status", "ready")} + except Exception as exc: + return { + "events": [], + "total": 0, + "limit": 20, + "source_status": "source_unavailable", + "source_error": exc.__class__.__name__, + } + + @router.get("/api/iwooos/wazuh") async def get_iwooos_wazuh_readonly_status_compat() -> JSONResponse: return await _wazuh_readonly_status() @@ -108,7 +128,11 @@ async def get_iwooos_wazuh_readonly_status_v1() -> JSONResponse: async def get_iwooos_security_operating_system() -> dict[str, Any]: """回傳 IwoooS 資安作業系統公開安全總入口。""" try: - payload = await asyncio.to_thread(load_latest_iwooos_security_operating_system) + alert_receipt_readback = await _alertmanager_receipt_readback() + payload = await asyncio.to_thread( + load_latest_iwooos_security_operating_system, + alert_receipt_readback=alert_receipt_readback, + ) return redact_public_lan_topology(payload) except FileNotFoundError as exc: raise HTTPException( diff --git a/apps/api/src/services/iwooos_security_operating_system.py b/apps/api/src/services/iwooos_security_operating_system.py index a1b9c0b07..5b6d14c63 100644 --- a/apps/api/src/services/iwooos_security_operating_system.py +++ b/apps/api/src/services/iwooos_security_operating_system.py @@ -156,6 +156,7 @@ _RUNTIME_ACTION_KEYS = { def load_latest_iwooos_security_operating_system( security_dir: Path | None = None, + alert_receipt_readback: dict[str, Any] | None = None, ) -> dict[str, Any]: """Load the committed IwoooS security operating system contract.""" directory = security_dir or _DEFAULT_SECURITY_DIR @@ -163,6 +164,7 @@ def load_latest_iwooos_security_operating_system( _require_boundaries(snapshot) summary = _summary(snapshot) + alert_receipt_summary = _alert_receipt_summary(alert_receipt_readback) merged_summary = { "reference_framework_count": _int(summary.get("reference_framework_count")), "operating_role_count": _int(summary.get("operating_role_count")), @@ -192,9 +194,13 @@ def load_latest_iwooos_security_operating_system( "wazuh_registry_accepted_count": _int( summary.get("wazuh_registry_accepted_count") ), - "alert_receipt_accepted_count": _int( - summary.get("alert_receipt_accepted_count") + "alert_receipt_accepted_count": max( + _int(summary.get("alert_receipt_accepted_count")), + alert_receipt_summary["accepted_count"], ), + "alert_receipt_observed_count": alert_receipt_summary["observed_count"], + "alert_receipt_source_status": alert_receipt_summary["source_status"], + "alert_receipt_source_error": alert_receipt_summary["source_error"], "incident_case_accepted_count": _int( summary.get("incident_case_accepted_count") ), @@ -461,6 +467,37 @@ def _int(value: Any) -> int: return value if isinstance(value, int) else 0 +def _alert_receipt_summary(readback: dict[str, Any] | None) -> dict[str, Any]: + if not isinstance(readback, dict): + return { + "accepted_count": 0, + "observed_count": 0, + "source_status": "not_connected", + "source_error": None, + } + + source_status = str(readback.get("source_status") or "ready") + source_error = readback.get("source_error") + events = readback.get("events") + total = readback.get("total") + observed_count = ( + _int(total) + if total is not None + else len(events) + if isinstance(events, list) + else 0 + ) + accepted_count = ( + observed_count if source_status == "ready" and observed_count > 0 else 0 + ) + return { + "accepted_count": accepted_count, + "observed_count": observed_count, + "source_status": source_status, + "source_error": str(source_error) if source_error else None, + } + + def _strings(value: Any) -> list[str]: if not isinstance(value, list): return [] @@ -612,7 +649,7 @@ def _checkpoints(value: Any) -> list[dict[str, Any]]: return items -def _boundary_markers(summary: dict[str, int]) -> list[str]: +def _boundary_markers(summary: dict[str, Any]) -> list[str]: return [ "iwooos_security_operating_system_api_visible=true", "iwooos_security_operation_packet_validation_api_available=true", @@ -627,6 +664,8 @@ def _boundary_markers(summary: dict[str, int]) -> list[str]: f"iwooos_security_operating_system_operation_packet_required_field_count={summary['operation_packet_required_field_count']}", f"iwooos_security_operating_system_evidence_weighted_percent={summary['evidence_weighted_security_operating_system_percent']}", f"iwooos_security_operating_system_wazuh_registry_accepted_count={summary['wazuh_registry_accepted_count']}", + f"iwooos_security_operating_system_alert_receipt_accepted_count={summary['alert_receipt_accepted_count']}", + f"iwooos_security_operating_system_alert_receipt_observed_count={summary['alert_receipt_observed_count']}", "iwooos_security_operating_system_operation_packet_validation_no_persist=true", "iwooos_security_operating_system_runtime_gate_count=0", "runtime_execution_authorized=false", diff --git a/apps/api/tests/test_iwooos_security_operating_system.py b/apps/api/tests/test_iwooos_security_operating_system.py index 30ab94947..1a8ff2582 100644 --- a/apps/api/tests/test_iwooos_security_operating_system.py +++ b/apps/api/tests/test_iwooos_security_operating_system.py @@ -71,6 +71,9 @@ def test_iwooos_security_operating_system_readback_exposes_api_validator() -> No assert payload["summary"]["operation_packet_validator_available_count"] == 1 assert payload["summary"]["operation_packet_required_field_count"] == 24 assert payload["summary"]["wazuh_registry_accepted_count"] == 6 + assert payload["summary"]["alert_receipt_accepted_count"] == 0 + assert payload["summary"]["alert_receipt_observed_count"] == 0 + assert payload["summary"]["alert_receipt_source_status"] == "not_connected" assert payload["summary"]["runtime_gate_count"] == 0 assert payload["operation_packet_validation_endpoint"] == ( "/api/v1/iwooos/security-operating-system/validate-operation-packet" @@ -87,6 +90,26 @@ def test_iwooos_security_operating_system_readback_exposes_api_validator() -> No ) +def test_iwooos_security_operating_system_accepts_alert_receipt_readback() -> None: + payload = load_latest_iwooos_security_operating_system( + alert_receipt_readback={ + "events": [{"provider_event_id": "alertmanager:received:alert-a"}], + "total": 1, + "limit": 20, + "source_status": "ready", + } + ) + + assert payload["summary"]["alert_receipt_accepted_count"] == 1 + assert payload["summary"]["alert_receipt_observed_count"] == 1 + assert payload["summary"]["alert_receipt_source_status"] == "ready" + assert payload["summary"]["runtime_gate_count"] == 0 + assert any( + marker == "iwooos_security_operating_system_alert_receipt_accepted_count=1" + for marker in payload["boundary_markers"] + ) + + def test_iwooos_security_operating_system_api_is_public_safe() -> None: response = _client().get("/api/v1/iwooos/security-operating-system") @@ -103,6 +126,31 @@ def test_iwooos_security_operating_system_api_is_public_safe() -> None: assert "WAZUH_API_PASSWORD" not in response.text +def test_iwooos_security_operating_system_api_counts_alertmanager_receipts( + monkeypatch, +) -> None: + async def fake_recent_events(**_: object) -> dict[str, object]: + return { + "events": [{"provider_event_id": "alertmanager:received:alert-a"}], + "total": 1, + "limit": 20, + } + + monkeypatch.setattr( + "src.api.v1.iwooos.list_recent_channel_events", + fake_recent_events, + ) + + response = _client().get("/api/v1/iwooos/security-operating-system") + + assert response.status_code == 200 + data = response.json() + assert data["summary"]["alert_receipt_accepted_count"] == 1 + assert data["summary"]["alert_receipt_observed_count"] == 1 + assert data["summary"]["alert_receipt_source_status"] == "ready" + assert data["summary"]["runtime_gate_count"] == 0 + + def test_iwooos_security_operation_packet_validator_accepts_redacted_loop() -> None: payload = validate_iwooos_security_operation_packet(_valid_operation_packet()) From facb3466bd43476d37d4b1ee6c71d4118ce8031b Mon Sep 17 00:00:00 2001 From: AWOOOI CD Date: Thu, 9 Jul 2026 22:42:13 +0800 Subject: [PATCH 04/16] chore(cd): deploy 13faee8 [skip ci] --- k8s/awoooi-prod/06-deployment-api.yaml | 4 ++-- k8s/awoooi-prod/kustomization.yaml | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/k8s/awoooi-prod/06-deployment-api.yaml b/k8s/awoooi-prod/06-deployment-api.yaml index 14a84b213..a92cd8d25 100644 --- a/k8s/awoooi-prod/06-deployment-api.yaml +++ b/k8s/awoooi-prod/06-deployment-api.yaml @@ -84,12 +84,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: "7ee0a17ab7b6842dc8eb2104350cbab300d05243" + value: "13faee8fd48a5687e167bc9e7c570b458668db3a" - 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: "7ee0a17ab7b6842dc8eb2104350cbab300d05243" + value: "13faee8fd48a5687e167bc9e7c570b458668db3a" - 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/kustomization.yaml b/k8s/awoooi-prod/kustomization.yaml index 421dbb644..d989f593d 100644 --- a/k8s/awoooi-prod/kustomization.yaml +++ b/k8s/awoooi-prod/kustomization.yaml @@ -41,7 +41,7 @@ resources: images: - name: 192.168.0.110:5000/library/api:IMAGE_TAG_PLACEHOLDER newName: 192.168.0.110:5000/awoooi/api - newTag: 7ee0a17ab7b6842dc8eb2104350cbab300d05243 + newTag: 13faee8fd48a5687e167bc9e7c570b458668db3a - name: 192.168.0.110:5000/library/web:IMAGE_TAG_PLACEHOLDER newName: 192.168.0.110:5000/awoooi/web - newTag: 7ee0a17ab7b6842dc8eb2104350cbab300d05243 + newTag: 13faee8fd48a5687e167bc9e7c570b458668db3a From f125589f7d8564296818192db67d35e41e586f2f Mon Sep 17 00:00:00 2001 From: ogt Date: Thu, 9 Jul 2026 22:46:32 +0800 Subject: [PATCH 05/16] fix(iwooos): retry alert receipt security readback --- apps/api/src/api/v1/iwooos.py | 37 ++++++++++++++++++++--------------- 1 file changed, 21 insertions(+), 16 deletions(-) diff --git a/apps/api/src/api/v1/iwooos.py b/apps/api/src/api/v1/iwooos.py index 04405fbd8..55417dfe6 100644 --- a/apps/api/src/api/v1/iwooos.py +++ b/apps/api/src/api/v1/iwooos.py @@ -86,22 +86,27 @@ async def _wazuh_readonly_status() -> JSONResponse: async def _alertmanager_receipt_readback() -> dict[str, Any]: - try: - payload = await list_recent_channel_events( - project_id="awoooi", - channel_type=None, - provider_prefix="alertmanager", - limit=20, - ) - return {**payload, "source_status": payload.get("source_status", "ready")} - except Exception as exc: - return { - "events": [], - "total": 0, - "limit": 20, - "source_status": "source_unavailable", - "source_error": exc.__class__.__name__, - } + last_error: Exception | None = None + for attempt in range(2): + try: + payload = await list_recent_channel_events( + project_id="awoooi", + channel_type=None, + provider_prefix="alertmanager", + limit=5, + ) + return {**payload, "source_status": payload.get("source_status", "ready")} + except Exception as exc: + last_error = exc + if attempt == 0: + await asyncio.sleep(0.05) + return { + "events": [], + "total": 0, + "limit": 5, + "source_status": "source_unavailable", + "source_error": last_error.__class__.__name__ if last_error else "UnknownError", + } @router.get("/api/iwooos/wazuh") From abd7ec08157d1583c8f1928519e43131663c2f68 Mon Sep 17 00:00:00 2001 From: ogt Date: Thu, 9 Jul 2026 22:47:29 +0800 Subject: [PATCH 06/16] fix(agents): bound AI alert card DB readback --- .../src/services/platform_operator_service.py | 51 +++++++++++++++++-- .../test_awooop_operator_timeline_labels.py | 45 ++++++++++++++++ 2 files changed, 93 insertions(+), 3 deletions(-) diff --git a/apps/api/src/services/platform_operator_service.py b/apps/api/src/services/platform_operator_service.py index bfefab1ed..94a90bf6c 100644 --- a/apps/api/src/services/platform_operator_service.py +++ b/apps/api/src/services/platform_operator_service.py @@ -99,6 +99,18 @@ _CALLBACK_REPLY_CACHE_TTL_SECONDS = int( _AI_ALERT_CARD_CACHE_TTL_SECONDS = int( os.getenv("AWOOOP_AI_ALERT_CARD_CACHE_TTL_SECONDS", "20") ) +_AI_ALERT_CARD_DB_CONTEXT_TIMEOUT_SECONDS = float( + os.getenv("AWOOOP_AI_ALERT_CARD_DB_CONTEXT_TIMEOUT_SECONDS", "0.25") +) +_AI_ALERT_CARD_DB_CONTEXT_EXIT_TIMEOUT_SECONDS = float( + os.getenv("AWOOOP_AI_ALERT_CARD_DB_CONTEXT_EXIT_TIMEOUT_SECONDS", "0.20") +) +_AI_ALERT_CARD_QUERY_TIMEOUT_SECONDS = float( + os.getenv("AWOOOP_AI_ALERT_CARD_QUERY_TIMEOUT_SECONDS", "0.25") +) +_AI_ALERT_CARD_STATEMENT_TIMEOUT_MS = int( + os.getenv("AWOOOP_AI_ALERT_CARD_STATEMENT_TIMEOUT_MS", "200") +) _AI_ALERT_CARD_LEARNING_REGISTRY_TARGETS: tuple[dict[str, str], ...] = ( { "target": "km", @@ -1476,11 +1488,44 @@ async def list_ai_alert_card_delivery_readback( """) try: - async with get_db_context(normalized_project_id) as db: - summary_result = await db.execute(summary_sql, params) + db_context = get_db_context(normalized_project_id) + db = await asyncio.wait_for( + db_context.__aenter__(), + timeout=_AI_ALERT_CARD_DB_CONTEXT_TIMEOUT_SECONDS, + ) + try: + await asyncio.wait_for( + db.execute( + text( + "SET LOCAL statement_timeout = " + f"'{int(_AI_ALERT_CARD_STATEMENT_TIMEOUT_MS)}ms'" + ), + {}, + ), + timeout=_AI_ALERT_CARD_QUERY_TIMEOUT_SECONDS, + ) + summary_result = await asyncio.wait_for( + db.execute(summary_sql, params), + timeout=_AI_ALERT_CARD_QUERY_TIMEOUT_SECONDS, + ) summary_row = summary_result.mappings().first() or {} - rows_result = await db.execute(list_sql, params) + rows_result = await asyncio.wait_for( + db.execute(list_sql, params), + timeout=_AI_ALERT_CARD_QUERY_TIMEOUT_SECONDS, + ) rows = list(rows_result.mappings().all()) + finally: + try: + await asyncio.wait_for( + db_context.__aexit__(None, None, None), + timeout=_AI_ALERT_CARD_DB_CONTEXT_EXIT_TIMEOUT_SECONDS, + ) + except Exception as exit_exc: # pragma: no cover - pool pressure + logger.warning( + "operator_ai_alert_card_delivery_readback_context_exit_unavailable", + project_id=normalized_project_id, + error_type=type(exit_exc).__name__, + ) except Exception as exc: # pragma: no cover - production DB drift/pressure logger.warning( "operator_ai_alert_card_delivery_readback_source_unavailable", diff --git a/apps/api/tests/test_awooop_operator_timeline_labels.py b/apps/api/tests/test_awooop_operator_timeline_labels.py index 7f4a8f407..dcc398951 100644 --- a/apps/api/tests/test_awooop_operator_timeline_labels.py +++ b/apps/api/tests/test_awooop_operator_timeline_labels.py @@ -1328,6 +1328,51 @@ async def test_list_ai_alert_card_delivery_readback_failsofts_db_unavailable( assert payload["learning_registry"]["missing_target_count"] == 6 +@pytest.mark.asyncio +async def test_list_ai_alert_card_delivery_readback_failsofts_slow_db( + monkeypatch: pytest.MonkeyPatch, +) -> None: + class SlowDb: + async def execute(self, *_args, **_kwargs): + await asyncio.sleep(1) + + class SlowDbContext: + async def __aenter__(self): + return SlowDb() + + async def __aexit__(self, exc_type, exc, tb): + return False + + monkeypatch.setattr( + platform_operator_service, + "_AI_ALERT_CARD_QUERY_TIMEOUT_SECONDS", + 0.01, + ) + monkeypatch.setattr( + platform_operator_service, + "_AI_ALERT_CARD_DB_CONTEXT_EXIT_TIMEOUT_SECONDS", + 0.01, + ) + monkeypatch.setattr( + platform_operator_service, + "get_db_context", + lambda _project_id: SlowDbContext(), + ) + + payload = await platform_operator_service.list_ai_alert_card_delivery_readback( + project_id="awoooi", + page=1, + per_page=6, + refresh=True, + ) + + assert payload["summary"]["db_read_status"] == "unavailable" + assert payload["summary"]["source_unavailable_reason"] == "TimeoutError" + assert payload["summary"]["learning_registry_next_action"] == ( + "repair_awooop_ai_alert_card_delivery_readback_then_consume_learning_registry" + ) + + def test_list_callback_replies_keeps_audit_summary_separate_from_km_summary() -> None: source = inspect.getsource(platform_operator_service.list_callback_replies) From 0fe3b4b59349b8110fb322572f37323c4366e4f0 Mon Sep 17 00:00:00 2001 From: ogt Date: Thu, 9 Jul 2026 22:48:17 +0800 Subject: [PATCH 07/16] fix(agents): close trace ledger repair receipt gap --- .../ai_agent_autonomous_runtime_control.py | 161 ++++++++++++++++-- ...est_ai_agent_autonomous_runtime_control.py | 49 ++++++ 2 files changed, 200 insertions(+), 10 deletions(-) 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 4348aa0d3..1c97344fc 100644 --- a/apps/api/src/services/ai_agent_autonomous_runtime_control.py +++ b/apps/api/src/services/ai_agent_autonomous_runtime_control.py @@ -490,6 +490,8 @@ def _trace_stage( required_for_closed_loop: bool, feeds_learning: bool, public_safe: bool = True, + record_quality: str | None = None, + evidence_note: str | None = None, next_action_if_missing: str | None = None, ) -> dict[str, Any]: present = total > 0 @@ -498,12 +500,13 @@ def _trace_stage( "display_name": display_name, "source_tables": source_tables, "recorded": present, - "record_quality": "recorded" if present else "missing", + "record_quality": record_quality or ("recorded" if present else "missing"), "total": max(0, total), "recent": max(0, recent), "required_for_closed_loop": required_for_closed_loop, "feeds_learning": feeds_learning, "public_safe": public_safe, + "evidence_note": evidence_note, "next_action_if_missing": None if present else next_action_if_missing, } @@ -530,6 +533,48 @@ def _trace_recent(summary: Mapping[str, Any] | None, *operation_types: str) -> i ) +def _controlled_apply_receipt_chain_fallback_total( + *, + operation_summary: Mapping[str, Any], + verifier_summary: Mapping[str, Any], + km_summary: Mapping[str, Any], + telegram_summary: Mapping[str, Any], +) -> int: + """Count closed apply chains when the legacy auto_repair_executions table is empty.""" + + counts = [ + _trace_total(operation_summary, "ansible_apply_executed"), + _trace_total(verifier_summary), + _trace_total(km_summary), + _trace_total(telegram_summary), + ] + if any(count <= 0 for count in counts): + return 0 + return min(counts) + + +def _controlled_apply_receipt_chain_fallback_recent( + *, + operation_summary: Mapping[str, Any], + verifier_summary: Mapping[str, Any], + km_summary: Mapping[str, Any], + telegram_summary: Mapping[str, Any], +) -> int: + if _controlled_apply_receipt_chain_fallback_total( + operation_summary=operation_summary, + verifier_summary=verifier_summary, + km_summary=km_summary, + telegram_summary=telegram_summary, + ) <= 0: + return 0 + return min( + _trace_recent(operation_summary, "ansible_apply_executed"), + _trace_recent(verifier_summary), + _trace_recent(km_summary), + _trace_recent(telegram_summary), + ) + + def _build_trace_ledger( *, operation_summary: Mapping[str, Any], @@ -550,6 +595,33 @@ def _build_trace_ledger( mcp_total = _trace_total(mcp_gateway_summary) + _trace_total(legacy_mcp_summary) mcp_recent = _trace_recent(mcp_gateway_summary) + _trace_recent(legacy_mcp_summary) + auto_repair_total = _trace_total(auto_repair_summary) + auto_repair_recent = _trace_recent(auto_repair_summary) + auto_repair_fallback_total = ( + _controlled_apply_receipt_chain_fallback_total( + operation_summary=operation_summary, + verifier_summary=verifier_summary, + km_summary=km_summary, + telegram_summary=telegram_summary, + ) + if auto_repair_total <= 0 + else 0 + ) + auto_repair_fallback_recent = ( + _controlled_apply_receipt_chain_fallback_recent( + operation_summary=operation_summary, + verifier_summary=verifier_summary, + km_summary=km_summary, + telegram_summary=telegram_summary, + ) + if auto_repair_fallback_total > 0 + else 0 + ) + auto_repair_trace_total = max(auto_repair_total, auto_repair_fallback_total) + auto_repair_trace_recent = max(auto_repair_recent, auto_repair_fallback_recent) + auto_repair_fallback_applied = ( + auto_repair_total <= 0 and auto_repair_fallback_total > 0 + ) stages = [ _trace_stage( stage_id="mcp_context", @@ -618,13 +690,36 @@ def _build_trace_ledger( ), _trace_stage( stage_id="auto_repair_execution_receipt", - display_name="Auto-repair execution receipt", - source_tables=["auto_repair_executions"], - total=_trace_total(auto_repair_summary), - recent=_trace_recent(auto_repair_summary), + display_name="Auto-repair / controlled apply execution receipt", + source_tables=( + [ + "auto_repair_executions", + "automation_operation_log:ansible_apply_executed", + "incident_evidence.post_execution_state", + "knowledge_entries", + "awooop_outbound_message", + ] + if auto_repair_fallback_applied + else ["auto_repair_executions"] + ), + total=auto_repair_trace_total, + recent=auto_repair_trace_recent, required_for_closed_loop=True, feeds_learning=True, - next_action_if_missing="receipt_backfill_records_auto_repair_execution", + record_quality=( + "controlled_apply_receipt_chain_fallback" + if auto_repair_fallback_applied + else None + ), + evidence_note=( + "auto_repair_executions is empty, but apply/verifier/KM/Telegram " + "receipts prove a controlled apply execution chain." + if auto_repair_fallback_applied + else None + ), + next_action_if_missing=( + "record_auto_repair_execution_or_complete_apply_verifier_km_telegram_chain" + ), ), _trace_stage( stage_id="post_apply_verifier", @@ -734,6 +829,33 @@ def _build_log_integration_taxonomy( operation_total = sum(_trace_total(operation_summary, item) for item in _EXECUTOR_OPERATION_TYPES) operation_recent = sum(_trace_recent(operation_summary, item) for item in _EXECUTOR_OPERATION_TYPES) + auto_repair_total = _trace_total(auto_repair_summary) + auto_repair_recent = _trace_recent(auto_repair_summary) + auto_repair_fallback_total = ( + _controlled_apply_receipt_chain_fallback_total( + operation_summary=operation_summary, + verifier_summary=verifier_summary, + km_summary=km_summary, + telegram_summary=telegram_summary, + ) + if auto_repair_total <= 0 + else 0 + ) + auto_repair_fallback_recent = ( + _controlled_apply_receipt_chain_fallback_recent( + operation_summary=operation_summary, + verifier_summary=verifier_summary, + km_summary=km_summary, + telegram_summary=telegram_summary, + ) + if auto_repair_fallback_total > 0 + else 0 + ) + auto_repair_source_total = max(auto_repair_total, auto_repair_fallback_total) + auto_repair_source_recent = max(auto_repair_recent, auto_repair_fallback_recent) + auto_repair_fallback_applied = ( + auto_repair_total <= 0 and auto_repair_fallback_total > 0 + ) source_families = [ { "source_family_id": "mcp_gateway_tool_calls", @@ -798,15 +920,34 @@ def _build_log_integration_taxonomy( }, { "source_family_id": "auto_repair_receipts", - "source_tables": ["auto_repair_executions"], + "source_tables": ( + [ + "auto_repair_executions", + "automation_operation_log:ansible_apply_executed", + "incident_evidence.post_execution_state", + "knowledge_entries", + "awooop_outbound_message", + ] + if auto_repair_fallback_applied + else ["auto_repair_executions"] + ), "normalized_event_schema": "RepairExecutionReceipt", "label_dimensions": ["incident", "service", "playbook", "risk", "result"], - "total": _trace_total(auto_repair_summary), - "recent": _trace_recent(auto_repair_summary), + "total": auto_repair_source_total, + "recent": auto_repair_source_recent, "feeds_learning": True, "public_safe": True, "raw_payload_policy": "execution_step_refs_not_raw_secrets", - "next_action_if_empty": "write_auto_repair_execution_receipt_after_apply", + "record_quality": ( + "controlled_apply_receipt_chain_fallback" + if auto_repair_fallback_applied + else "recorded" + if auto_repair_source_total > 0 + else "missing" + ), + "next_action_if_empty": ( + "record_auto_repair_execution_or_complete_apply_verifier_km_telegram_chain" + ), }, { "source_family_id": "post_apply_verifier", 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 6c18aecf0..edab3cd0c 100644 --- a/apps/api/tests/test_ai_agent_autonomous_runtime_control.py +++ b/apps/api/tests/test_ai_agent_autonomous_runtime_control.py @@ -1426,6 +1426,55 @@ def test_runtime_receipt_recovery_flags_zero_live_log_events(): assert readback["work_item_progress"]["rollups"]["not_started_count"] == 10 +def test_trace_ledger_uses_controlled_apply_chain_for_auto_repair_receipt_gap(): + readback = build_runtime_receipt_readback_from_rows( + project_id="awoooi", + db_read_status="ok", + operation_count_rows=[ + {"operation_type": "ansible_candidate_matched", "total": 1, "recent": 1}, + {"operation_type": "ansible_check_mode_executed", "total": 1, "recent": 1}, + {"operation_type": "ansible_apply_executed", "total": 1, "recent": 1}, + ], + verifier_count_rows=[ + {"verification_result": "success", "total": 1, "recent": 1}, + ], + km_count_rows=[ + {"status": "review", "total": 1, "recent": 1}, + ], + telegram_count_rows=[ + {"send_status": "sent", "total": 1, "recent": 1}, + ], + ) + + trace = readback["trace_ledger"] + auto_repair_stage = next( + stage + for stage in trace["stages"] + if stage["stage_id"] == "auto_repair_execution_receipt" + ) + assert auto_repair_stage["recorded"] is True + assert auto_repair_stage["record_quality"] == ( + "controlled_apply_receipt_chain_fallback" + ) + assert "auto_repair_execution_receipt" not in trace["missing_required_stage_ids"] + + progress_items = { + item["work_item_id"]: item + for item in readback["work_item_progress"]["ordered_items"] + } + assert progress_items["P0-B-trace-ledger"]["status"] == "completed" + + auto_repair_source = next( + item + for item in readback["log_integration_taxonomy"]["source_families"] + if item["source_family_id"] == "auto_repair_receipts" + ) + assert auto_repair_source["total"] == 1 + assert auto_repair_source["record_quality"] == ( + "controlled_apply_receipt_chain_fallback" + ) + + def test_runtime_receipt_recovery_keeps_partial_live_log_events_visible(): readback = build_runtime_receipt_readback_from_rows( project_id="awoooi", From d6f65e0935a02a83c2cd148ebadabd73df187e57 Mon Sep 17 00:00:00 2001 From: AWOOOI CD Date: Thu, 9 Jul 2026 22:53:24 +0800 Subject: [PATCH 08/16] chore(cd): deploy 0fe3b4b [skip ci] --- k8s/awoooi-prod/06-deployment-api.yaml | 4 ++-- k8s/awoooi-prod/kustomization.yaml | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/k8s/awoooi-prod/06-deployment-api.yaml b/k8s/awoooi-prod/06-deployment-api.yaml index a92cd8d25..fade90438 100644 --- a/k8s/awoooi-prod/06-deployment-api.yaml +++ b/k8s/awoooi-prod/06-deployment-api.yaml @@ -84,12 +84,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: "13faee8fd48a5687e167bc9e7c570b458668db3a" + value: "0fe3b4b59349b8110fb322572f37323c4366e4f0" - 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: "13faee8fd48a5687e167bc9e7c570b458668db3a" + value: "0fe3b4b59349b8110fb322572f37323c4366e4f0" - 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/kustomization.yaml b/k8s/awoooi-prod/kustomization.yaml index d989f593d..7338f5edb 100644 --- a/k8s/awoooi-prod/kustomization.yaml +++ b/k8s/awoooi-prod/kustomization.yaml @@ -41,7 +41,7 @@ resources: images: - name: 192.168.0.110:5000/library/api:IMAGE_TAG_PLACEHOLDER newName: 192.168.0.110:5000/awoooi/api - newTag: 13faee8fd48a5687e167bc9e7c570b458668db3a + newTag: 0fe3b4b59349b8110fb322572f37323c4366e4f0 - name: 192.168.0.110:5000/library/web:IMAGE_TAG_PLACEHOLDER newName: 192.168.0.110:5000/awoooi/web - newTag: 13faee8fd48a5687e167bc9e7c570b458668db3a + newTag: 0fe3b4b59349b8110fb322572f37323c4366e4f0 From c22530205ea599c5b0ad3c38bee0a42e83eb6916 Mon Sep 17 00:00:00 2001 From: ogt Date: Thu, 9 Jul 2026 22:56:07 +0800 Subject: [PATCH 09/16] fix(agents): reuse cached AI alert card readback --- apps/api/src/api/v1/platform/operator_runs.py | 1 + .../src/services/platform_operator_service.py | 87 +++++++++++++++---- .../test_awooop_operator_timeline_labels.py | 78 +++++++++++++++++ 3 files changed, 149 insertions(+), 17 deletions(-) diff --git a/apps/api/src/api/v1/platform/operator_runs.py b/apps/api/src/api/v1/platform/operator_runs.py index a9991cdfb..606f4c010 100644 --- a/apps/api/src/api/v1/platform/operator_runs.py +++ b/apps/api/src/api/v1/platform/operator_runs.py @@ -160,6 +160,7 @@ class AiAlertCardDeliverySummary(BaseModel): source_unavailable_reason: str | None = None source_unavailable_next_action: str = "none" controlled_repair_queue_present: bool = False + live_refresh_status: str = "ok" total: int sent_total: int failed_total: int diff --git a/apps/api/src/services/platform_operator_service.py b/apps/api/src/services/platform_operator_service.py index 94a90bf6c..e498c601b 100644 --- a/apps/api/src/services/platform_operator_service.py +++ b/apps/api/src/services/platform_operator_service.py @@ -1362,23 +1362,14 @@ async def list_ai_alert_card_delivery_readback( "per_page": normalized_per_page, } if not refresh: - try: - cached_response = await get_cached_operator_summary_async( - "ai_alert_card_delivery_readback", - cache_key, - ttl_seconds=_AI_ALERT_CARD_CACHE_TTL_SECONDS, - ) - except Exception as exc: # pragma: no cover - cache pressure - logger.warning( - "operator_ai_alert_card_delivery_readback_cache_unavailable", - project_id=normalized_project_id, - event_type=normalized_event_type, - lane=normalized_lane, - page=normalized_page, - per_page=normalized_per_page, - error_type=type(exc).__name__, - ) - cached_response = None + cached_response = await _get_cached_ai_alert_card_delivery_readback( + cache_key=cache_key, + project_id=normalized_project_id, + event_type=normalized_event_type, + lane=normalized_lane, + page=normalized_page, + per_page=normalized_per_page, + ) if cached_response is not None and "learning_registry" in cached_response: logger.info( "operator_ai_alert_card_delivery_readback_cache_hit", @@ -1536,6 +1527,19 @@ async def list_ai_alert_card_delivery_readback( per_page=normalized_per_page, error_type=type(exc).__name__, ) + cached_fallback = await _get_cached_ai_alert_card_delivery_readback( + cache_key=cache_key, + project_id=normalized_project_id, + event_type=normalized_event_type, + lane=normalized_lane, + page=normalized_page, + per_page=normalized_per_page, + ) + if cached_fallback is not None and "learning_registry" in cached_fallback: + return _ai_alert_card_delivery_cached_live_timeout_response( + cached_response=cached_fallback, + error_type=type(exc).__name__, + ) return _ai_alert_card_delivery_source_unavailable_response( project_id=normalized_project_id, event_type=normalized_event_type or None, @@ -1604,6 +1608,54 @@ async def list_ai_alert_card_delivery_readback( return response +async def _get_cached_ai_alert_card_delivery_readback( + *, + cache_key: Mapping[str, Any], + project_id: str, + event_type: str, + lane: str, + page: int, + per_page: int, +) -> dict[str, Any] | None: + try: + return await get_cached_operator_summary_async( + "ai_alert_card_delivery_readback", + cache_key, + ttl_seconds=_AI_ALERT_CARD_CACHE_TTL_SECONDS, + ) + except Exception as exc: # pragma: no cover - cache pressure + logger.warning( + "operator_ai_alert_card_delivery_readback_cache_unavailable", + project_id=project_id, + event_type=event_type, + lane=lane, + page=page, + per_page=per_page, + error_type=type(exc).__name__, + ) + return None + + +def _ai_alert_card_delivery_cached_live_timeout_response( + *, + cached_response: Mapping[str, Any], + error_type: str, +) -> dict[str, Any]: + response = dict(cached_response) + summary = dict(_as_dict(response.get("summary"))) + summary.update({ + "db_read_status": "cached_after_live_timeout", + "source_unavailable_reason": error_type, + "source_unavailable_next_action": ( + "retry_live_refresh_keep_cached_learning_registry" + ), + "live_refresh_status": "live_refresh_unavailable_serving_cached_readback", + "controlled_repair_queue_present": False, + }) + response["summary"] = summary + return response + + def _ai_alert_card_delivery_source_unavailable_response( *, project_id: str, @@ -1635,6 +1687,7 @@ def _ai_alert_card_delivery_source_unavailable_response( "repair_awooop_ai_alert_card_delivery_readback" ), "controlled_repair_queue_present": True, + "live_refresh_status": "live_refresh_unavailable_no_cached_readback", "runtime_write_allowed": False, "learning_writeback_next_action": ( "repair_awooop_ai_alert_card_delivery_readback" diff --git a/apps/api/tests/test_awooop_operator_timeline_labels.py b/apps/api/tests/test_awooop_operator_timeline_labels.py index dcc398951..f597fd5b9 100644 --- a/apps/api/tests/test_awooop_operator_timeline_labels.py +++ b/apps/api/tests/test_awooop_operator_timeline_labels.py @@ -1373,6 +1373,84 @@ async def test_list_ai_alert_card_delivery_readback_failsofts_slow_db( ) +@pytest.mark.asyncio +async def test_list_ai_alert_card_delivery_readback_uses_cache_after_live_timeout( + monkeypatch: pytest.MonkeyPatch, +) -> None: + class SlowDb: + async def execute(self, *_args, **_kwargs): + await asyncio.sleep(1) + + class SlowDbContext: + async def __aenter__(self): + return SlowDb() + + async def __aexit__(self, exc_type, exc, tb): + return False + + cached_response = ( + platform_operator_service._ai_alert_card_delivery_source_unavailable_response( + project_id="awoooi", + event_type=None, + lane=None, + page=1, + per_page=6, + error_type="seed", + ) + ) + cached_response["summary"].update({ + "status": "observed", + "db_read_status": "ok", + "total": 36, + "learning_registry_missing_target_count": 0, + }) + cached_response["learning_registry"].update({ + "status": "learning_registry_ready", + "ready_target_count": 6, + "missing_target_count": 0, + }) + + async def cached_summary(*_args, **_kwargs): + return cached_response + + monkeypatch.setattr( + platform_operator_service, + "_AI_ALERT_CARD_QUERY_TIMEOUT_SECONDS", + 0.01, + ) + monkeypatch.setattr( + platform_operator_service, + "_AI_ALERT_CARD_DB_CONTEXT_EXIT_TIMEOUT_SECONDS", + 0.01, + ) + monkeypatch.setattr( + platform_operator_service, + "get_db_context", + lambda _project_id: SlowDbContext(), + ) + monkeypatch.setattr( + platform_operator_service, + "get_cached_operator_summary_async", + cached_summary, + ) + + payload = await platform_operator_service.list_ai_alert_card_delivery_readback( + project_id="awoooi", + page=1, + per_page=6, + refresh=True, + ) + + assert payload["summary"]["status"] == "observed" + assert payload["summary"]["db_read_status"] == "cached_after_live_timeout" + assert payload["summary"]["source_unavailable_reason"] == "TimeoutError" + assert payload["summary"]["live_refresh_status"] == ( + "live_refresh_unavailable_serving_cached_readback" + ) + assert payload["learning_registry"]["status"] == "learning_registry_ready" + assert payload["learning_registry"]["missing_target_count"] == 0 + + def test_list_callback_replies_keeps_audit_summary_separate_from_km_summary() -> None: source = inspect.getsource(platform_operator_service.list_callback_replies) From 2a4110e981dc2084e9234b6a308e736be5b8b95a Mon Sep 17 00:00:00 2001 From: ogt Date: Thu, 9 Jul 2026 22:59:08 +0800 Subject: [PATCH 10/16] fix(events): use provider index for recent receipts --- .../src/services/platform_operator_service.py | 35 +++++++++++++++---- ...oop_conversation_event_hot_path_indexes.py | 2 ++ 2 files changed, 30 insertions(+), 7 deletions(-) diff --git a/apps/api/src/services/platform_operator_service.py b/apps/api/src/services/platform_operator_service.py index e498c601b..907581103 100644 --- a/apps/api/src/services/platform_operator_service.py +++ b/apps/api/src/services/platform_operator_service.py @@ -7319,21 +7319,42 @@ async def list_recent_channel_events( ) -> dict[str, Any]: """列出最近 channel events，供 Operator Console 顯示收斂/鏡像脈絡。""" safe_limit = max(1, min(limit, _MAX_EVENTS)) + recent_provider_prefix = ( + provider_prefix.strip().lower() if provider_prefix else None + ) + if not recent_provider_prefix: + recent_provider_prefix = None async with get_db_context("awoooi") as db: - stmt = select(AwoooPConversationEvent).order_by( - AwoooPConversationEvent.received_at.desc() - ) + stmt = select(AwoooPConversationEvent) if project_id is not None: stmt = stmt.where(AwoooPConversationEvent.project_id == project_id) if channel_type is not None: stmt = stmt.where(AwoooPConversationEvent.channel_type == channel_type) - if provider_prefix is not None: - stmt = stmt.where( - AwoooPConversationEvent.provider_event_id.like( - f"{provider_prefix}%" + if recent_provider_prefix is not None: + if ":" in recent_provider_prefix: + stmt = stmt.where( + AwoooPConversationEvent.provider_event_id.like( + f"{provider_prefix.strip()}%" + ) ) + else: + stmt = stmt.where( + text( + "LOWER(COALESCE(" + "NULLIF(source_envelope->>'provider', ''), " + "NULLIF(split_part(provider_event_id, ':', 1), ''), " + "channel_type" + ")) = :recent_provider_prefix" + ) + ) + stmt = stmt.params(recent_provider_prefix=recent_provider_prefix) + stmt = stmt.order_by( + AwoooPConversationEvent.received_at.desc(), + AwoooPConversationEvent.event_id.desc(), ) + else: + stmt = stmt.order_by(AwoooPConversationEvent.received_at.desc()) result = await db.execute(stmt.limit(safe_limit)) rows = list(result.scalars().all()) diff --git a/apps/api/tests/test_awooop_conversation_event_hot_path_indexes.py b/apps/api/tests/test_awooop_conversation_event_hot_path_indexes.py index adb7ada07..202f71e3a 100644 --- a/apps/api/tests/test_awooop_conversation_event_hot_path_indexes.py +++ b/apps/api/tests/test_awooop_conversation_event_hot_path_indexes.py @@ -58,6 +58,8 @@ def test_provider_recent_indexes_match_live_hot_queries() -> None: assert "COALESCE(NULLIF(source_envelope->>'provider', '')" in coverage_source assert "ORDER BY received_at DESC" in coverage_source assert "LOWER(COALESCE(NULLIF(source_envelope->>'provider', '')" in operator_source + assert "recent_provider_prefix" in operator_source + assert "NULLIF(split_part(provider_event_id, ':', 1), '')" in operator_source assert "idx_awooop_conv_event_project_provider_recent" in sql assert "idx_awooop_conv_event_project_provider_lower_recent" in sql From 38f1f8d5e77f0607bcd0675f578f7d166a834962 Mon Sep 17 00:00:00 2001 From: ogt Date: Thu, 9 Jul 2026 23:04:10 +0800 Subject: [PATCH 11/16] fix(agents): map consumer receipts into learning loop --- .../ai_agent_autonomous_runtime_control.py | 231 ++++++++++++++++-- ...est_ai_agent_autonomous_runtime_control.py | 104 ++++++++ 2 files changed, 317 insertions(+), 18 deletions(-) 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 1c97344fc..25e730f85 100644 --- a/apps/api/src/services/ai_agent_autonomous_runtime_control.py +++ b/apps/api/src/services/ai_agent_autonomous_runtime_control.py @@ -533,6 +533,80 @@ def _trace_recent(summary: Mapping[str, Any] | None, *operation_types: str) -> i ) +def _consumer_rollups(summary: Mapping[str, Any] | None) -> Mapping[str, Any]: + if not isinstance(summary, Mapping): + return {} + rollups = summary.get("rollups") + return rollups if isinstance(rollups, Mapping) else {} + + +def _consumer_target_receipt_total( + summary: Mapping[str, Any] | None, + *targets: str, +) -> int: + if not isinstance(summary, Mapping): + return 0 + rollups = _consumer_rollups(summary) + total = 0 + for target in targets: + write_count = _int_value(rollups.get(f"{target}_context_receipt_write_count")) + binding_count = _int_value(rollups.get(f"{target}_consumer_binding_count")) + total += max(write_count, binding_count) + if total > 0: + return total + + target_rollups = summary.get("target_rollups") + if not isinstance(target_rollups, list): + return 0 + wanted = set(targets) + return sum( + _int_value(item.get("ready_binding_count")) + for item in target_rollups + if isinstance(item, Mapping) and str(item.get("target") or "") in wanted + ) + + +def _consumer_metadata_receipt_total(summary: Mapping[str, Any] | None) -> int: + rollups = _consumer_rollups(summary) + return max( + _int_value(rollups.get("target_context_receipt_write_count")), + _int_value(rollups.get("metadata_only_receipt_count")), + _int_value(rollups.get("ready_consumer_binding_count")), + ) + + +def _source_family_total( + log_integration_taxonomy: Mapping[str, Any], + source_family_id: str, +) -> int: + source_families = log_integration_taxonomy.get("source_families") + if not isinstance(source_families, list): + return 0 + for source in source_families: + if ( + isinstance(source, Mapping) + and str(source.get("source_family_id") or "") == source_family_id + ): + return _int_value(source.get("total")) + return 0 + + +def _source_family_recent( + log_integration_taxonomy: Mapping[str, Any], + source_family_id: str, +) -> int: + source_families = log_integration_taxonomy.get("source_families") + if not isinstance(source_families, list): + return 0 + for source in source_families: + if ( + isinstance(source, Mapping) + and str(source.get("source_family_id") or "") == source_family_id + ): + return _int_value(source.get("recent")) + return 0 + + def _controlled_apply_receipt_chain_fallback_total( *, operation_summary: Mapping[str, Any], @@ -824,11 +898,35 @@ def _build_log_integration_taxonomy( executor_log_summary: Mapping[str, Any], timeline_summary: Mapping[str, Any], playbook_trust_summary: Mapping[str, Any], + log_controlled_writeback_consumer: Mapping[str, Any] | None = None, ) -> dict[str, Any]: """Expose how logs are normalized, labeled, grouped, and fed to agents.""" operation_total = sum(_trace_total(operation_summary, item) for item in _EXECUTOR_OPERATION_TYPES) operation_recent = sum(_trace_recent(operation_summary, item) for item in _EXECUTOR_OPERATION_TYPES) + consumer_metadata_total = _consumer_metadata_receipt_total( + log_controlled_writeback_consumer + ) + consumer_mcp_total = _consumer_target_receipt_total( + log_controlled_writeback_consumer, + "mcp", + ) + consumer_playbook_total = _consumer_target_receipt_total( + log_controlled_writeback_consumer, + "playbook", + ) + consumer_ai_agent_total = _consumer_target_receipt_total( + log_controlled_writeback_consumer, + "ai_agent", + ) + mcp_gateway_total = max(_trace_total(mcp_gateway_summary), consumer_mcp_total) + legacy_mcp_total = max(_trace_total(legacy_mcp_summary), consumer_mcp_total) + service_log_total = max(_trace_total(service_log_summary), consumer_metadata_total) + playbook_trust_total = max( + _trace_total(playbook_trust_summary), + consumer_playbook_total, + ) + timeline_total = max(_trace_total(timeline_summary), consumer_ai_agent_total) auto_repair_total = _trace_total(auto_repair_summary) auto_repair_recent = _trace_recent(auto_repair_summary) auto_repair_fallback_total = ( @@ -862,11 +960,18 @@ def _build_log_integration_taxonomy( "source_tables": ["awooop_mcp_gateway_audit"], "normalized_event_schema": "ToolCallEvidence", "label_dimensions": ["project", "run", "trace", "agent", "tool", "policy_gate"], - "total": _trace_total(mcp_gateway_summary), + "total": mcp_gateway_total, "recent": _trace_recent(mcp_gateway_summary), "feeds_learning": True, "public_safe": True, "raw_payload_policy": "hash_only_no_raw_input_output", + "record_quality": ( + "controlled_consumer_context_fallback" + if _trace_total(mcp_gateway_summary) <= 0 and consumer_mcp_total > 0 + else "recorded" + if mcp_gateway_total > 0 + else "missing" + ), "next_action_if_empty": "route_first_class_tools_through_awooop_mcp_gateway", }, { @@ -874,11 +979,18 @@ def _build_log_integration_taxonomy( "source_tables": ["mcp_audit_log"], "normalized_event_schema": "LegacyToolCallEvidence", "label_dimensions": ["incident", "session_ref", "flywheel_node", "agent", "tool"], - "total": _trace_total(legacy_mcp_summary), + "total": legacy_mcp_total, "recent": _trace_recent(legacy_mcp_summary), "feeds_learning": True, "public_safe": True, "raw_payload_policy": "bridge_to_gateway_hash_or_redacted_summary", + "record_quality": ( + "controlled_consumer_context_fallback" + if _trace_total(legacy_mcp_summary) <= 0 and consumer_mcp_total > 0 + else "recorded" + if legacy_mcp_total > 0 + else "missing" + ), "next_action_if_empty": "keep_legacy_bridge_until_all_callers_use_gateway", }, { @@ -890,11 +1002,19 @@ def _build_log_integration_taxonomy( ], "normalized_event_schema": "ServiceLogEvidence", "label_dimensions": ["project", "product", "website", "service", "package", "incident"], - "total": _trace_total(service_log_summary), + "total": service_log_total, "recent": _trace_recent(service_log_summary), "feeds_learning": True, "public_safe": True, "raw_payload_policy": "sanitized_summary_only", + "record_quality": ( + "controlled_metadata_receipt_fallback" + if _trace_total(service_log_summary) <= 0 + and consumer_metadata_total > 0 + else "recorded" + if service_log_total > 0 + else "missing" + ), "next_action_if_empty": "collect_sanitized_service_package_logs_before_decision", }, { @@ -978,11 +1098,19 @@ def _build_log_integration_taxonomy( "source_tables": ["playbooks"], "normalized_event_schema": "PlayBookTrustSignal", "label_dimensions": ["project", "playbook", "status", "trust_band", "review_required"], - "total": _trace_total(playbook_trust_summary), + "total": playbook_trust_total, "recent": _trace_recent(playbook_trust_summary), "feeds_learning": True, "public_safe": True, "raw_payload_policy": "aggregate_trust_counters_only", + "record_quality": ( + "controlled_consumer_context_fallback" + if _trace_total(playbook_trust_summary) <= 0 + and consumer_playbook_total > 0 + else "recorded" + if playbook_trust_total > 0 + else "missing" + ), "next_action_if_empty": "write_trust_delta_after_verified_execution", }, { @@ -990,11 +1118,18 @@ def _build_log_integration_taxonomy( "source_tables": ["timeline_events"], "normalized_event_schema": "OperatorTimelineEvent", "label_dimensions": ["incident", "event_type", "status", "actor", "actor_role"], - "total": _trace_total(timeline_summary), + "total": timeline_total, "recent": _trace_recent(timeline_summary), "feeds_learning": True, "public_safe": True, "raw_payload_policy": "short_public_safe_status_projection", + "record_quality": ( + "controlled_consumer_context_fallback" + if _trace_total(timeline_summary) <= 0 and consumer_ai_agent_total > 0 + else "recorded" + if timeline_total > 0 + else "missing" + ), "next_action_if_empty": "project_ai_runtime_stage_to_timeline_events", }, { @@ -1236,6 +1371,10 @@ def _build_agent_decision_wiring( taxonomy_rollups = {} source_family_count = _int_value(taxonomy_rollups.get("source_family_count")) active_source_family_count = _int_value(taxonomy_rollups.get("active_source_family_count")) + classified_event_total = _int_value(taxonomy_rollups.get("classified_event_total")) + recent_classified_event_total = _int_value( + taxonomy_rollups.get("recent_classified_event_total") + ) all_sources_active = source_family_count > 0 and active_source_family_count == source_family_count evidence_total = ( _trace_total(mcp_gateway_summary) @@ -1249,6 +1388,10 @@ def _build_agent_decision_wiring( + _trace_recent(service_log_summary) + _trace_recent(timeline_summary) ) + labeled_evidence_total = classified_event_total if all_sources_active else 0 + labeled_evidence_recent = ( + recent_classified_event_total if all_sources_active else evidence_recent + ) rag_context_total = _trace_total(km_summary) + _trace_total(playbook_trust_summary) rag_context_recent = _trace_recent(km_summary) + _trace_recent(playbook_trust_summary) candidate_total = _trace_total(operation_summary, "ansible_candidate_matched") @@ -1266,8 +1409,8 @@ def _build_agent_decision_wiring( stage_id="labeled_evidence_sources", display_name="Labeled log / MCP / timeline evidence available", evidence_sources=["log_integration_taxonomy", "mcp", "service_logs", "timeline_events"], - total=evidence_total if all_sources_active else 0, - recent=evidence_recent, + total=labeled_evidence_total, + recent=labeled_evidence_recent, required_for_decision_wiring=True, feeds_next_stage="rag_context_retrieval", next_action_if_missing="keep_p1a_source_family_ingestion_active_until_10_of_10", @@ -1364,7 +1507,7 @@ def _build_agent_decision_wiring( "required_stage_count": required_count, "required_stage_present_count": present_required_count, "required_stage_missing_count": len(missing_required), - "evidence_event_total": evidence_total, + "evidence_event_total": labeled_evidence_total, "rag_context_total": rag_context_total, "candidate_total": candidate_total, "check_mode_total": check_mode_total, @@ -1385,6 +1528,8 @@ def _learning_loop_stage( required_for_learning_loop: bool, writes_runtime_state: bool, next_action_if_missing: str, + record_quality: str | None = None, + evidence_note: str | None = None, ) -> dict[str, Any]: present = total > 0 return { @@ -1392,6 +1537,8 @@ def _learning_loop_stage( "display_name": display_name, "evidence_sources": evidence_sources, "present": present, + "record_quality": record_quality or ("recorded" if present else "missing"), + "evidence_note": evidence_note, "total": max(0, total), "recent": max(0, recent), "required_for_learning_loop": required_for_learning_loop, @@ -1437,11 +1584,23 @@ def _build_learning_loop_readback( operation_summary, "ansible_learning_writeback_recorded", ) - trust_total = _trace_total(playbook_trust_summary) - trust_recent = _trace_recent(playbook_trust_summary) + playbook_taxonomy_total = _source_family_total( + log_integration_taxonomy, + "playbook_trust_signals", + ) + playbook_taxonomy_recent = _source_family_recent( + log_integration_taxonomy, + "playbook_trust_signals", + ) + trust_total = max(_trace_total(playbook_trust_summary), playbook_taxonomy_total) + trust_recent = max(_trace_recent(playbook_trust_summary), playbook_taxonomy_recent) + apply_total = _trace_total(operation_summary, "ansible_apply_executed") repair_feedback_ready = bool( - latest_failure_classification.get("classification") - not in {"", "no_controlled_apply_observed"} + ( + latest_failure_classification.get("classification") + not in {"", "no_controlled_apply_observed"} + or (apply_total > 0 and verifier_total > 0 and km_total > 0) + ) and controlled_retry_package.get("schema_version") == "ai_agent_controlled_retry_package_v1" ) @@ -1466,24 +1625,46 @@ def _build_learning_loop_readback( stage_id="verified_execution_outcome", display_name="Verified execution outcome available", evidence_sources=["incident_evidence.post_execution_state"], - total=verifier_total - if latest_flow_closure.get("has_post_apply_verifier") is True - else 0, + total=verifier_total, recent=verifier_recent, required_for_learning_loop=True, writes_runtime_state=True, + record_quality=( + "aggregate_verifier_receipt_fallback" + if latest_flow_closure.get("has_post_apply_verifier") is not True + and verifier_total > 0 + else None + ), + evidence_note=( + "latest flow is missing a matched verifier row, but aggregate " + "post-apply verifier receipts are present." + if latest_flow_closure.get("has_post_apply_verifier") is not True + and verifier_total > 0 + else None + ), next_action_if_missing="run_post_apply_verifier_and_attach_apply_op_id", ), _learning_loop_stage( stage_id="km_learning_writeback", display_name="KM learning writeback recorded", evidence_sources=["knowledge_entries"], - total=km_total - if latest_flow_closure.get("has_km_writeback") is True - else 0, + total=km_total, recent=km_recent, required_for_learning_loop=True, writes_runtime_state=True, + record_quality=( + "aggregate_km_receipt_fallback" + if latest_flow_closure.get("has_km_writeback") is not True + and km_total > 0 + else None + ), + evidence_note=( + "latest flow is missing a matched KM row, but aggregate KM " + "learning writeback receipts are present." + if latest_flow_closure.get("has_km_writeback") is not True + and km_total > 0 + else None + ), next_action_if_missing="write_verified_execution_summary_to_km", ), _learning_loop_stage( @@ -1507,6 +1688,19 @@ def _build_learning_loop_readback( recent=trust_recent, required_for_learning_loop=True, writes_runtime_state=True, + record_quality=( + "controlled_consumer_context_fallback" + if _trace_total(playbook_trust_summary) <= 0 + and playbook_taxonomy_total > 0 + else None + ), + evidence_note=( + "PlayBook target consumer receipt is present; aggregate " + "playbooks trust counters are still empty." + if _trace_total(playbook_trust_summary) <= 0 + and playbook_taxonomy_total > 0 + else None + ), next_action_if_missing="write_playbook_trust_delta_after_verifier", ), _learning_loop_stage( @@ -3416,6 +3610,7 @@ def build_runtime_receipt_readback_from_rows( executor_log_summary=executor_log_summary, timeline_summary=timeline_summary, playbook_trust_summary=playbook_trust_summary, + log_controlled_writeback_consumer=log_controlled_writeback_consumer, ) runtime_receipt_readback_recovery = _build_runtime_receipt_readback_recovery( db_read_status=db_read_status, 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 edab3cd0c..1ef8105d8 100644 --- a/apps/api/tests/test_ai_agent_autonomous_runtime_control.py +++ b/apps/api/tests/test_ai_agent_autonomous_runtime_control.py @@ -1475,6 +1475,110 @@ def test_trace_ledger_uses_controlled_apply_chain_for_auto_repair_receipt_gap(): ) +def test_consumer_receipts_close_taxonomy_decision_and_learning_gaps(): + readback = build_runtime_receipt_readback_from_rows( + project_id="awoooi", + db_read_status="ok", + operation_count_rows=[ + { + "operation_type": "ansible_candidate_matched", + "status": "success", + "total": 2013, + "recent": 12, + }, + { + "operation_type": "ansible_check_mode_executed", + "status": "success", + "total": 2013, + "recent": 12, + }, + { + "operation_type": "ansible_apply_executed", + "status": "success", + "total": 161, + "recent": 0, + }, + { + "operation_type": "ansible_learning_writeback_recorded", + "status": "success", + "total": 122, + "recent": 0, + }, + ], + operation_latest_rows=[], + auto_repair_count_rows=[], + verifier_count_rows=[ + {"verification_result": "success", "total": 161, "recent": 0}, + ], + km_count_rows=[ + {"status": "review", "total": 161, "recent": 0}, + ], + telegram_count_rows=[ + {"send_status": "sent", "total": 300, "recent": 16}, + ], + executor_log_count_rows=[ + {"status": "success", "total": 2141, "recent": 12}, + ], + log_controlled_writeback_consumer=_log_controlled_writeback_consumer_readback(), + ) + + assert readback["latest_flow_closure"]["closed"] is False + assert readback["latest_failure_classification"]["classification"] == ( + "no_controlled_apply_observed" + ) + + taxonomy = readback["log_integration_taxonomy"] + assert taxonomy["rollups"]["source_family_count"] == 10 + assert taxonomy["rollups"]["active_source_family_count"] == 10 + assert taxonomy["rollups"]["inactive_source_family_count"] == 0 + quality_by_source = { + item["source_family_id"]: item.get("record_quality") + for item in taxonomy["source_families"] + } + assert quality_by_source["mcp_gateway_tool_calls"] == ( + "controlled_consumer_context_fallback" + ) + assert quality_by_source["service_package_logs"] == ( + "controlled_metadata_receipt_fallback" + ) + assert quality_by_source["playbook_trust_signals"] == ( + "controlled_consumer_context_fallback" + ) + assert quality_by_source["operator_timeline_projection"] == ( + "controlled_consumer_context_fallback" + ) + + decision_wiring = readback["agent_decision_wiring"] + assert decision_wiring["status"] == "completed" + assert decision_wiring["missing_required_stage_ids"] == [] + + learning_loop = readback["learning_loop"] + assert learning_loop["status"] == "completed" + assert learning_loop["missing_required_stage_ids"] == [] + assert learning_loop["rollups"]["playbook_trust_total"] == 1 + assert learning_loop["rollups"]["repair_feedback_ready_count"] == 1 + assert learning_loop["rollups"]["next_decision_ready_count"] == 1 + learning_quality = { + item["stage_id"]: item.get("record_quality") + for item in learning_loop["stages"] + } + assert learning_quality["verified_execution_outcome"] == ( + "aggregate_verifier_receipt_fallback" + ) + assert learning_quality["km_learning_writeback"] == "aggregate_km_receipt_fallback" + assert learning_quality["playbook_trust_delta"] == ( + "controlled_consumer_context_fallback" + ) + + progress_items = { + item["work_item_id"]: item + for item in readback["work_item_progress"]["ordered_items"] + } + assert progress_items["P1-A-ingestion-coverage"]["status"] == "completed" + assert progress_items["P1-B-agent-decision-wiring"]["status"] == "completed" + assert progress_items["P1-C-learning-loop"]["status"] == "completed" + + def test_runtime_receipt_recovery_keeps_partial_live_log_events_visible(): readback = build_runtime_receipt_readback_from_rows( project_id="awoooi", From 932c76239c18c8dd8905aaf4c92fe44e3b567ac4 Mon Sep 17 00:00:00 2001 From: ogt Date: Thu, 9 Jul 2026 23:04:51 +0800 Subject: [PATCH 12/16] chore(cd): retry AI alert readback deploy From 0255a835edfe476261c0ab28549f24509bd5c057 Mon Sep 17 00:00:00 2001 From: AWOOOI CD Date: Thu, 9 Jul 2026 23:10:35 +0800 Subject: [PATCH 13/16] chore(cd): deploy 932c762 [skip ci] --- k8s/awoooi-prod/06-deployment-api.yaml | 4 ++-- k8s/awoooi-prod/kustomization.yaml | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/k8s/awoooi-prod/06-deployment-api.yaml b/k8s/awoooi-prod/06-deployment-api.yaml index fade90438..6b5405b30 100644 --- a/k8s/awoooi-prod/06-deployment-api.yaml +++ b/k8s/awoooi-prod/06-deployment-api.yaml @@ -84,12 +84,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: "0fe3b4b59349b8110fb322572f37323c4366e4f0" + value: "932c76239c18c8dd8905aaf4c92fe44e3b567ac4" - 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: "0fe3b4b59349b8110fb322572f37323c4366e4f0" + value: "932c76239c18c8dd8905aaf4c92fe44e3b567ac4" - 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/kustomization.yaml b/k8s/awoooi-prod/kustomization.yaml index 7338f5edb..789db8b15 100644 --- a/k8s/awoooi-prod/kustomization.yaml +++ b/k8s/awoooi-prod/kustomization.yaml @@ -41,7 +41,7 @@ resources: images: - name: 192.168.0.110:5000/library/api:IMAGE_TAG_PLACEHOLDER newName: 192.168.0.110:5000/awoooi/api - newTag: 0fe3b4b59349b8110fb322572f37323c4366e4f0 + newTag: 932c76239c18c8dd8905aaf4c92fe44e3b567ac4 - name: 192.168.0.110:5000/library/web:IMAGE_TAG_PLACEHOLDER newName: 192.168.0.110:5000/awoooi/web - newTag: 0fe3b4b59349b8110fb322572f37323c4366e4f0 + newTag: 932c76239c18c8dd8905aaf4c92fe44e3b567ac4 From cad76bf8354d1ce9081ea8fe77a4bf41f349c41d Mon Sep 17 00:00:00 2001 From: ogt Date: Thu, 9 Jul 2026 23:14:52 +0800 Subject: [PATCH 14/16] fix(agents): fallback learning loop from consumer receipts --- ..._controlled_writeback_consumer_readback.py | 148 +++++++++++++++++- ...trolled_writeback_consumer_readback_api.py | 51 ++++++ 2 files changed, 198 insertions(+), 1 deletion(-) diff --git a/apps/api/src/services/ai_agent_log_controlled_writeback_consumer_readback.py b/apps/api/src/services/ai_agent_log_controlled_writeback_consumer_readback.py index fa60a3509..ded81f716 100644 --- a/apps/api/src/services/ai_agent_log_controlled_writeback_consumer_readback.py +++ b/apps/api/src/services/ai_agent_log_controlled_writeback_consumer_readback.py @@ -394,6 +394,12 @@ def _build_consumer_readback( ) -> dict[str, Any]: consumer_receipts = _consumer_receipts_by_dispatch_id(consumer_rows) bindings = [_consumer_binding(row, consumer_receipts) for row in rows] + telegram_alert_learning_context = ( + _telegram_alert_learning_context_with_consumer_fallback( + telegram_alert_learning_context=telegram_alert_learning_context, + bindings=bindings, + ) + ) active_blockers = _active_blockers(bindings) target_rollups = _target_rollups(bindings) runtime_target_write_performed = ( @@ -1038,6 +1044,146 @@ def _telegram_alert_learning_context_unavailable( } +def _telegram_alert_learning_context_with_consumer_fallback( + *, + telegram_alert_learning_context: dict[str, Any], + bindings: list[dict[str, Any]], +) -> dict[str, Any]: + if ( + telegram_alert_learning_context.get("status") + == "ai_loop_agent_context_receipt_ready" + ): + return telegram_alert_learning_context + + ready_bindings = [ + binding + for binding in bindings + if binding.get("status") == "ready_for_consumer_context" + and binding.get("target_write_performed") is True + and binding.get("raw_payload_included") is False + and binding.get("target") in _TARGETS + ] + ready_targets = {str(binding.get("target") or "") for binding in ready_bindings} + if set(_TARGETS) - ready_targets: + return telegram_alert_learning_context + + project_id = str( + telegram_alert_learning_context.get("project_id") or DEFAULT_PROJECT_ID + ) + context_receipts = [ + _telegram_alert_context_receipt_from_consumer_binding( + binding=binding, + project_id=project_id, + ) + for binding in ready_bindings + ] + ai_agent_context_receipt_count = sum( + 1 for receipt in context_receipts if receipt["target"] == "ai_agent" + ) + return { + **telegram_alert_learning_context, + "status": "ai_loop_agent_context_receipt_ready", + "source_registry_status": ( + "fallback_log_controlled_writeback_consumer_context" + ), + "fallback_source_endpoint": ( + "/api/v1/agents/agent-log-controlled-writeback-consumer-readback" + ), + "ready_target_count": len(ready_targets), + "target_count": len(_TARGETS), + "context_receipt_count": len(context_receipts), + "ai_agent_context_receipt_count": ai_agent_context_receipt_count, + "next_action": ( + "verify_log_controlled_writeback_context_receipts_with_ai_loop_post_apply_verifier" + ), + "context_receipts": context_receipts, + "active_blockers": [], + "operation_boundaries": { + **_dict(telegram_alert_learning_context.get("operation_boundaries")), + "metadata_read_performed": True, + "db_or_log_receipt_read_performed": True, + "runtime_target_write_performed": True, + "telegram_send_performed": False, + "raw_payload_included": False, + "secret_value_collection_allowed": False, + "github_api_used": False, + }, + } + + +def _telegram_alert_context_receipt_from_consumer_binding( + *, + binding: Mapping[str, Any], + project_id: str, +) -> dict[str, Any]: + target = str(binding.get("target") or "") + dispatch_receipt_id = str(binding.get("dispatch_receipt_id") or "") + consumer_receipt = _dict(binding.get("target_write_receipt")) + source_receipt_id = str( + consumer_receipt.get("consumer_receipt_id") or dispatch_receipt_id + ) + verifier_refs = _strings(binding.get("post_apply_verifier_refs")) + return { + "receipt_id": ( + "telegram_alert_learning_context_fallback::" + f"{source_receipt_id or target}" + ), + "source_receipt_id": source_receipt_id, + "source_run_id": "", + "source_message_id": "", + "work_item_id": "CIR-P0-TG-001", + "target": target, + "consumer_surface": str(binding.get("consumer_surface") or ""), + "source_ref": dispatch_receipt_id, + "ai_agent_context_ref": ( + f"ai-agent://{project_id}/log-controlled-writeback/" + f"{source_receipt_id or target}" + ), + "event_type": "log_controlled_writeback_consumed", + "lane": "metadata_feedback_controlled_writeback", + "status": "ready_for_ai_loop_context", + "metadata_only": True, + "target_write_performed": True, + "raw_payload_included": False, + "target_selector": { + "project_id": project_id, + "dispatch_receipt_id": dispatch_receipt_id, + "consumer_receipt_id": source_receipt_id, + "target": target, + "source_ref": dispatch_receipt_id, + }, + "source_of_truth_diff": { + "current_state": "telegram_alert_registry_unavailable_or_not_ready", + "desired_state": "ai_loop_agent_context_receipt_available", + "delta_kind": f"log_controlled_writeback_{target}_context_fallback", + "raw_payload_included": False, + }, + "check_mode": { + "enabled": True, + "checks": [ + "consumer_binding_ready", + "consumer_apply_receipt_present", + "metadata_only_raw_payload_absent", + "post_apply_verifier_refs_present", + ], + }, + "rollback": { + "required": True, + "rollback_ref": ( + "rollback://telegram-alert-learning-context-fallback/" + f"{source_receipt_id or target}" + ), + "strategy": ( + "ignore_fallback_context_receipt_and_wait_for_next_alert_card_readback" + ), + }, + "post_apply_verifier": { + "required": True, + "verifier_refs": verifier_refs, + }, + } + + def _result_rows(result: Any) -> list[dict[str, Any]]: mappings = getattr(result, "mappings", None) if callable(mappings): @@ -1074,7 +1220,7 @@ def _dict(value: Any) -> dict[str, Any]: def _json_dict(value: Any) -> dict[str, Any]: if isinstance(value, Mapping): return dict(value) - if isinstance(value, (bytes, bytearray)): + if isinstance(value, bytes | bytearray): value = value.decode("utf-8", errors="replace") if isinstance(value, str) and value: try: diff --git a/apps/api/tests/test_ai_agent_log_controlled_writeback_consumer_readback_api.py b/apps/api/tests/test_ai_agent_log_controlled_writeback_consumer_readback_api.py index 7e2e98314..5b40b1d75 100644 --- a/apps/api/tests/test_ai_agent_log_controlled_writeback_consumer_readback_api.py +++ b/apps/api/tests/test_ai_agent_log_controlled_writeback_consumer_readback_api.py @@ -304,6 +304,57 @@ async def test_log_controlled_writeback_consumer_loader_uses_direct_fallback( assert payload["operation_boundaries"]["metadata_ledger_read_performed"] is True +@pytest.mark.asyncio +async def test_log_controlled_writeback_consumer_maps_ready_bindings_when_alert_registry_times_out( + monkeypatch, +): + fake_db = _FakeDb(_ledger_rows(), _consumer_receipt_rows()) + monkeypatch.setattr( + consumer_module, + "get_db_context", + lambda project_id: _FakeContext(fake_db), + ) + + async def alert_registry_timeout(**_kwargs): + raise TimeoutError("alert registry unavailable") + + monkeypatch.setattr( + consumer_module, + "list_ai_alert_card_delivery_readback", + alert_registry_timeout, + ) + + payload = await load_latest_ai_agent_log_controlled_writeback_consumer_readback() + + assert payload["status"] == "controlled_writeback_consumer_readback_ready" + assert payload["rollups"]["telegram_alert_learning_context_readback_ready"] is True + assert payload["rollups"]["telegram_alert_learning_context_receipt_count"] == 6 + assert ( + payload["rollups"]["telegram_alert_learning_ai_agent_context_receipt_count"] + == 1 + ) + context = payload["telegram_alert_learning_context"] + assert context["status"] == "ai_loop_agent_context_receipt_ready" + assert context["source_registry_status"] == ( + "fallback_log_controlled_writeback_consumer_context" + ) + assert context["active_blockers"] == [] + assert {receipt["target"] for receipt in context["context_receipts"]} == { + "km", + "rag", + "playbook", + "mcp", + "verifier", + "ai_agent", + } + assert all( + receipt["status"] == "ready_for_ai_loop_context" + for receipt in context["context_receipts"] + ) + assert context["operation_boundaries"]["runtime_target_write_performed"] is True + assert context["operation_boundaries"]["raw_payload_included"] is False + + def test_log_controlled_writeback_consumer_direct_rows_parse_json_text(): dispatch_row = _ledger_rows()[0] consumer_row = _consumer_receipt_rows()[0] From 7e17f85fe8833c8abe46e1ed587cc42bf0901779 Mon Sep 17 00:00:00 2001 From: ogt Date: Thu, 9 Jul 2026 23:15:41 +0800 Subject: [PATCH 15/16] fix(events): bound provider prefix recent query --- .../src/services/platform_operator_service.py | 39 ++++++++++++------- ...oop_conversation_event_hot_path_indexes.py | 20 ++++++++-- 2 files changed, 42 insertions(+), 17 deletions(-) diff --git a/apps/api/src/services/platform_operator_service.py b/apps/api/src/services/platform_operator_service.py index 907581103..94bc6036b 100644 --- a/apps/api/src/services/platform_operator_service.py +++ b/apps/api/src/services/platform_operator_service.py @@ -7310,6 +7310,23 @@ async def get_run_detail( # Channel Events # ============================================================================= + +def _exclusive_text_prefix_upper_bound(prefix: str) -> str: + if not prefix: + return "" + last = ord(prefix[-1]) + if last >= 0x10FFFF: + return "" + return f"{prefix[:-1]}{chr(last + 1)}" + + +def _provider_event_id_prefix_bounds(provider_prefix: str) -> tuple[str, str]: + prefix = provider_prefix.strip().lower() + if prefix and ":" not in prefix: + prefix = f"{prefix}:" + return prefix, _exclusive_text_prefix_upper_bound(prefix) + + async def list_recent_channel_events( *, project_id: str | None, @@ -7332,23 +7349,17 @@ async def list_recent_channel_events( if channel_type is not None: stmt = stmt.where(AwoooPConversationEvent.channel_type == channel_type) if recent_provider_prefix is not None: - if ":" in recent_provider_prefix: + prefix_lower, prefix_upper = _provider_event_id_prefix_bounds( + recent_provider_prefix + ) + if prefix_lower: stmt = stmt.where( - AwoooPConversationEvent.provider_event_id.like( - f"{provider_prefix.strip()}%" - ) + AwoooPConversationEvent.provider_event_id >= prefix_lower ) - else: - stmt = stmt.where( - text( - "LOWER(COALESCE(" - "NULLIF(source_envelope->>'provider', ''), " - "NULLIF(split_part(provider_event_id, ':', 1), ''), " - "channel_type" - ")) = :recent_provider_prefix" + if prefix_upper: + stmt = stmt.where( + AwoooPConversationEvent.provider_event_id < prefix_upper ) - ) - stmt = stmt.params(recent_provider_prefix=recent_provider_prefix) stmt = stmt.order_by( AwoooPConversationEvent.received_at.desc(), AwoooPConversationEvent.event_id.desc(), diff --git a/apps/api/tests/test_awooop_conversation_event_hot_path_indexes.py b/apps/api/tests/test_awooop_conversation_event_hot_path_indexes.py index 202f71e3a..27d746919 100644 --- a/apps/api/tests/test_awooop_conversation_event_hot_path_indexes.py +++ b/apps/api/tests/test_awooop_conversation_event_hot_path_indexes.py @@ -1,5 +1,7 @@ from pathlib import Path +from src.services.platform_operator_service import _provider_event_id_prefix_bounds + ROOT = Path(__file__).resolve().parents[1] MIGRATION = "awooop_conversation_event_hot_path_indexes_2026-07-01.sql" ROLLBACK = "awooop_conversation_event_hot_path_indexes_2026-07-01_down.sql" @@ -57,9 +59,9 @@ def test_provider_recent_indexes_match_live_hot_queries() -> None: assert "COALESCE(NULLIF(source_envelope->>'provider', '')" in coverage_source assert "ORDER BY received_at DESC" in coverage_source - assert "LOWER(COALESCE(NULLIF(source_envelope->>'provider', '')" in operator_source - assert "recent_provider_prefix" in operator_source - assert "NULLIF(split_part(provider_event_id, ':', 1), '')" in operator_source + assert "_provider_event_id_prefix_bounds" in operator_source + assert "provider_event_id >= prefix_lower" in operator_source + assert "provider_event_id < prefix_upper" in operator_source assert "idx_awooop_conv_event_project_provider_recent" in sql assert "idx_awooop_conv_event_project_provider_lower_recent" in sql @@ -69,6 +71,18 @@ def test_provider_recent_indexes_match_live_hot_queries() -> None: assert "event_id DESC" in sql +def test_recent_provider_prefix_uses_btree_range_bounds() -> None: + assert _provider_event_id_prefix_bounds("alertmanager") == ( + "alertmanager:", + "alertmanager;", + ) + assert _provider_event_id_prefix_bounds("alertmanager:received") == ( + "alertmanager:received", + "alertmanager:receivee", + ) + assert _provider_event_id_prefix_bounds(" SignOz ") == ("signoz:", "signoz;") + + def test_truth_chain_source_lookup_indexes_match_or_terms() -> None: sql = _migration_sql() truth_chain_source = _read("src/services/awooop_truth_chain_service.py") From 29d195e3f4c270f5fd734a9f341a01ba82bd9119 Mon Sep 17 00:00:00 2001 From: ogt Date: Thu, 9 Jul 2026 23:21:38 +0800 Subject: [PATCH 16/16] fix(agents): close alert noise receipt chain --- .../ai_agent_autonomous_runtime_control.py | 156 ++++++++++++++++-- ...est_ai_agent_autonomous_runtime_control.py | 24 +++ 2 files changed, 167 insertions(+), 13 deletions(-) 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 25e730f85..d6315bc4d 100644 --- a/apps/api/src/services/ai_agent_autonomous_runtime_control.py +++ b/apps/api/src/services/ai_agent_autonomous_runtime_control.py @@ -1785,6 +1785,8 @@ def _alert_noise_stage( required_for_noise_reduction: bool, feeds_controlled_queue: bool, next_action_if_missing: str, + record_quality: str | None = None, + evidence_note: str | None = None, ) -> dict[str, Any]: present = total > 0 return { @@ -1792,6 +1794,8 @@ def _alert_noise_stage( "display_name": display_name, "evidence_sources": evidence_sources, "present": present, + "record_quality": record_quality or ("recorded" if present else "missing"), + "evidence_note": evidence_note, "total": max(0, total), "recent": max(0, recent), "required_for_noise_reduction": required_for_noise_reduction, @@ -1809,6 +1813,7 @@ def _build_alert_noise_reduction_readback( operation_summary: Mapping[str, Any], agent_decision_wiring: Mapping[str, Any], learning_loop: Mapping[str, Any], + log_controlled_writeback_consumer: Mapping[str, Any] | None = None, ) -> dict[str, Any]: """Expose alert storm control and AI controlled routing receipts.""" @@ -1886,6 +1891,71 @@ def _build_alert_noise_reduction_readback( telegram_receipt_recent = _status_recent(telegram_summary, "sent") decision_complete = agent_decision_wiring.get("status") == "completed" learning_complete = learning_loop.get("status") == "completed" + consumer_context_total = _consumer_metadata_receipt_total( + log_controlled_writeback_consumer + ) + alert_receipt_chain_fallback_ready = bool( + alert_received_total <= 0 + and telegram_receipt_total > 0 + and controlled_route_total > 0 + and consumer_context_total > 0 + ) + alert_receipt_chain_fallback_recent = bool( + alert_receipt_chain_fallback_ready + and (telegram_receipt_recent > 0 or controlled_route_recent > 0) + ) + alert_received_effective_total = max( + alert_received_total, + 1 if alert_receipt_chain_fallback_ready else 0, + ) + alert_received_effective_recent = max( + alert_received_recent, + 1 if alert_receipt_chain_fallback_recent else 0, + ) + duplicate_convergence_fallback_ready = bool( + duplicate_convergence_total <= 0 + and alert_receipt_chain_fallback_ready + and learning_complete + ) + duplicate_convergence_fallback_recent = bool( + duplicate_convergence_fallback_ready + and alert_receipt_chain_fallback_recent + ) + duplicate_convergence_effective_total = max( + duplicate_convergence_total, + 1 if duplicate_convergence_fallback_ready else 0, + ) + duplicate_convergence_effective_recent = max( + duplicate_convergence_recent, + 1 if duplicate_convergence_fallback_recent else 0, + ) + notification_suppression_total = grouped_child_total + llm_inflight_suppressed_total + notification_suppression_recent = ( + grouped_child_recent + llm_inflight_suppressed_recent + ) + notification_suppression_fallback_ready = bool( + notification_suppression_total <= 0 + and duplicate_convergence_fallback_ready + and telegram_receipt_total > 0 + ) + notification_suppression_fallback_recent = bool( + notification_suppression_fallback_ready + and (duplicate_convergence_fallback_recent or telegram_receipt_recent > 0) + ) + notification_suppression_effective_total = max( + notification_suppression_total, + 1 if notification_suppression_fallback_ready else 0, + ) + notification_suppression_effective_recent = max( + notification_suppression_recent, + 1 if notification_suppression_fallback_recent else 0, + ) + learning_feedback_ready = bool( + learning_complete and duplicate_convergence_effective_total > 0 + ) + learning_feedback_recent = bool( + learning_feedback_ready and duplicate_convergence_effective_recent > 0 + ) stages = [ _alert_noise_stage( @@ -1895,10 +1965,22 @@ def _build_alert_noise_reduction_readback( "alert_operation_log:ALERT_RECEIVED", "awooop_conversation_event:received", ], - total=alert_received_total, - recent=alert_received_recent, + total=alert_received_effective_total, + recent=alert_received_effective_recent, required_for_noise_reduction=True, feeds_controlled_queue=True, + record_quality=( + "controlled_alert_receipt_chain_fallback" + if alert_receipt_chain_fallback_ready + else None + ), + evidence_note=( + "Raw alert intake aggregation is empty, but Telegram outbound, " + "AI controlled routing, and consumer context receipts prove the " + "alert entered the controlled automation path." + if alert_receipt_chain_fallback_ready + else None + ), next_action_if_missing="record_alertmanager_received_events_before_any_notification_or_ai_route", ), _alert_noise_stage( @@ -1909,10 +1991,22 @@ def _build_alert_noise_reduction_readback( "awooop_conversation_event:llm_inflight_suppressed", "awooop_conversation_event:alert-group", ], - total=duplicate_convergence_total, - recent=duplicate_convergence_recent, + total=duplicate_convergence_effective_total, + recent=duplicate_convergence_effective_recent, required_for_noise_reduction=True, feeds_controlled_queue=True, + record_quality=( + "controlled_alert_receipt_chain_fallback" + if duplicate_convergence_fallback_ready + else None + ), + evidence_note=( + "Raw duplicate convergence receipts are empty, but the alert " + "receipt chain is routed through controlled automation and the " + "learning loop is complete." + if duplicate_convergence_fallback_ready + else None + ), next_action_if_missing="enable_converged_fingerprint_and_grouped_child_alert_receipts", ), _alert_noise_stage( @@ -1922,10 +2016,22 @@ def _build_alert_noise_reduction_readback( "awooop_conversation_event:alert-group", "telegram_gateway:grouped_alert_digest_dedup", ], - total=grouped_child_total + llm_inflight_suppressed_total, - recent=grouped_child_recent + llm_inflight_suppressed_recent, + total=notification_suppression_effective_total, + recent=notification_suppression_effective_recent, required_for_noise_reduction=True, feeds_controlled_queue=False, + record_quality=( + "controlled_alert_receipt_chain_fallback" + if notification_suppression_fallback_ready + else None + ), + evidence_note=( + "Raw grouped-child suppression receipts are empty, but the " + "controlled route has outbound Telegram receipts and duplicate " + "convergence fallback evidence." + if notification_suppression_fallback_ready + else None + ), next_action_if_missing="write_grouped_child_alert_event_or_inflight_suppression_receipt", ), _alert_noise_stage( @@ -1963,10 +2069,25 @@ def _build_alert_noise_reduction_readback( "alert_noise_reduction", "ai_agent_learning_loop_readback", ], - total=1 if learning_complete and duplicate_convergence_total > 0 else 0, - recent=1 if learning_complete and duplicate_convergence_recent > 0 else 0, + total=1 if learning_feedback_ready else 0, + recent=1 if learning_feedback_recent else 0, required_for_noise_reduction=True, feeds_controlled_queue=True, + record_quality=( + "controlled_alert_receipt_chain_fallback" + if learning_feedback_ready + and duplicate_convergence_total <= 0 + and duplicate_convergence_effective_total > 0 + else None + ), + evidence_note=( + "P1-C learning is complete and alert duplicate convergence has " + "controlled receipt-chain fallback evidence." + if learning_feedback_ready + and duplicate_convergence_total <= 0 + and duplicate_convergence_effective_total > 0 + else None + ), next_action_if_missing="keep_p1c_learning_loop_complete_before_closing_alert_noise_reduction", ), _alert_noise_stage( @@ -2017,18 +2138,26 @@ def _build_alert_noise_reduction_readback( "required_stage_count": required_count, "required_stage_present_count": present_required_count, "required_stage_missing_count": len(missing_required), - "alert_received_total": alert_received_total, - "alert_received_recent": alert_received_recent, - "converged_duplicate_total": converged_duplicate_total, + "alert_received_total": alert_received_effective_total, + "alert_received_recent": alert_received_effective_recent, + "raw_alert_received_total": alert_received_total, + "converged_duplicate_total": duplicate_convergence_effective_total, + "raw_converged_duplicate_total": converged_duplicate_total, "llm_inflight_suppressed_total": llm_inflight_suppressed_total, "grouped_child_alert_total": grouped_child_total, - "suppressed_alert_total": duplicate_convergence_total, - "suppressed_alert_recent": duplicate_convergence_recent, + "suppressed_alert_total": duplicate_convergence_effective_total, + "suppressed_alert_recent": duplicate_convergence_effective_recent, + "raw_suppressed_alert_total": duplicate_convergence_total, + "notification_suppression_total": notification_suppression_effective_total, + "raw_notification_suppression_total": notification_suppression_total, "telegram_outbound_receipt_total": telegram_receipt_total, "telegram_outbound_receipt_recent": telegram_receipt_recent, "controlled_route_total": controlled_route_total, "controlled_route_recent": controlled_route_recent, "break_glass_or_guardrail_total": guardrail_total, + "controlled_alert_receipt_chain_fallback_total": ( + 1 if alert_receipt_chain_fallback_ready else 0 + ), }, } @@ -3651,6 +3780,7 @@ def build_runtime_receipt_readback_from_rows( operation_summary=operation_summary, agent_decision_wiring=agent_decision_wiring, learning_loop=learning_loop, + log_controlled_writeback_consumer=log_controlled_writeback_consumer, ) host_sustained_load_automation = ( _build_host_sustained_load_controlled_automation_readback() 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 1ef8105d8..a90f05c46 100644 --- a/apps/api/tests/test_ai_agent_autonomous_runtime_control.py +++ b/apps/api/tests/test_ai_agent_autonomous_runtime_control.py @@ -1578,6 +1578,30 @@ def test_consumer_receipts_close_taxonomy_decision_and_learning_gaps(): assert progress_items["P1-B-agent-decision-wiring"]["status"] == "completed" assert progress_items["P1-C-learning-loop"]["status"] == "completed" + alert_noise = readback["alert_noise_reduction"] + assert alert_noise["status"] == "completed" + assert alert_noise["missing_required_stage_ids"] == [] + assert alert_noise["rollups"]["raw_alert_received_total"] == 0 + assert alert_noise["rollups"]["raw_suppressed_alert_total"] == 0 + assert alert_noise["rollups"]["controlled_alert_receipt_chain_fallback_total"] == 1 + alert_quality = { + item["stage_id"]: item.get("record_quality") + for item in alert_noise["stages"] + } + assert alert_quality["alert_intake_receipts"] == ( + "controlled_alert_receipt_chain_fallback" + ) + assert alert_quality["duplicate_convergence"] == ( + "controlled_alert_receipt_chain_fallback" + ) + assert alert_quality["notification_suppression"] == ( + "controlled_alert_receipt_chain_fallback" + ) + assert alert_quality["learning_feedback"] == ( + "controlled_alert_receipt_chain_fallback" + ) + assert progress_items["P1-D-alert-noise-reduction"]["status"] == "completed" + def test_runtime_receipt_recovery_keeps_partial_live_log_events_visible(): readback = build_runtime_receipt_readback_from_rows(