fix(api): share operator summary cache through redis

apps/api/src/services/operator_summary_cache.py

@@ -66,6 +66,30 @@ def _with_cache_meta(value: dict[str, Any], meta: dict[str, Any]) -> dict[str, A
     return response
 
 
+def _redis_key(namespace: str, key_parts: dict[str, Any]) -> str:
+    return f"awooop:operator_summary:{_cache_key(namespace, key_parts)}"
+
+
+def _json_default(value: Any) -> str:
+    if isinstance(value, datetime):
+        return value.isoformat()
+    return str(value)
+
+
+def _decode_redis_value(raw: Any) -> dict[str, Any] | None:
+    if raw is None:
+        return None
+    if isinstance(raw, bytes):
+        raw = raw.decode("utf-8")
+    if not isinstance(raw, str):
+        return None
+    try:
+        payload = json.loads(raw)
+    except json.JSONDecodeError:
+        return None
+    return payload if isinstance(payload, dict) else None
+
+
 def get_cached_operator_summary(
     namespace: str,
     key_parts: dict[str, Any],
@@ -92,6 +116,58 @@ def get_cached_operator_summary(
     )
 
 
+async def get_cached_operator_summary_async(
+    namespace: str,
+    key_parts: dict[str, Any],
+    *,
+    ttl_seconds: int,
+    now_monotonic: float | None = None,
+) -> dict[str, Any] | None:
+    """Return a shared Redis cache hit, falling back to process memory."""
+    redis_key = _redis_key(namespace, key_parts)
+    try:
+        from src.core.redis_client import get_redis
+
+        redis_client = get_redis()
+        payload = _decode_redis_value(await redis_client.get(redis_key))
+        if payload is not None and isinstance(payload.get("value"), dict):
+            stored_epoch = float(payload.get("stored_epoch") or 0.0)
+            now_epoch = time.time()
+            ttl_value = max(1, int(ttl_seconds))
+            age_seconds = now_epoch - stored_epoch
+            if 0 <= age_seconds < ttl_value:
+                stored_at_raw = payload.get("stored_at")
+                try:
+                    stored_at = datetime.fromisoformat(str(stored_at_raw))
+                except ValueError:
+                    stored_at = datetime.now(UTC) - timedelta(seconds=age_seconds)
+                record = _CacheRecord(
+                    value=payload["value"],
+                    stored_at=stored_at,
+                    stored_monotonic=(now_monotonic or time.monotonic())
+                    - age_seconds,
+                    ttl_seconds=ttl_value,
+                )
+                return _with_cache_meta(
+                    record.value,
+                    _cache_meta(
+                        status="hit",
+                        record=record,
+                        age_seconds=age_seconds,
+                    ),
+                )
+            await redis_client.delete(redis_key)
+    except Exception:
+        pass
+
+    return get_cached_operator_summary(
+        namespace,
+        key_parts,
+        ttl_seconds=ttl_seconds,
+        now_monotonic=now_monotonic,
+    )
+
+
 def store_operator_summary(
     namespace: str,
     key_parts: dict[str, Any],
@@ -117,6 +193,46 @@ def store_operator_summary(
     )
 
 
+async def store_operator_summary_async(
+    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 in Redis and process memory."""
+    stored_at = now_utc or datetime.now(UTC)
+    ttl_value = max(1, int(ttl_seconds))
+    response = store_operator_summary(
+        namespace,
+        key_parts,
+        value,
+        ttl_seconds=ttl_value,
+        now_monotonic=now_monotonic,
+        now_utc=stored_at,
+    )
+    payload = {
+        "value": deepcopy(value),
+        "stored_at": stored_at.isoformat(),
+        "stored_epoch": time.time(),
+        "ttl_seconds": ttl_value,
+    }
+    try:
+        from src.core.redis_client import get_redis
+
+        redis_client = get_redis()
+        await redis_client.set(
+            _redis_key(namespace, key_parts),
+            json.dumps(payload, ensure_ascii=False, default=_json_default),
+            ex=ttl_value,
+        )
+    except Exception:
+        pass
+    return response
+
+
 def clear_operator_summary_cache() -> None:
     """Clear process-local cache for tests and controlled operator refreshes."""
     _CACHE.clear()