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 040dd5ff2..501958458 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 @@ -11,8 +11,11 @@ prove this route has rolled out. from __future__ import annotations import asyncio +import copy import json +import time from collections.abc import Mapping +from datetime import UTC, datetime from typing import Any from sqlalchemy import text @@ -51,6 +54,13 @@ _CONSUMER_DIRECT_CONNECT_TIMEOUT_SECONDS = 1.5 _CONSUMER_DIRECT_QUERY_TIMEOUT_SECONDS = 2.0 _CONSUMER_TELEGRAM_CONTEXT_TIMEOUT_SECONDS = 1.5 _CONSUMER_STATEMENT_TIMEOUT_MS = 1500 +_CONSUMER_READBACK_CACHE_TTL_SECONDS = 20.0 +_consumer_readback_cache: dict[ + tuple[int, int, str], tuple[float, dict[str, Any]] +] = {} +_consumer_readback_inflight: dict[ + tuple[int, int, str], asyncio.Task[dict[str, Any]] +] = {} logger = get_logger(__name__) @@ -58,7 +68,48 @@ async def load_latest_ai_agent_log_controlled_writeback_consumer_readback( *, project_id: str = DEFAULT_PROJECT_ID, ) -> dict[str, Any]: - """Return live consumer bindings for LOG controlled writeback receipts.""" + """Return live consumer bindings without duplicating the same DB read burst.""" + + loop = asyncio.get_running_loop() + cache_key = (id(loop), id(get_db_context), project_id) + cached = _consumer_readback_cache_get(cache_key=cache_key) + if cached is not None: + return cached + + task = _consumer_readback_inflight.get(cache_key) + coalesced = task is not None + if task is None: + task = loop.create_task( + _load_latest_ai_agent_log_controlled_writeback_consumer_readback_uncached( + project_id=project_id + ) + ) + _consumer_readback_inflight[cache_key] = task + + try: + payload = await asyncio.shield(task) + payload = _with_consumer_readback_freshness( + payload, + cache_hit=False, + cache_age_seconds=0.0, + inflight_coalesced=coalesced, + ) + if _consumer_readback_cacheable(payload): + _consumer_readback_cache[cache_key] = ( + time.monotonic(), + copy.deepcopy(payload), + ) + return payload + finally: + if task.done() and _consumer_readback_inflight.get(cache_key) is task: + _consumer_readback_inflight.pop(cache_key, None) + + +async def _load_latest_ai_agent_log_controlled_writeback_consumer_readback_uncached( + *, + project_id: str, +) -> dict[str, Any]: + """Read current receipts once; the public loader owns cache/coalescing.""" try: rows, consumer_rows = await _load_consumer_rows_with_retry(project_id=project_id) @@ -94,6 +145,68 @@ async def load_latest_ai_agent_log_controlled_writeback_consumer_readback( ) +def _consumer_readback_cache_get( + *, + cache_key: tuple[int, int, str], +) -> dict[str, Any] | None: + now = time.monotonic() + for key, (stored_at, _payload) in list(_consumer_readback_cache.items()): + if now - stored_at > _CONSUMER_READBACK_CACHE_TTL_SECONDS: + _consumer_readback_cache.pop(key, None) + + cached = _consumer_readback_cache.get(cache_key) + if cached is None: + return None + stored_at, payload = cached + return _with_consumer_readback_freshness( + payload, + cache_hit=True, + cache_age_seconds=max(0.0, now - stored_at), + inflight_coalesced=False, + ) + + +def _consumer_readback_cacheable(payload: Mapping[str, Any]) -> bool: + boundaries = payload.get("operation_boundaries") + return bool( + isinstance(boundaries, Mapping) + and boundaries.get("metadata_ledger_read_performed") is True + and boundaries.get("consumer_apply_receipt_read_performed") is True + ) + + +def _with_consumer_readback_freshness( + payload: Mapping[str, Any], + *, + cache_hit: bool, + cache_age_seconds: float, + inflight_coalesced: bool, +) -> dict[str, Any]: + result = copy.deepcopy(dict(payload)) + prior = result.get("readback_freshness") + observed_at = ( + str(prior.get("observed_at") or "") + if isinstance(prior, Mapping) + else "" + ) + result["readback_freshness"] = { + "observed_at": observed_at or datetime.now(UTC).isoformat(), + "cache_ttl_seconds": _CONSUMER_READBACK_CACHE_TTL_SECONDS, + "cache_hit": cache_hit, + "cache_age_seconds": round(cache_age_seconds, 3), + "stale": False, + "inflight_coalesced": inflight_coalesced, + } + return result + + +def _reset_consumer_readback_cache_for_tests() -> None: + """Reset process-local readback state between focused unit tests.""" + + _consumer_readback_cache.clear() + _consumer_readback_inflight.clear() + + async def _load_consumer_rows_with_retry( *, project_id: str, diff --git a/apps/api/src/services/awooop_ansible_check_mode_service.py b/apps/api/src/services/awooop_ansible_check_mode_service.py index 903899a83..9edb2b681 100644 --- a/apps/api/src/services/awooop_ansible_check_mode_service.py +++ b/apps/api/src/services/awooop_ansible_check_mode_service.py @@ -7733,9 +7733,18 @@ async def claim_catalog_drift_failed_check_modes( { "project_id": project_id, "candidate_max_age_hours": max(1, max_age_hours), - "scan_limit": min(100, max(10, limit * 10)), + "scan_limit": 100, }, ) + ranked_rows: list[ + tuple[ + int, + dict[str, Any], + AnsibleCheckModeClaim, + str, + str, + ] + ] = [] for row_value in result.mappings().all(): row = dict(row_value) previous_input = _json_loads(row.get("input")) @@ -7744,6 +7753,9 @@ async def claim_catalog_drift_failed_check_modes( or previous_input.get("playbook_path") or "" ) + previous_catalog_id = str( + previous_input.get("catalog_id") or "" + ) previous_catalog_revision = str( previous_input.get("catalog_revision") or "" ) @@ -7768,6 +7780,33 @@ async def claim_catalog_drift_failed_check_modes( or not (path_changed or revision_changed) ): continue + catalog_route_changed = bool( + previous_catalog_id + and canonical_claim.catalog_id != previous_catalog_id + ) + replay_priority = ( + 0 if catalog_route_changed else 1 if path_changed else 2 + ) + ranked_rows.append( + ( + replay_priority, + row, + canonical_claim, + previous_check_path, + previous_catalog_revision, + ) + ) + + for ( + _, + row, + canonical_claim, + previous_check_path, + previous_catalog_revision, + ) in sorted(ranked_rows, key=lambda item: item[0]): + current_catalog_revision = str( + canonical_claim.input_payload.get("catalog_revision") or "" + ) catalog_replay_revision = ( current_catalog_revision or f"path:{canonical_claim.playbook_path}" 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 5b40b1d75..044c19d59 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 @@ -1,5 +1,6 @@ from __future__ import annotations +import asyncio import json import pytest @@ -63,6 +64,8 @@ class _FailingContext: @pytest.fixture(autouse=True) def _stub_telegram_alert_card_readback(monkeypatch): + consumer_module._reset_consumer_readback_cache_for_tests() + async def fake_alert_card_readback(**_kwargs): return _telegram_alert_card_readback() @@ -71,6 +74,8 @@ def _stub_telegram_alert_card_readback(monkeypatch): "list_ai_alert_card_delivery_readback", fake_alert_card_readback, ) + yield + consumer_module._reset_consumer_readback_cache_for_tests() def _ledger_rows() -> list[dict]: @@ -215,6 +220,69 @@ async def test_log_controlled_writeback_consumer_loader_reads_live_ledger(monkey ] +@pytest.mark.asyncio +async def test_log_controlled_writeback_consumer_loader_reuses_fresh_success( + monkeypatch, +): + fake_db = _FakeDb(_ledger_rows(), _consumer_receipt_rows()) + monkeypatch.setattr( + consumer_module, + "get_db_context", + lambda project_id: _FakeContext(fake_db), + ) + + first = await load_latest_ai_agent_log_controlled_writeback_consumer_readback() + second = await load_latest_ai_agent_log_controlled_writeback_consumer_readback() + + assert first["readback_freshness"]["cache_hit"] is False + assert second["readback_freshness"]["cache_hit"] is True + assert second["readback_freshness"]["stale"] is False + assert second["readback_freshness"]["cache_age_seconds"] >= 0 + assert fake_db.params == [ + {"operation_type": "log_controlled_writeback_dispatched"}, + {"operation_type": "log_controlled_writeback_consumed"}, + ] + + +@pytest.mark.asyncio +async def test_log_controlled_writeback_consumer_loader_coalesces_concurrent_reads( + monkeypatch, +): + started = asyncio.Event() + release = asyncio.Event() + call_count = 0 + + async def delayed_rows(*, project_id: str): + nonlocal call_count + assert project_id == "awoooi" + call_count += 1 + started.set() + await release.wait() + return _ledger_rows(), _consumer_receipt_rows() + + monkeypatch.setattr( + consumer_module, + "_load_consumer_rows_with_retry", + delayed_rows, + ) + + first_task = asyncio.create_task( + load_latest_ai_agent_log_controlled_writeback_consumer_readback() + ) + await started.wait() + second_task = asyncio.create_task( + load_latest_ai_agent_log_controlled_writeback_consumer_readback() + ) + await asyncio.sleep(0) + release.set() + first, second = await asyncio.gather(first_task, second_task) + + assert call_count == 1 + assert first["readback_freshness"]["inflight_coalesced"] is False + assert second["readback_freshness"]["inflight_coalesced"] is True + assert first["consumer_bindings"] == second["consumer_bindings"] + + @pytest.mark.asyncio async def test_log_controlled_writeback_consumer_loader_reads_apply_receipts(monkeypatch): fake_db = _FakeDb(_ledger_rows(), _consumer_receipt_rows()) @@ -270,6 +338,39 @@ async def test_log_controlled_writeback_consumer_loader_degrades_on_db_pressure( assert payload["operation_boundaries"]["runtime_target_write_performed"] is False +@pytest.mark.asyncio +async def test_log_controlled_writeback_consumer_loader_does_not_cache_db_fallback( + monkeypatch, +): + monkeypatch.setattr( + consumer_module, + "get_db_context", + lambda project_id: _FailingContext(), + ) + direct_attempt_count = 0 + + async def fake_direct_unavailable(*, project_id: str): + nonlocal direct_attempt_count + assert project_id == "awoooi" + direct_attempt_count += 1 + return None + + monkeypatch.setattr( + consumer_module, + "_load_consumer_rows_direct", + fake_direct_unavailable, + ) + + first = await load_latest_ai_agent_log_controlled_writeback_consumer_readback() + second = await load_latest_ai_agent_log_controlled_writeback_consumer_readback() + + assert direct_attempt_count == 2 + assert first["readback_freshness"]["cache_hit"] is False + assert second["readback_freshness"]["cache_hit"] is False + assert first["readback"]["db_read_status"] == "unavailable" + assert second["readback"]["db_read_status"] == "unavailable" + + @pytest.mark.asyncio async def test_log_controlled_writeback_consumer_loader_uses_direct_fallback( monkeypatch, diff --git a/apps/api/tests/test_host_disk_pressure_ansible_catalog.py b/apps/api/tests/test_host_disk_pressure_ansible_catalog.py index d373e032c..1314b40c0 100644 --- a/apps/api/tests/test_host_disk_pressure_ansible_catalog.py +++ b/apps/api/tests/test_host_disk_pressure_ansible_catalog.py @@ -150,6 +150,109 @@ def test_historical_catalog_replay_is_per_failed_row_and_provider_silent() -> No assert "telegram_provider_delivery" in send_source +@pytest.mark.asyncio +async def test_catalog_replay_prioritizes_semantic_route_changes( + monkeypatch: pytest.MonkeyPatch, +) -> None: + from src.services import awooop_ansible_check_mode_service as service + + class Result: + def __init__(self, *, rows=None, scalar=None): + self._rows = rows or [] + self._scalar = scalar + + def mappings(self): + return self + + def all(self): + return self._rows + + def scalar_one_or_none(self): + return self._scalar + + class DB: + def __init__(self): + self.params: list[dict] = [] + + async def execute(self, _statement, params): + self.params.append(params) + if len(self.params) == 1: + return Result( + rows=[ + { + "op_id": "00000000-0000-0000-0000-000000000301", + "parent_op_id": ( + "00000000-0000-0000-0000-000000000300" + ), + "incident_id": "INC-NODE-REVISION-ONLY", + "incident_alertname": "NodeExporterDown", + "incident_alert_category": "availability", + "incident_affected_services": [ + "node-exporter-110" + ], + "input": { + "source_candidate_op_id": ( + "00000000-0000-0000-0000-000000000300" + ), + "incident_id": "INC-NODE-REVISION-ONLY", + "catalog_id": "ansible:110-devops", + "catalog_revision": "legacy-revision", + "playbook_path": ( + "infra/ansible/playbooks/" + "110-node-exporter-repair.yml" + ), + "inventory_hosts": ["host_110"], + "risk_level": "low", + }, + }, + { + "op_id": "00000000-0000-0000-0000-000000000311", + "parent_op_id": ( + "00000000-0000-0000-0000-000000000310" + ), + "incident_id": "INC-DISK-SEMANTIC-ROUTE", + "incident_alertname": "HostOutOfDiskSpace", + "incident_alert_category": "host_resource", + "incident_affected_services": [ + "node-exporter-188" + ], + "input": { + "source_candidate_op_id": ( + "00000000-0000-0000-0000-000000000310" + ), + "incident_id": "INC-DISK-SEMANTIC-ROUTE", + "catalog_id": "ansible:110-devops", + "catalog_revision": "legacy-revision", + "playbook_path": ( + "infra/ansible/playbooks/110-devops.yml" + ), + "inventory_hosts": ["host_110"], + "risk_level": "medium", + }, + }, + ] + ) + return Result( + scalar="00000000-0000-0000-0000-000000000312" + ) + + db = DB() + + @asynccontextmanager + async def fake_get_db_context(_project_id): + yield db + + monkeypatch.setattr(service, "get_db_context", fake_get_db_context) + + claims = await claim_catalog_drift_failed_check_modes(limit=1) + + assert db.params[0]["scan_limit"] == 100 + assert len(claims) == 1 + assert claims[0].incident_id == "INC-DISK-SEMANTIC-ROUTE" + assert claims[0].catalog_id == "ansible:188-disk-pressure" + assert claims[0].input_payload["catalog_route_changed"] is True + + def test_verified_wrong_catalog_is_reconciled_before_failure_replays() -> None: claim_source = inspect.getsource( claim_semantically_misrouted_verified_applies diff --git a/config/security/runtime-image-mirror-policy.json b/config/security/runtime-image-mirror-policy.json index 4c3ee3a50..e386d21a1 100644 --- a/config/security/runtime-image-mirror-policy.json +++ b/config/security/runtime-image-mirror-policy.json @@ -129,6 +129,21 @@ "container": "argocd-application-controller" } }, + { + "asset_id": "runtime-image:argocd-redis-secret-init", + "source_index_digest": "sha256:16b92ba472fbb9287459cc52e0ecff07288dff461209955098edb56ce866fe49", + "platform_digest": "sha256:2f25a42949ea69c0dd33f4ce1918c6a01039d6c14a7ecc1d19088504a9d3e94f", + "target_registry_digest": "sha256:8b7e25d7e6036d6750c5bda8920cbd80d8902c25771666f6767dc97bd7fab8fc", + "push_ref": "registry.wooo.work/awoooi/runtime-mirror/argocd:v3.3.6-linux-amd64", + "runtime_ref": "192.168.0.110:5000/awoooi/runtime-mirror/argocd@sha256:8b7e25d7e6036d6750c5bda8920cbd80d8902c25771666f6767dc97bd7fab8fc", + "workload": { + "kind": "deployment", + "namespace": "argocd", + "name": "argocd-redis", + "container": "secret-init", + "container_kind": "init_container" + } + }, { "asset_id": "runtime-image:kured", "source_index_digest": "sha256:2c5d73bb4517a269def38a6cd54d34d82be81793bea9ff1bb35c6533515ad209", diff --git a/k8s/awoooi-prod/06-deployment-api.yaml b/k8s/awoooi-prod/06-deployment-api.yaml index b529c6e5e..21a5ad4a7 100644 --- a/k8s/awoooi-prod/06-deployment-api.yaml +++ b/k8s/awoooi-prod/06-deployment-api.yaml @@ -160,12 +160,12 @@ spec: - name: AWOOOI_BUILD_COMMIT_SHA # 2026-06-29 Codex: CD rewrites this to the deployed image tag so # production deploy readback does not rely on a stale static snapshot. - value: "72bcbd7882f02990bfca6eaf67ea68cfa233c334" + value: "31de3a709815f33a11ba6de6e78b771b7160a245" - 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: "72bcbd7882f02990bfca6eaf67ea68cfa233c334" + value: "31de3a709815f33a11ba6de6e78b771b7160a245" - name: DATABASE_POOL_SIZE # 2026-07-01 Codex: production role `awoooi` currently has a low # connection limit. Keep API pool conservative until DB role diff --git a/k8s/awoooi-prod/08-deployment-ansible-executor-broker.yaml b/k8s/awoooi-prod/08-deployment-ansible-executor-broker.yaml index b26fc8b96..1d35cc038 100644 --- a/k8s/awoooi-prod/08-deployment-ansible-executor-broker.yaml +++ b/k8s/awoooi-prod/08-deployment-ansible-executor-broker.yaml @@ -66,7 +66,7 @@ spec: - name: DATABASE_NULL_POOL value: "true" - name: AWOOOI_BUILD_COMMIT_SHA - value: "72bcbd7882f02990bfca6eaf67ea68cfa233c334" + value: "31de3a709815f33a11ba6de6e78b771b7160a245" - name: ENABLE_AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WORKER value: "false" - name: ENABLE_AWOOOP_ANSIBLE_CHECK_MODE_WORKER diff --git a/k8s/awoooi-prod/08-deployment-worker.yaml b/k8s/awoooi-prod/08-deployment-worker.yaml index 9bcacfe99..df3125539 100644 --- a/k8s/awoooi-prod/08-deployment-worker.yaml +++ b/k8s/awoooi-prod/08-deployment-worker.yaml @@ -173,7 +173,7 @@ spec: - name: DATABASE_BOOTSTRAP_DDL_ENABLED value: "false" - name: AWOOOI_BUILD_COMMIT_SHA - value: "72bcbd7882f02990bfca6eaf67ea68cfa233c334" + value: "31de3a709815f33a11ba6de6e78b771b7160a245" - name: ENABLE_AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WORKER value: "true" - name: ENABLE_SECURITY_CONTROL_PLANE_MAINTENANCE_WORKER diff --git a/k8s/awoooi-prod/kustomization.yaml b/k8s/awoooi-prod/kustomization.yaml index 30b7db0af..14fabb734 100644 --- a/k8s/awoooi-prod/kustomization.yaml +++ b/k8s/awoooi-prod/kustomization.yaml @@ -40,9 +40,9 @@ resources: # ⚠️ 重要: name 必須與 deployment YAML 中的 image 完全匹配 (含 tag) # newName + newTag 由 CI 透過 kustomize edit set image 注入 images: -- digest: sha256:0f371518701b3fb115f67b16b9a6993adc98e0277c344a074a869ed5db28ee58 +- digest: sha256:a99b3b276def966c03be56f7e8e15feb69d5e2029c8d4ec11dd4b839709351b8 name: 192.168.0.110:5000/library/api:IMAGE_TAG_PLACEHOLDER newName: 192.168.0.110:5000/awoooi/api -- digest: sha256:18d791008184d6582e7f9f3ead8ed9b8d41c5e704f626d5e3686fd3b9e2b4a6c +- digest: sha256:9f3d1caa5808b2c690f37bd41c1776f9b881215b6211eb64b46bec2d5daf3474 name: 192.168.0.110:5000/library/web:IMAGE_TAG_PLACEHOLDER newName: 192.168.0.110:5000/awoooi/web diff --git a/scripts/security/tests/test_runtime_image_mirror_controller.py b/scripts/security/tests/test_runtime_image_mirror_controller.py index 8bd472241..4cfe9bf69 100644 --- a/scripts/security/tests/test_runtime_image_mirror_controller.py +++ b/scripts/security/tests/test_runtime_image_mirror_controller.py @@ -47,7 +47,7 @@ class RuntimeImageMirrorControllerTest(unittest.TestCase): self.assertEqual(policy.work_item_id, "AIA-P0-006-02A") self.assertEqual(policy.risk_level, "high") - self.assertEqual(len(policy.images), 12) + self.assertEqual(len(policy.images), 13) self.assertNotIn("github.com", raw) self.assertNotIn("ghcr.io", raw) self.assertIn('"external_pull_allowed": false', raw) @@ -73,7 +73,7 @@ class RuntimeImageMirrorControllerTest(unittest.TestCase): ) container_kinds = [image.workload.container_kind for image in policy.images] self.assertEqual(container_kinds.count("container"), 10) - self.assertEqual(container_kinds.count("init_container"), 2) + self.assertEqual(container_kinds.count("init_container"), 3) def test_policy_loads_and_validates_init_container_kind(self) -> None: image = self._image_with_container_kind("init_container") @@ -304,6 +304,29 @@ class RuntimeImageMirrorControllerTest(unittest.TestCase): image.runtime_ref.endswith("@" + image.target_registry_digest) ) + def test_argocd_redis_init_reuses_verified_staging_artifact(self) -> None: + policy = controller.load_policy(POLICY_PATH) + staging = controller.load_staging_policy(STAGING_POLICY_PATH).images[0] + image = next( + image + for image in policy.images + if image.asset_id == "runtime-image:argocd-redis-secret-init" + ) + + self.assertEqual(image.source_index_digest, staging.source_index_digest) + self.assertEqual(image.platform_digest, staging.platform_digest) + self.assertEqual(image.push_ref, staging.push_ref) + self.assertEqual(image.workload.kind, "deployment") + self.assertEqual(image.workload.namespace, "argocd") + self.assertEqual(image.workload.name, "argocd-redis") + self.assertEqual(image.workload.container, "secret-init") + self.assertEqual(image.workload.container_kind, "init_container") + self.assertEqual( + image.target_registry_digest, + "sha256:8b7e25d7e6036d6750c5bda8920cbd80d8902c25771666f6767dc97bd7fab8fc", + ) + self.assertTrue(image.runtime_ref.endswith("@" + image.target_registry_digest)) + def test_policy_rejects_external_pull_and_mutable_runtime_ref(self) -> None: payload = json.loads(POLICY_PATH.read_text(encoding="utf-8")) with tempfile.TemporaryDirectory() as temp: