fix(awooop): cache heavy operator summaries

apps/api/src/services/awooop_truth_chain_service.py

@@ -26,12 +26,19 @@ from src.services.operator_outcome import build_operator_outcome
 from src.services.awooop_ansible_check_mode_service import detect_ansible_transport_blockers
 from src.services.awooop_ansible_audit_service import build_ansible_truth
 from src.services.drift_repeat_state import build_drift_repeat_state
+from src.services.operator_summary_cache import (
+    get_cached_operator_summary,
+    store_operator_summary,
+)
 
 logger = structlog.get_logger(__name__)
 
 _MAX_ROWS = 100
 _JSON_TEXT_FIELDS = {"gate_result", "source_envelope"}
 _QUALITY_SUMMARY_CONCURRENCY = 8
+_QUALITY_SUMMARY_CACHE_TTL_SECONDS = int(
+    os.getenv("AWOOOP_QUALITY_SUMMARY_CACHE_TTL_SECONDS", "30")
+)
 
 
 def _clean(value: Any) -> Any:
@@ -1738,13 +1745,36 @@ async def fetch_automation_quality_summary(
     project_id: str = "awoooi",
     hours: int = 24,
     limit: int = 200,
+    refresh: bool = False,
 ) -> dict[str, Any]:
     """Return a recent incident-level quality summary for the automation flywheel."""
     bounded_hours = max(1, min(int(hours), 168))
     bounded_limit = max(1, min(int(limit), 500))
+    normalized_project_id = project_id or "awoooi"
+    cache_key = {
+        "project_id": normalized_project_id,
+        "hours": bounded_hours,
+        "limit": bounded_limit,
+    }
+    if not refresh:
+        cached_summary = get_cached_operator_summary(
+            "truth_chain_quality_summary",
+            cache_key,
+            ttl_seconds=_QUALITY_SUMMARY_CACHE_TTL_SECONDS,
+        )
+        if cached_summary is not None:
+            logger.info(
+                "awooop_automation_quality_summary_cache_hit",
+                project_id=normalized_project_id,
+                window_hours=bounded_hours,
+                limit=bounded_limit,
+                ttl_seconds=_QUALITY_SUMMARY_CACHE_TTL_SECONDS,
+            )
+            return cached_summary
+
     cutoff = datetime.now(UTC) - timedelta(hours=bounded_hours)
 
-    async with get_db_context(project_id) as db:
+    async with get_db_context(normalized_project_id) as db:
         incidents = await _fetch_all(
             db,
             """
@@ -1800,7 +1830,7 @@ async def fetch_automation_quality_summary(
             LIMIT :limit
             """,
             {
-                "project_id": project_id,
+                "project_id": normalized_project_id,
                 "cutoff": cutoff,
                 "limit": bounded_limit,
             },
@@ -1813,7 +1843,10 @@ async def fetch_automation_quality_summary(
         if not incident_id:
             return None
         async with semaphore:
-            truth_chain = await fetch_truth_chain(source_id=incident_id, project_id=project_id)
+            truth_chain = await fetch_truth_chain(
+                source_id=incident_id,
+                project_id=normalized_project_id,
+            )
         return {
             "incident": truth_chain.get("incident") or incident,
             "truth_status": truth_chain.get("truth_status") or {},
@@ -1828,17 +1861,24 @@ async def fetch_automation_quality_summary(
     ]
 
     summary = summarize_automation_quality_records(
-        project_id=project_id,
+        project_id=normalized_project_id,
         window_hours=bounded_hours,
         records=records,
         limit=bounded_limit,
     )
     logger.info(
         "awooop_automation_quality_summary_fetched",
-        project_id=project_id,
+        project_id=normalized_project_id,
         window_hours=bounded_hours,
         incident_total=summary["incident_total"],
         evaluated_total=summary["evaluated_total"],
         can_claim_full_auto_repair=summary["production_claim"]["can_claim_full_auto_repair"],
+        cache_status="miss",
+        cache_ttl_seconds=_QUALITY_SUMMARY_CACHE_TTL_SECONDS,
+    )
+    return store_operator_summary(
+        "truth_chain_quality_summary",
+        cache_key,
+        summary,
+        ttl_seconds=_QUALITY_SUMMARY_CACHE_TTL_SECONDS,
     )
-    return summary

apps/api/src/services/operator_summary_cache.py

@@ -0,0 +1,122 @@
+"""Short TTL cache for read-only AwoooP operator summaries.
+
+This cache intentionally lives in the API pod memory. It reduces repeated heavy
+operator-console reads without becoming a new source of truth.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import time
+from copy import deepcopy
+from dataclasses import dataclass
+from datetime import UTC, datetime, timedelta
+from typing import Any
+
+
+_CACHE_SCHEMA_VERSION = "operator_summary_cache_v1"
+_CACHE_SOURCE = "api_pod_memory"
+
+
+@dataclass(slots=True)
+class _CacheRecord:
+    value: dict[str, Any]
+    stored_at: datetime
+    stored_monotonic: float
+    ttl_seconds: int
+
+
+_CACHE: dict[str, _CacheRecord] = {}
+
+
+def _cache_key(namespace: str, key_parts: dict[str, Any]) -> str:
+    payload = json.dumps(
+        {"namespace": namespace, "key_parts": key_parts},
+        ensure_ascii=False,
+        sort_keys=True,
+        default=str,
+    )
+    digest = hashlib.sha256(payload.encode("utf-8")).hexdigest()
+    return f"{namespace}:{digest}"
+
+
+def _cache_meta(
+    *,
+    status: str,
+    record: _CacheRecord,
+    age_seconds: float,
+) -> dict[str, Any]:
+    ttl_seconds = max(1, int(record.ttl_seconds))
+    expires_at = record.stored_at + timedelta(seconds=ttl_seconds)
+    return {
+        "schema_version": _CACHE_SCHEMA_VERSION,
+        "status": status,
+        "source": _CACHE_SOURCE,
+        "ttl_seconds": ttl_seconds,
+        "age_seconds": round(max(0.0, age_seconds), 3),
+        "stored_at": record.stored_at.isoformat(),
+        "expires_at": expires_at.isoformat(),
+    }
+
+
+def _with_cache_meta(value: dict[str, Any], meta: dict[str, Any]) -> dict[str, Any]:
+    response = deepcopy(value)
+    response["cache"] = meta
+    return response
+
+
+def get_cached_operator_summary(
+    namespace: str,
+    key_parts: dict[str, Any],
+    *,
+    ttl_seconds: int,
+    now_monotonic: float | None = None,
+) -> dict[str, Any] | None:
+    """Return cached summary with hit metadata, or None if absent/expired."""
+    cache_key = _cache_key(namespace, key_parts)
+    record = _CACHE.get(cache_key)
+    if record is None:
+        return None
+
+    now_value = time.monotonic() if now_monotonic is None else now_monotonic
+    ttl_value = max(1, int(ttl_seconds))
+    age_seconds = now_value - record.stored_monotonic
+    if age_seconds >= ttl_value:
+        _CACHE.pop(cache_key, None)
+        return None
+
+    return _with_cache_meta(
+        record.value,
+        _cache_meta(status="hit", record=record, age_seconds=age_seconds),
+    )
+
+
+def store_operator_summary(
+    namespace: str,
+    key_parts: dict[str, Any],
+    value: dict[str, Any],
+    *,
+    ttl_seconds: int,
+    now_monotonic: float | None = None,
+    now_utc: datetime | None = None,
+) -> dict[str, Any]:
+    """Store a fresh summary and return it with miss metadata."""
+    cache_key = _cache_key(namespace, key_parts)
+    stored_at = now_utc or datetime.now(UTC)
+    record = _CacheRecord(
+        value=deepcopy(value),
+        stored_at=stored_at,
+        stored_monotonic=time.monotonic() if now_monotonic is None else now_monotonic,
+        ttl_seconds=max(1, int(ttl_seconds)),
+    )
+    _CACHE[cache_key] = record
+    return _with_cache_meta(
+        record.value,
+        _cache_meta(status="miss", record=record, age_seconds=0.0),
+    )
+
+
+def clear_operator_summary_cache() -> None:
+    """Clear process-local cache for tests and controlled operator refreshes."""
+    _CACHE.clear()

apps/api/src/services/platform_operator_service.py

@@ -9,6 +9,7 @@ ADR-106（AwoooP Agent Platform）
 from __future__ import annotations
 
 import asyncio
+import os
 import re
 import time
 import uuid
@@ -59,6 +60,10 @@ from src.services.ollama_failover_manager import (
 )
 from src.services.ollama_health_monitor import HealthReport, HealthStatus
 from src.services.operator_outcome import build_operator_outcome
+from src.services.operator_summary_cache import (
+    get_cached_operator_summary,
+    store_operator_summary,
+)
 from src.services.run_state_machine import transition
 
 logger = structlog.get_logger(__name__)
@@ -74,6 +79,9 @@ _MAX_STEP_SUMMARY_CHARS = 128
 _AI_ROUTE_STATUS_SELECT_TIMEOUT_SECONDS = 12.0
 _AI_ROUTE_STATUS_CONNECTIVITY_TIMEOUT_SECONDS = 2.5
 _REMEDIATION_HISTORY_LIMIT = 20
+_CALLBACK_REPLY_CACHE_TTL_SECONDS = int(
+    os.getenv("AWOOOP_CALLBACK_REPLY_CACHE_TTL_SECONDS", "20")
+)
 _INCIDENT_ID_RE = re.compile(r"\bINC-\d{8}-[A-Z0-9]{4,}\b")
 _REMEDIATION_STATUS_FILTERS = {
     "mcp_observed",
@@ -304,11 +312,13 @@ async def list_callback_replies(
     incident_id: str | None,
     page: int,
     per_page: int,
+    refresh: bool = False,
 ) -> dict[str, Any]:
     """列出 Telegram detail/history callback reply evidence，不改 runtime 狀態。"""
     _validate_callback_reply_status_filter(callback_reply_status)
     callback_action = _validate_callback_reply_action_filter(action)
     _validate_incident_id_filter(incident_id)
+    normalized_project_id = project_id or "awoooi"
 
     if callback_reply_status == "no_callback":
         return {
@@ -318,6 +328,33 @@ async def list_callback_replies(
             "per_page": per_page,
         }
 
+    cache_key = {
+        "project_id": project_id or "__all__",
+        "callback_reply_status": callback_reply_status or "",
+        "action": callback_action or "",
+        "incident_id": incident_id or "",
+        "page": page,
+        "per_page": per_page,
+    }
+    if not refresh:
+        cached_response = get_cached_operator_summary(
+            "callback_replies",
+            cache_key,
+            ttl_seconds=_CALLBACK_REPLY_CACHE_TTL_SECONDS,
+        )
+        if cached_response is not None:
+            logger.info(
+                "operator_callback_replies_cache_hit",
+                project_id=normalized_project_id,
+                callback_reply_status=callback_reply_status,
+                action=callback_action,
+                incident_id=incident_id,
+                page=page,
+                per_page=per_page,
+                ttl_seconds=_CALLBACK_REPLY_CACHE_TTL_SECONDS,
+            )
+            return cached_response
+
     where_clauses = [
         "m.source_envelope ? 'callback_reply'",
     ]
@@ -399,14 +436,14 @@ async def list_callback_replies(
         LIMIT :limit OFFSET :offset
     """)
 
-    async with get_db_context(project_id or "awoooi") as db:
+    async with get_db_context(normalized_project_id) as db:
         count_result = await db.execute(count_sql, params)
         total = count_result.scalar_one()
         rows_result = await db.execute(list_sql, params)
         rows = list(rows_result.mappings().all())
         audit_summary = await _fetch_callback_reply_audit_summary(
             db,
-            project_id=project_id or "awoooi",
+            project_id=normalized_project_id,
         )
 
     items = [_callback_reply_event_item(row) for row in rows]
@@ -459,13 +496,31 @@ async def list_callback_replies(
             km_completion_summary_cache[summary_cache_key] = km_summary
         item["km_stale_completion_summary"] = km_summary
 
-    return {
+    response = {
         "items": items,
         "total": total,
         "page": page,
         "per_page": per_page,
         "summary": audit_summary,
     }
+    logger.info(
+        "operator_callback_replies_fetched",
+        project_id=normalized_project_id,
+        callback_reply_status=callback_reply_status,
+        action=callback_action,
+        incident_id=incident_id,
+        page=page,
+        per_page=per_page,
+        total=total,
+        cache_status="miss",
+        cache_ttl_seconds=_CALLBACK_REPLY_CACHE_TTL_SECONDS,
+    )
+    return store_operator_summary(
+        "callback_replies",
+        cache_key,
+        response,
+        ttl_seconds=_CALLBACK_REPLY_CACHE_TTL_SECONDS,
+    )
 
 
 async def _fetch_callback_reply_audit_summary(