PolyGateway/src/polygateway/middleware/cache.py

"""CacheMW: 响应缓存中间件;key 公式在此(算法一份),后端是笨 KV。

ARCH §7.5: key = sha256(canonical_json({model, messages_digest, namespace,
salt}));多模态 part 先摘要再 hash(防 Video-Tree 整段 base64 进 hash 的
开销);namespace 必填防跨项目/租户毒化;只缓存阶梯通过的成功响应
(StructuredMW 在内层,能返回即已通过)。读写失败静默降级(铁律)。
"""

from __future__ import annotations

import asyncio
import dataclasses
import hashlib
import json
import uuid
from typing import TYPE_CHECKING, Any

from loguru import logger

from polygateway.types import ChatRequest, Effort, LLMResponse, ThinkingObservation

if TYPE_CHECKING:
    from collections.abc import Mapping

    from polygateway.ports import CacheBackend, CallNext, StructuredOutputStrategy

_KEY_PREFIX = "pgw:cache:"
_RESPONSE_FIELDS = {f.name for f in dataclasses.fields(LLMResponse)}


def _coerce_observation(raw: Any) -> ThinkingObservation:
    """缓存里的三态取值 → 枚举;域外取值降级为 `UNKNOWN`,**不作废整条缓存**。

    方向选择的理由: `_rehydrate` 对 JSON 里的**新字段**已经是宽容的(先按
    `_RESPONSE_FIELDS` 过滤),对同一字段的**新取值**却不该是致命的。真实场景是
    多个项目共用一个 Redis,先升级的那个写入了本版没有的取值,未升级的项目若把
    这些条目判成未命中,就会每次真打网关、随后覆写回旧值,两个版本互相打对方的
    缓存(表现是命中率莫名腰斩,而通用的"重建失败"文案给不出任何线索)。一个纯
    可观测性字段不该有能力废掉内容完好的缓存响应——"整条作废"留给真正破坏内容
    完整性的失败(JSON 坏了、结构化重建不过)。

    降级到 `UNKNOWN` 而不是别的态: 它的语义恰好就是"本次判不出来",对一个本库
    读不懂的取值,这是唯一诚实的说法。
    """
    try:
        return ThinkingObservation(raw)
    except ValueError:
        logger.warning(
            "缓存条目的 thinking_observation 取值 {!r} 不在本版取值域内(多半由更新版本的"
            "进程写入),已降级为 UNKNOWN;响应内容照常复活——可观测性字段不作废缓存",
            raw,
        )
        return ThinkingObservation.UNKNOWN


def digest_messages(messages: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """多模态 content part 先各自 sha256 摘要再参与序列化;文本原文参与。

    与遥测落库共用同一函数(ARCH §7.8),保证缓存 key 与遥测口径一致。
    """
    digested = []
    for msg in messages:
        content = msg.get("content")
        if isinstance(content, list):
            parts = [_digest_part(part) for part in content]
            digested.append({**msg, "content": parts})
        else:
            digested.append(msg)
    return digested


def _digest_part(part: Any) -> Any:
    if isinstance(part, dict) and part.get("type") == "image_url":
        url = str(part.get("image_url", {}).get("url", ""))
        return {"type": "image_url", "sha256": hashlib.sha256(url.encode()).hexdigest()}
    return part


def build_cache_key(
    model_fingerprint: str,
    messages: list[dict[str, Any]],
    namespace: str,
    salt: str | None,
    *,
    sampling: Mapping[str, Any] | None = None,
    reasoning_effort: Effort | None = None,
) -> str:
    """缓存 key 公式;salt 仅非 None 时参与(VT 旧键语义: 不传 salt 键形不变)。

    `sampling` 仅**非空**时参与(与 salt 的"仅非 None"不同——空串是有意义的
    salt,而空采样参数与不传无语义差别)。它必须进 key: 否则同 messages 跑 5 个
    seed 会全部命中第一次的响应,标准差恒为 0 且不报错(issue #4 决策 C)。

    `reasoning_effort` 是**请求级**档位(issue #20),仅非 `None` 时参与。它不能靠
    `model_fingerprint` 代劳: 后者是**装配期**算出的集合级指纹,一次调用改档位不会
    让它变一个字节;不进 key 则同 messages 跑 low 与 max 互相命中,是 issue #4
    「5 个 seed 全命中同一响应」的逐字翻版。

    判据用 `is not None` 而非真值: `Effort.NONE`(明确要求不推理)与 `None`
    (不表态)语义不同——前者拿到的是没有推理过程的响应,合并即毒化。
    """
    key_obj: dict[str, Any] = {
        "model": model_fingerprint,
        "messages": digest_messages(messages),
        "namespace": namespace,
    }
    if salt is not None:
        key_obj["salt"] = salt
    if sampling:
        key_obj["sampling"] = dict(sampling)
    if reasoning_effort is not None:
        key_obj["reasoning_effort"] = str(reasoning_effort)
    payload = json.dumps(key_obj, sort_keys=True, ensure_ascii=False)
    return _KEY_PREFIX + hashlib.sha256(payload.encode("utf-8")).hexdigest()


class CacheMW:
    """洋葱第二层(遥测内、结构化外)。

    model_fingerprint 由装配层从源列表计算(多源 scope = 排序去重的 model
    名合集);源集合变化 → key 变化 → 一次性冷启动,换取跨源集合零毒化。
    """

    def __init__(
        self,
        *,
        backend: CacheBackend,
        model_fingerprint: str,
        default_namespace: str,
        ttl_s: int,
        strategy: StructuredOutputStrategy | None = None,
    ) -> None:
        if not default_namespace.strip():
            raise ValueError("缓存 namespace 不能为空(防跨项目/租户毒化)")
        if ttl_s <= 0:
            raise ValueError("缓存 TTL 必须 > 0(禁止永不过期)")
        self._backend = backend
        self._fingerprint = model_fingerprint
        self._namespace = default_namespace
        self._ttl_s = ttl_s
        self._strategy = strategy

    async def __call__(self, request: ChatRequest, call_next: CallNext) -> LLMResponse:
        namespace = request.cache_namespace or self._namespace
        # 读 sampling 而非 overlay: 语义明确,且不依赖"CacheMW 恰在 StructuredMW
        # 外侧"这一层序巧合——结构化注入不该改变缓存身份(设计决策 C)
        key = build_cache_key(
            self._fingerprint,
            request.messages,
            namespace,
            request.cache_salt,
            sampling=request.sampling,
            # 请求级档位必须逐次进 key: `self._fingerprint` 是装配期的集合级指纹,
            # 同一个 client 上 low 与 max 两次调用在它眼里毫无分别(issue #20)
            reasoning_effort=request.reasoning_effort,
        )
        cached = await self._safe_get(key)
        if cached is not None:
            hit = self._rehydrate(cached, request)
            if hit is not None:
                return hit
        response = await call_next(request)
        await self._safe_set(key, self._serialize(response))
        return response

    # —— 命中路径 ——

    def _rehydrate(self, raw: str, request: ChatRequest) -> LLMResponse | None:
        """反序列化 + 按本次调用的 structured 档零网络重建;失败按未命中。"""
        try:
            data = json.loads(raw)
            fields = {k: v for k, v in data.items() if k in _RESPONSE_FIELDS}
            structured_data = self._rebuild_structured(fields.get("content", ""), request)
            # JSON 里存的是 StrEnum 的字符串值,不转就复活成裸 str,与字段注解分叉
            # (下游 `is ThinkingObservation.OBSERVED` 会在命中路径上静默为 False);
            # 键缺失即升级前写入的旧条目,交给 dataclass 默认值
            if "thinking_observation" in fields:
                fields["thinking_observation"] = _coerce_observation(fields["thinking_observation"])
            fields.update(
                cache_hit=True,
                latency_ms=0,
                ttft_ms=None,
                max_inter_token_ms=None,
                call_id=str(uuid.uuid4()),
                structured_data=structured_data,
            )
            return LLMResponse(**fields)
        except Exception as exc:
            logger.warning("缓存命中重建失败,按未命中回源: {}", exc)
            return None

    def _rebuild_structured(self, content: str, request: ChatRequest) -> Any | None:
        """对缓存 content 重跑阶梯②③(schema 变更后旧缓存自动重校验,设计 §2.1)。"""
        if request.structured is None:
            return None
        if self._strategy is None:
            raise ValueError("structured 调用需要装配 strategy 才能命中重建")
        parsed = self._strategy.parse(content)
        if request.structured == "json":
            return parsed
        return request.structured.model_validate(parsed)

    # —— 写路径与降级 ——

    def _serialize(self, response: LLMResponse) -> str:
        data = dataclasses.asdict(response)
        data.pop("structured_data", None)  # pydantic 实例不可 JSON 往返(设计 §2.1)
        return json.dumps(data, ensure_ascii=False)

    async def _safe_get(self, key: str) -> str | None:
        try:
            return await self._backend.get(key)
        except asyncio.CancelledError:
            raise
        except Exception as exc:
            logger.warning("缓存读取失败,降级为未命中: {}", exc)
            return None

    async def _safe_set(self, key: str, value: str) -> None:
        try:
            await self._backend.set(key, value, self._ttl_s)
        except asyncio.CancelledError:
            raise
        except Exception as exc:
            logger.warning("缓存写入失败,跳过缓存: {}", exc)