fix(security): restore compliance scanner readback

apps/api/src/jobs/compliance_scanner_job.py

@@ -29,7 +29,7 @@ from __future__ import annotations
 import asyncio
 import json as _json
 import time as _time
-from datetime import datetime, timedelta, timezone
+from datetime import UTC, datetime, timedelta, timezone
 from typing import Any
 
 import structlog
@@ -42,6 +42,7 @@ logger = structlog.get_logger(__name__)
 _FIRST_DELAY_SEC = 180
 _LOOP_BACKOFF_SEC = 1800
 _DAILY_TRIGGER_HOUR_TAIPEI = 3
+_PROJECT_ID = "awoooi"
 
 # 7 維 compliance (ADR-090 schema CHECK)
 _DIMENSIONS = (
@@ -143,10 +144,11 @@ async def scan_once(triggered_by: str = "cron") -> dict[str, int]:
 async def _fetch_active_assets() -> list[dict[str, Any]]:
     """從 asset_inventory 撈所有 active asset."""
     from sqlalchemy import text as _sql
+
     from src.db.base import get_db_context
 
     try:
-        async with get_db_context() as db:
+        async with get_db_context(_PROJECT_ID) as db:
             result = await db.execute(
                 _sql("""
                     SELECT asset_id, asset_key, asset_type, metadata
@@ -196,6 +198,7 @@ async def _write_compliance_for_asset(asset: dict[str, Any]) -> tuple[int, int,
     Returns: (snapshots_written, violations_count, warnings_count)
     """
     from sqlalchemy import text as _sql
+
     from src.db.base import get_db_context
 
     snapshots = 0
@@ -206,7 +209,7 @@ async def _write_compliance_for_asset(asset: dict[str, Any]) -> tuple[int, int,
     dimension_results = await asyncio.to_thread(_evaluate_all_dimensions, asset)
 
     try:
-        async with get_db_context() as db:
+        async with get_db_context(_PROJECT_ID) as db:
             for dim, (status, detail) in dimension_results.items():
                 await db.execute(
                     _sql("""
@@ -297,10 +300,9 @@ def _check_ssl_cert(asset: dict[str, Any]) -> tuple[str, dict]:
     if not https_target:
         return ("unknown", {"reason": "no https scrape_url / instance"})
 
-    import ssl
     import socket
+    import ssl
     from urllib.parse import urlparse
-    from datetime import datetime
 
     try:
         parsed = urlparse(https_target)
@@ -331,8 +333,8 @@ def _check_ssl_cert(asset: dict[str, Any]) -> tuple[str, dict]:
 
 def _check_ssl_cert_via_verified_socket(hostname: str, port: int) -> tuple[str, dict]:
     """用 verified socket 拿 dict form cert, 取 notAfter 判斷剩餘天數."""
-    import ssl
     import socket
+    import ssl
     from datetime import datetime
 
     try:
@@ -383,7 +385,7 @@ def _check_secret_rotation(asset: dict[str, Any]) -> tuple[str, dict]:
     except (ValueError, TypeError):
         return ("unknown", {"reason": f"unparseable timestamp: {created_ts[:50]}"})
 
-    now_utc = datetime.now(timezone.utc)
+    now_utc = datetime.now(UTC)
     age_days = (now_utc - created).days
 
     if age_days > _SECRET_ROTATION_WARNING_DAYS:
@@ -437,6 +439,7 @@ async def _llm_analyze_compliance_posture(
     """
     try:
         import json as _j
+
         from src.services.llm_json_parser import parse_llm_json_response
         from src.services.openclaw import get_openclaw
 
@@ -470,8 +473,12 @@ async def _send_telegram_posture(
     """推 Telegram 合規摘要 + 互動按鈕 (P0 修)."""
     try:
         import html
+
         from src.core.config import settings
-        from src.services.ai_advisory_helpers import build_ai_advisory_keyboard, is_snoozed
+        from src.services.ai_advisory_helpers import (
+            build_ai_advisory_keyboard,
+            is_snoozed,
+        )
         from src.services.telegram_gateway import get_telegram_gateway
 
         target_chat_id = settings.SRE_GROUP_CHAT_ID
@@ -543,11 +550,12 @@ async def _send_telegram_posture(
 async def _log_aol(stats: dict[str, int], duration_ms: int, triggered_by: str, error: str | None) -> None:
     try:
         from sqlalchemy import text as _sql
+
         from src.db.base import get_db_context
 
         aol_status = "failed" if error else "success"
 
-        async with get_db_context() as db:
+        async with get_db_context(_PROJECT_ID) as db:
             await db.execute(
                 _sql("""
                     INSERT INTO automation_operation_log (

apps/api/src/services/iwooos_security_asset_control_plane.py

@@ -1486,10 +1486,15 @@ class IwoooSSecurityAssetControlPlaneService:
                 source_id="security_compliance",
                 statement=_sql(
                     """
-                    WITH latest AS (
+                    WITH recent AS MATERIALIZED (
+                        SELECT snapshot_id, asset_id, dimension, status, detected_at
+                        FROM asset_compliance_snapshot
+                        ORDER BY snapshot_id DESC
+                        LIMIT 100000
+                    ), latest AS (
                         SELECT DISTINCT ON (asset_id, dimension)
                             asset_id, dimension, status
-                        FROM asset_compliance_snapshot
+                        FROM recent
                         ORDER BY asset_id, dimension, detected_at DESC
                     )
                     SELECT dimension, status, count(*) AS cnt

apps/api/tests/test_compliance_scanner_job.py

@@ -0,0 +1,72 @@
+from __future__ import annotations
+
+import asyncio
+
+from src.jobs import compliance_scanner_job
+
+
+class _AssetRow:
+    asset_id = 1
+    asset_key = "host/test"
+    asset_type = "host"
+    metadata = {}
+
+
+class _Result:
+    def fetchall(self):
+        return [_AssetRow()]
+
+
+class _Database:
+    async def execute(self, statement, parameters=None):
+        return _Result()
+
+
+class _Context:
+    async def __aenter__(self):
+        return _Database()
+
+    async def __aexit__(self, exc_type, exc, traceback):
+        return False
+
+
+def test_background_compliance_scanner_scopes_database_operations(monkeypatch) -> None:
+    requested_projects: list[str | None] = []
+
+    def fake_db_context(project_id: str | None = None):
+        requested_projects.append(project_id)
+        return _Context()
+
+    monkeypatch.setattr("src.db.base.get_db_context", fake_db_context)
+    monkeypatch.setattr(
+        compliance_scanner_job,
+        "_evaluate_all_dimensions",
+        lambda asset: {"audit_log_enabled": ("compliant", {"source": "test"})},
+    )
+
+    async def exercise_compliance_operations() -> None:
+        assets = await compliance_scanner_job._fetch_active_assets()
+        assert assets == [
+            {
+                "asset_id": 1,
+                "asset_key": "host/test",
+                "asset_type": "host",
+                "metadata": {},
+            }
+        ]
+
+        written, violations, warnings = (
+            await compliance_scanner_job._write_compliance_for_asset(assets[0])
+        )
+        assert (written, violations, warnings) == (1, 0, 0)
+
+        await compliance_scanner_job._log_aol(
+            stats={"assets_scanned": 1, "snapshots_written": 1},
+            duration_ms=10,
+            triggered_by="cron",
+            error=None,
+        )
+
+    asyncio.run(exercise_compliance_operations())
+
+    assert requested_projects == ["awoooi"] * 3

apps/api/tests/test_iwooos_security_asset_control_plane.py

@@ -345,6 +345,7 @@ def test_service_uses_canonical_awoooi_project_scope(monkeypatch) -> None:
 def test_service_bounds_source_concurrency_for_database_budget(monkeypatch) -> None:
     service = IwoooSSecurityAssetControlPlaneService()
     concurrency = {"active": 0, "maximum": 0}
+    statements: list[str] = []
 
     class EmptyResult:
         def fetchall(self):
@@ -358,6 +359,7 @@ def test_service_bounds_source_concurrency_for_database_budget(monkeypatch) -> N
 
     class SlowDb:
         async def execute(self, statement):
+            statements.append(str(statement))
             concurrency["active"] += 1
             concurrency["maximum"] = max(concurrency["maximum"], concurrency["active"])
             await asyncio.sleep(0.02)
@@ -381,6 +383,13 @@ def test_service_bounds_source_concurrency_for_database_budget(monkeypatch) -> N
     assert payload["summary"]["source_count"] == 9
     assert payload["summary"]["ready_source_count"] == 9
     assert payload["summary"]["unavailable_source_count"] == 0
+    compliance_query = next(
+        statement
+        for statement in statements
+        if "asset_compliance_snapshot" in statement
+    )
+    assert "ORDER BY snapshot_id DESC" in compliance_query
+    assert "LIMIT 100000" in compliance_query
 
 
 def test_public_api_returns_live_aggregate(monkeypatch) -> None: