projectAIpopular/gateway/proxy/routes.py

"""代理面路由（T-P4）：学生面对话 + 管理面 CRUD。

主时序（§7，缓存分支 T-P6 接入）：
  auth -> 限流/日额 -> body 校验(413) -> 预扣 try_hold(est) -> 上游流式 tee
  -> usage 归一 -> compute -> settle(actual, 回补) -> record(流水)
  异常：首 token 前 failover 均失败 -> void；流中失败 -> aborted 按已收 usage 结算。

管理面：X-Admin-Key（hmac.compare_digest；未配置仅 loopback）。
"""
from __future__ import annotations

import asyncio
import json
import time
import uuid
from typing import Any, Dict, Optional

from fastapi import APIRouter, Request
from fastapi.responses import JSONResponse, StreamingResponse

from gateway.proxy.auth import authenticate, verify_admin
from gateway.proxy.config import ProxyConfig
from gateway.proxy.errors import (
    AdminAuthError,
    BalanceError,
    BodyTooLargeError,
    ProxyAuthError,
    QuotaError,
    SuspendedError,
    UpstreamError,
)
from gateway.proxy.pricing import compute
from gateway.proxy.upstream import (UpstreamAborted, failover_stats as upstream_failover_stats,
                                    filter_usage_chunk, stream as upstream_stream)


def install_error_handlers(app) -> None:
    """把 ProxyError 家族映射为 §5.3 错误码（401/402/403/413/429/502）。"""
    from gateway.proxy.errors import ProxyError

    @app.exception_handler(ProxyError)
    async def _proxy_error_handler(request: Request, exc: ProxyError):
        return JSONResponse(
            {"error": {"message": str(exc), "type": exc.code}},
            status_code=exc.status_code,
            headers={"WWW-Authenticate": "Bearer"} if exc.status_code == 401 else None)


def build_proxy_router(cfg: ProxyConfig, pool, settings_provider=None,
                       review_getter=None) -> APIRouter:
    """组装代理面路由（唯一组装点）。

    settings_provider：返回设置对象的回调（api.py 注入 settings_store）。
    未注入时语义分析器分流不启用（None 安全，既有测试不受影响）。
    review_getter：返回人工检验队列的回调（api.py 注入 get_review）；
    未注入时 T1 审计抽样跳过（不报错）。
    """
    router = APIRouter(prefix="/proxy")
    ledger = __import__("gateway.proxy.ledger", fromlist=["Ledger"]).Ledger.init_db(cfg.db_path)
    # ---------------- 学生面 ----------------
    @router.get("/v1/models", tags=["proxy"])
    async def list_models():
        """学生面：池内允许代理的模型列表（OpenAI /models 形状）。"""
        entries = pool.list().get("entries", [])
        seen, data = set(), []
        for e in entries:
            if not e.get("enabled") or e.get("backend") in ("mock",):
                continue
            mid = e.get("model") or ""
            if mid and mid not in seen:
                seen.add(mid)
                data.append({"id": mid, "object": "model", "owned_by": "campus-proxy"})
        return {"object": "list", "data": data}

    @router.post("/v1/chat/completions", tags=["proxy"])
    async def chat_completions(request: Request):
        raw_body = await request.body()
        if len(raw_body) > cfg.max_body_chars:
            raise BodyTooLargeError(f"请求体超过 {cfg.max_body_chars} 字符")
        try:
            body = json.loads(raw_body or b"{}")
        except json.JSONDecodeError:
            return JSONResponse({"error": {"message": "请求体不是合法 JSON"}}, status_code=400)

        ctx = authenticate(request.headers.get("authorization", ""), ledger,
                           _limits(), time.time())
        if not _limits().acquire_slot(ctx["key_id"]):
            raise QuotaError("并发请求已达该 key 上限")

        # 语义分析器 live 分流（D-G7：mode=live 才启用；任何异常不影响代理可用性）。
        # T-X4 修复：settings_provider 由 api.py 注入（原实现引用未导入的
        # settings_store，NameError 被 except 吞掉，分流从未实际生效）。
        # 2026-09 优化：sense 运行时按配置单例化（原每请求重建 SenseStore+Grader：
        # 每请求跑全量 DDL、重读工件文件，且 T-X3 决策缓存随 Grader 丢弃、命中率恒 0）。
        tier_used = None
        request_id = "px" + uuid.uuid4().hex[:12]
        if settings_provider is not None:
            try:
                from gateway.sense.config import build_sense_config
                scfg = build_sense_config(settings_provider().to_dict())
                if scfg.enabled and scfg.mode == "live":
                    _sstore, grader = _get_sense_runtime(scfg)
                    qtext = "\n".join(str(m.get("content") or "")
                                      for m in (body.get("messages") or []))
                    d = await grader.decide(qtext or str(body.get("model") or ""),
                                            "proxy")
                    tier_used = d.tier
                    # 档位 -> 池条目（§2 消费方表）：t1/t2/t3 映射模型替换请求模型
                    hint = scfg.tier_pool_hint("proxy", tier_used)
                    e = _entry_by_tier_hint(pool, hint)
                    if e is not None:
                        body = {**body, "model": e["model"]}
            except Exception:
                pass  # D-G4：分级故障不影响代理可用性
        try:
            resp = await _run_chat(body, dict(request.headers), ctx, cfg, ledger, pool,
                                   request_id=request_id)
            if tier_used:
                resp.headers["x-campus-tier"] = tier_used
                # T1 审计抽样（§9.4）：按 review.sample_rate 入队人工核；
                # promo 标签供人工 verdict 回填晋升表（T-X5）。
                # 2026-09 修复：原块引用未定义的 load_config/get_review/request_id，
                # 三个 NameError 被 except 吞掉，抽样从未入队；review 队列现由
                # api.py 经 review_getter 注入（避免 proxy -> api 反向依赖）。
                try:
                    import random as _random
                    from router_system.config import load_config
                    from gateway.sense.promotion import promotion_label
                    rate = float(load_config().get("review", {}).get("sample_rate", 0.1))
                    if review_getter is not None and _random.random() < rate:
                        review_getter().enqueue(
                            request_id + "-sense", str(body.get("model") or "proxy"),
                            "(sense T1 审计抽样)",
                            tags=["sense_t1", "promo:" + promotion_label("proxy", "")],
                            reason="sense_audit")
                except Exception:
                    pass
            return resp
        finally:
            _limits().release_slot(ctx["key_id"])

    def _entry_by_tier_hint(pool, hint: str) -> Optional[Dict[str, Any]]:
        """按档位名选池条目（tier_hint 优先，缺省按 tier 字段映射）。"""
        entries = _pool_entries(pool)
        for e in entries:
            if not e.get("enabled") or e.get("backend") == "mock":
                continue
            if (e.get("tier_hint") or e.get("tier")) == hint:
                return e
        return None

    # ---------------- 管理面 ----------------
    def _guard_admin(request: Request) -> None:
        host = request.client.host if request.client else ""
        if not verify_admin(request.headers.get("x-admin-key", ""), cfg.admin_key, host):
            raise AdminAuthError("管理面鉴权失败")

    @router.post("/admin/students", tags=["proxy-admin"])
    async def admin_create_student(request: Request, payload: dict):
        _guard_admin(request)
        name = str((payload or {}).get("name") or "").strip()
        if not name:
            return JSONResponse({"error": "name 必填"}, status_code=400)
        sid = ledger.upsert_student(
            name, klass=str((payload or {}).get("class") or ""),
            balance_yuan=float((payload or {}).get("balance_yuan", 0) or 0),
            daily_cap_yuan=float((payload or {}).get("daily_cap_yuan", 5) or 5))
        return {"student_id": sid, **(ledger.get_student(sid) or {})}

    @router.post("/admin/students/{student_id}/topup", tags=["proxy-admin"])
    async def admin_topup(student_id: int, request: Request, payload: dict):
        _guard_admin(request)
        amount = float((payload or {}).get("amount_yuan", 0) or 0)
        new_balance = ledger.topup(student_id, amount)
        if new_balance is None:
            return JSONResponse({"error": f"学生不存在: {student_id}"}, status_code=404)
        return {"ok": True, "balance_milli": new_balance}

    @router.post("/admin/keys", tags=["proxy-admin"])
    async def admin_issue_key(request: Request, payload: dict):
        _guard_admin(request)
        from gateway.proxy.auth import issue_key
        student_id = int((payload or {}).get("student_id", 0) or 0)
        try:
            return issue_key(ledger, student_id,
                             rpm_cap=payload.get("rpm_cap"),
                             day_cap_req=payload.get("day_cap_req"))
        except ProxyAuthError as e:
            return JSONResponse({"error": str(e)}, status_code=404)

    @router.post("/admin/keys/{key_id}/revoke", tags=["proxy-admin"])
    async def admin_revoke_key(key_id: int, request: Request):
        _guard_admin(request)
        ok = ledger.revoke_key(key_id)
        return {"ok": ok}

    @router.get("/admin/stats", tags=["proxy-admin"])
    async def admin_stats(request: Request, since: int = 0):
        """命中率-毛利看板（§5.2 口径）：h_g/h_p/revenue/cost/margin/by_bucket。

        2026-09 优化：SQL 聚合下推（原 list_usage(limit=500000) 全量拉回
        Python 扫 4 遍，随流水线性劣化）；响应字段与四舍五入口径不变。
        """
        _guard_admin(request)
        from gateway.proxy.ledgerutil import _today
        import time as _t
        now = _t.time()
        today = _today(now)
        s = ledger.usage_stats()
        n = s["requests"]
        cached = s["gateway_cached"]
        h_g = cached / n if n else 0.0
        in_hit = s["in_hit_tok"]
        in_miss = s["in_miss_tok"]
        h_p = in_hit / (in_hit + in_miss) if (in_hit + in_miss) else 0.0
        return {"requests": n, "h_g": round(h_g, 4), "h_p": round(h_p, 4),
                "revenue_milli": s["revenue_milli"], "cost_milli": s["cost_milli"],
                "margin_milli": s["revenue_milli"] - s["cost_milli"],
                "by_bucket": s["by_bucket"],
                "today": today,
                "upstream_failover": upstream_failover_stats()}

    @router.get("/admin/sense-decisions", tags=["proxy-admin"])
    async def admin_sense_decisions(request: Request, limit: int = 50):
        """路由决策留痕查询（T-X4）：reasons/candidate_scores/rejected 三元结构。

        管理面鉴权同 /admin/stats；sense 未启用时返回空集（不报错，便于前端
        统一渲染）。
        """
        _guard_admin(request)
        try:
            from gateway.sense.config import build_sense_config
            from gateway.sense.store import SenseStore
            if settings_provider is None:
                return {"enabled": False, "decisions": []}
            scfg = build_sense_config(settings_provider().to_dict())
            if not scfg.enabled:
                return {"enabled": False, "decisions": []}
            sstore = SenseStore.init_db(scfg.db_path)
            return {"enabled": True,
                    "decisions": sstore.list_decisions(limit=limit)}
        except Exception:
            return {"enabled": False, "decisions": []}

    @router.get("/admin/ledger", tags=["proxy-admin"])
    async def admin_ledger(request: Request, student_id: int = 0,
                           limit: int = 50, offset: int = 0):
        """流水分页（§5.2）。"""
        _guard_admin(request)
        return ledger.list_usage(student_id=student_id or None,
                                 limit=max(1, min(limit, 200)),
                                 offset=max(0, offset))

    @router.get("/admin/students", tags=["proxy-admin"])
    async def admin_list_students(request: Request):
        """学生列表（key 管理卡片）。

        注：学生表直读查询因安全扫描误报暂缓入库（ledger.list_students 待补），
        当前列表由签发/充值时的写入响应累积（前端本地态）；端点先返回 501。
        """
        _guard_admin(request)
        return JSONResponse({"error": {"message": "学生列表查询待补（安全扫描误报阻塞）",
                                       "type": "not_implemented"}},
                            status_code=501)

    return router


def _limits():
    from gateway.proxy.auth import _AUTH_SINGLETON_LIMITS
    return _AUTH_SINGLETON_LIMITS


def _pool_entries(pool) -> List[Dict[str, Any]]:
    """池条目快照（转发链内部专用）：优先未打码通道 usable_entries，
    回退打码 list()（兼容测试注入的简化 pool）。HTTP 响应一律用 list()。"""
    fn = getattr(pool, "usable_entries", None)
    if fn is not None:
        return fn()
    return pool.list().get("entries", [])


def _estimate_hold_milli(body: dict, cfg: ProxyConfig, model: str) -> int:
    """预扣估算（宁可高估，D-P11）：in 按字符/3，out 按 min(max_tokens,4096)。

    只能用已配置模型的峰值单价；未配置模型按 default 桶兜底价 or 最高价。
    """
    price = cfg.price(model)
    if price is None:
        prices = cfg.model_prices.values()
        if not prices:
            return 1
        in_p = max(p.in_miss for p in prices)
        hit_p = max(p.in_hit for p in prices)
        out_p = max(p.out for p in prices)
    else:
        in_p, hit_p, out_p = price.in_miss, price.in_hit, price.out
    prompt_chars = sum(len(str(m.get("content") or ""))
                       for m in (body.get("messages") or []))
    est_in = prompt_chars // 3
    est_out = int(body.get("max_tokens") or 1024)
    est_out = min(est_out, 4096)
    cost = (est_in * in_p + est_out * out_p) / 1_000_000
    return max(1, int(round(cost)))          # 保守：全部按未命中价


def _resolve_entry(pool, model: str, cfg: ProxyConfig) -> Optional[Dict[str, Any]]:
    """按模型名取池条目（仅启用的真实后端）。"""
    e = pool.find_by_model(model)
    if e and e.get("enabled") and e.get("backend") not in ("mock",) and e.get("base_url"):
        return e
    return None


def _budget_headers(budget_mode: str) -> Dict[str, str]:
    """预算档位响应头（normal 不发，避免噪音）。"""
    return {"X-Budget-Mode": budget_mode} if budget_mode and budget_mode != "normal" else {}


def _request_needs(body: dict) -> Dict[str, Any]:
    """从请求体推断能力需求（T-X6）：多模态图片 -> vision；带 tools -> tools；
    上下文需求 = 提示字符/3 + min(max_tokens, 4096)（与预扣估算同口径）。"""
    vision = False
    for m in body.get("messages") or []:
        content = m.get("content")
        if isinstance(content, list):
            for part in content:
                if isinstance(part, dict) and ("image_url" in part
                                               or part.get("type") == "image_url"):
                    vision = True
    prompt_chars = sum(len(str(m.get("content") or ""))
                       if not isinstance(m.get("content"), list)
                       else sum(len(str(p.get("text") or "")) for p in m["content"]
                                if isinstance(p, dict))
                       for m in (body.get("messages") or []))
    est_out = min(int(body.get("max_tokens") or 1024), 4096)
    return {"vision": vision,
            "tools": bool(body.get("tools")),
            "min_context_tokens": prompt_chars // 3 + est_out}


def _entry_meets(entry: Dict[str, Any], needs: Dict[str, Any]) -> bool:
    """条目能力位是否满足请求需求（capabilities 缺省全兼容）。"""
    cap = entry.get("capabilities") or {}
    if needs["vision"] and not cap.get("vision", True):
        return False
    if needs["tools"] and not cap.get("tools", True):
        return False
    ctx = int(cap.get("context_window") or 0)
    if needs["min_context_tokens"] > 0 and 0 < ctx < needs["min_context_tokens"]:
        return False
    return True


def _fallback_chain(pool, entry: Dict[str, Any], max_total: int = 3) -> List[Dict[str, Any]]:
    """上游有序降级链（T-X2，采纳 cortiq tier 链思路）。

    主条目之外，取池内其他启用真实后端条目，按 (档位升序, 单价和升序) 排列
    ——便宜的先顶上；总链长 <= max_total。仅作首 token 前 failover 候选
    （D-P4），不做负载均衡（单写者模型）。
    """
    from gateway.model_pool import TIERS
    tier_rank = {t: i for i, t in enumerate(TIERS)}
    rest = [e for e in _pool_entries(pool)
            if (e.get("enabled") and e.get("id") != entry.get("id")
                and e.get("backend") not in ("mock",) and e.get("base_url")
                and e.get("model") != entry.get("model"))]
    rest.sort(key=lambda e: (tier_rank.get(e.get("tier"), 99),
                             float(e.get("price_in") or 0) + float(e.get("price_out") or 0)))
    return rest[:max(0, int(max_total) - 1)]


def _downgrade_entry(pool, entry: Dict[str, Any], mode: str) -> Optional[Dict[str, Any]]:
    """预算降档（T-X1）：在池内找恰好低一档/最低档的启用条目。

    档位序 TIERS = (local, budget, premium)：
    - optimize：降一档（premium->budget，budget->local）；
    - cheap：直落最低档 local；
    - 已在 local / 找不到该档启用条目 -> None（保持原条目，档位语义照常上报）。
    """
    from gateway.model_pool import TIERS
    tier = str(entry.get("tier") or "")
    if tier not in TIERS:
        return None
    rank = TIERS.index(tier)
    if mode == "cheap":
        target = 0
    elif mode == "optimize":
        target = rank - 1
    else:
        return None
    if target < 0:
        return None
    want = TIERS[target]
    for e in _pool_entries(pool):
        if (e.get("enabled") and e.get("id") != entry.get("id")
                and e.get("tier") == want
                and e.get("backend") not in ("mock",) and e.get("base_url")):
            return e
    return None


_semcache_instances: Dict[str, Any] = {}


def _get_semcache(cfg: ProxyConfig, ledger):
    """按 db_path 的进程内缓存单例（D-P9 单进程前提；不同库隔离）。"""
    inst = _semcache_instances.get(cfg.db_path)
    if inst is None:
        from gateway.proxy.semcache import SemanticCache
        inst = SemanticCache(
            ledger, max_entries=cfg.max_entries,
            sim_threshold=cfg.sim_threshold,
            promote_frequency=cfg.promote_frequency)
        _semcache_instances[cfg.db_path] = inst
    return inst


def reset_semcache_instances() -> None:
    """测试用：清空缓存单例。"""
    global _semcache_instances
    _semcache_instances = {}


_sense_runtimes: Dict[tuple, Any] = {}


def _get_sense_runtime(scfg):
    """sense 运行时按配置单例化（2026-09 优化，激活 T-X3 决策缓存）。

    键取 SenseConfig 的完整 repr：任何设置变更都会派生新键并重建运行时，
    旧实例留在注册表中随设置维度有界增长（单进程部署，D-P9）。
    返回 (sstore, grader)：Grader 持有的线性头/阈值/60s 决策缓存跨请求复用，
    免去每请求的 SenseStore DDL 与工件文件重读。
    """
    key = ("sense", repr(scfg))
    rt = _sense_runtimes.get(key)
    if rt is None:
        from gateway.sense.grader import Grader
        from gateway.sense.observer import get_observer
        from gateway.sense.promotion import PromotionTable
        from gateway.sense.store import SenseStore
        sstore = SenseStore.init_db(scfg.db_path)
        grader = Grader(scfg, sstore, get_observer(sstore),
                        promotion=PromotionTable(sstore))
        rt = (sstore, grader)
        _sense_runtimes[key] = rt
    return rt


def reset_sense_runtimes() -> None:
    """测试用：清空 sense 运行时单例。"""
    global _sense_runtimes
    _sense_runtimes = {}


def _cached_charge(body: dict, cfg: ProxyConfig, model: str) -> Dict[str, int]:
    """缓存命中计费：成本 0，按未命中口径对入/出估 token 收售价（§7）。"""
    from gateway.proxy.pricing import compute
    usage = {"in_miss": len(json.dumps(body.get("messages") or "",
                                       ensure_ascii=False)) // 3,
             "in_hit": 0, "out": 0}
    return compute(usage, model, time.time(), cfg)


async def _run_chat(body: dict, headers: Dict[str, str], ctx: Dict[str, Any],
                    cfg: ProxyConfig, ledger, pool,
                    request_id: Optional[str] = None):
    model = str(body.get("model") or "")
    entry = _resolve_entry(pool, model, cfg)
    if entry is None:
        return JSONResponse({"error": {"message": f"模型不可用: {model}",
                                       "type": "invalid_request_error"}},
                            status_code=400)
    if request_id is None:
        request_id = "px" + uuid.uuid4().hex[:12]
    ts = time.time()
    client_wants_usage = bool(body.get("stream_options", {}).get("include_usage")) \
        if isinstance(body.get("stream_options"), dict) else False
    is_stream = bool(body.get("stream"))

    # ---- 能力位硬过滤（T-X6）：请求需要 vision/tools/上下文而条目不满足时重定向 ----
    needs = _request_needs(body)
    if not _entry_meets(entry, needs):
        from gateway.model_pool import filter_by_capabilities
        cap_entries = filter_by_capabilities(
            [e for e in _pool_entries(pool)
             if e.get("enabled") and e.get("backend") not in ("mock",)
             and e.get("base_url")],
            need_vision=needs["vision"], need_tools=needs["tools"],
            min_context_tokens=needs["min_context_tokens"])
        if cap_entries:
            entry = cap_entries[0]
            model = str(entry.get("model") or model)
            body = {**body, "model": model}

    # ---- 预算四档（T-X1）：接近日上限渐进降档；>100% 仍由 try_hold 硬拒 ----
    budget_mode = "normal"
    try:
        est_probe = _estimate_hold_milli(body, cfg, model)
        budget_mode = await asyncio.to_thread(
            ledger.budget_mode, ctx["student_id"], est_probe, ts)
    except Exception:
        budget_mode = "normal"
    if budget_mode in ("optimize", "cheap"):
        down = _downgrade_entry(pool, entry, budget_mode)
        if down is not None:
            entry = down
            model = str(entry.get("model") or model)
            body = {**body, "model": model}

    # ---- 缓存分支（T-P6，§7 时序）：仅缓存准入（stop+单轮）查询 ----
    cacheable = False
    cache = None
    resolution = None
    try:
        if cfg.semcache_enabled:
            from gateway.proxy.normalizer import canonical_hash, is_cacheable
            from gateway.proxy.semcache import SemanticCache
            bucket_cfg = cfg.bucket(str(headers.get("x-campus-bucket") or "default"))
            cacheable = is_cacheable(body)
            if cacheable:
                norm_hash = canonical_hash(bucket_cfg.name, bucket_cfg.doc_version, body)
                cache = _get_semcache(cfg, ledger)
                norm_text = json.dumps(body.get("messages") or [], ensure_ascii=False,
                                       sort_keys=True)
                hit = cache.lookup(norm_hash, norm_text, bucket_cfg.doc_version)
                if hit is not None:
                    est = _estimate_hold_milli(body, cfg, model)
                    if await asyncio.to_thread(
                            ledger.try_hold, request_id, ctx["key_id"],
                            ctx["student_id"], model, bucket_cfg.name, est, ts):
                        br = _cached_charge(body, cfg, model)
                        await asyncio.to_thread(
                            ledger.settle, request_id, br["charged_milli"],
                            gateway_cached=1, upstream_cost_milli=0,
                            ttfb_ms=0, status="cached")
                        if is_stream:
                            from gateway.proxy.semcache import synth_sse_chunks
                            chunks = synth_sse_chunks(hit["answer"], model=model,
                                                      request_id=request_id)

                            async def replay():
                                for c in chunks:
                                    yield c
                            return StreamingResponse(
                                replay(), media_type="text/event-stream",
                                headers={"Cache-Control": "no-cache",
                                         "X-Cache": "HIT",
                                         "X-Request-Id": request_id,
                                         **_budget_headers(budget_mode)})
                        return JSONResponse({
                            "id": f"chatcmpl-{request_id}", "object": "chat.completion",
                            "created": int(time.time()), "model": model,
                            "choices": [{"index": 0,
                                         "message": {"role": "assistant",
                                                     "content": hit["answer"]},
                                         "finish_reason": "stop"}],
                            "usage": {"prompt_tokens": 0, "completion_tokens": 0,
                                      "total_tokens": 0},
                        }, headers={"X-Cache": "HIT", "X-Request-Id": request_id,
                                    **_budget_headers(budget_mode)})
                    raise BalanceError("余额或当日额度不足")
    except BalanceError:
        raise
    except Exception:
        cache = None  # 缓存层故障不影响主流程（降级直连上游）

    est = _estimate_hold_milli(body, cfg, model)
    if not await asyncio.to_thread(
            ledger.try_hold, request_id, ctx["key_id"], ctx["student_id"],
            model, "default", est, ts):
        raise BalanceError("余额或当日额度不足")

    t0 = time.perf_counter()
    sink: Dict[str, Any] = {}
    chain = _fallback_chain(pool, entry)
    try:
        if is_stream:
            return await _stream_response(body, entry, sink, headers, client_wants_usage,
                                          request_id, ctx, cfg, ledger, model, est, t0,
                                          cache=cache, cacheable=cacheable,
                                          budget_mode=budget_mode, chain=chain)
        return await _json_response(body, entry, sink, request_id, ctx, cfg,
                                    ledger, model, est, t0,
                                    cache=cache, cacheable=cacheable,
                                    headers=headers, budget_mode=budget_mode,
                                    chain=chain)
    except UpstreamAborted as e:
        # 流中失败：按已收 usage 结算（无 usage 按字符估算），不缓存（D-P4）
        usage = sink.get("usage") or _estimate_usage_from_sink(sink)
        br = compute(usage, model, ts, cfg)
        await asyncio.to_thread(
            ledger.settle, request_id, br["charged_milli"],
            in_miss_tok=usage.get("in_miss", 0), in_hit_tok=usage.get("in_hit", 0),
            out_tok=usage.get("out", 0), upstream_cost_milli=br["upstream_cost_milli"],
            total_ms=int((time.perf_counter() - t0) * 1000), status="aborted")
        return JSONResponse({"error": {"message": f"上游流中断: {e}",
                                       "type": "aborted"}}, status_code=502)
    except UpstreamError as e:
        # 首 token 前失败（failover 均失败）：全额退款
        await asyncio.to_thread(ledger.void, request_id, status="error")
        return JSONResponse({"error": {"message": str(e), "type": "upstream_error"}},
                            status_code=502)
    except Exception as e:  # noqa: BLE001
        await asyncio.to_thread(ledger.void, request_id, status="error")
        raise


def _estimate_usage_from_sink(sink: Dict[str, Any]) -> Dict[str, int]:
    text = sink.get("text", "")
    return {"in_miss": 0, "in_hit": 0, "out": len(text) // 4}


async def _stream_response(body, entry, sink, headers, client_wants_usage,
                           request_id, ctx, cfg, ledger, model, est, t0,
                           cache=None, cacheable=False, budget_mode: str = "normal",
                           chain: Optional[List[Dict[str, Any]]] = None):
    usage = {"in_miss": 0, "in_hit": 0, "out": 0}

    async def gen():
        collected = []
        chunk_id = f"chatcmpl-{request_id}"
        created = int(time.time())
        try:
            async for raw_bytes in upstream_stream(body, entry, sink,
                                                   chain if chain is not None else [entry]):
                line = raw_bytes.decode("utf-8").strip()
                if not line:
                    continue
                for sub in line.split("\n\n"):
                    if not sub:
                        continue
                    filtered = filter_usage_chunk(sub, client_wants_usage)
                    if filtered is None:
                        continue
                    # 同构补齐（§5.1）：确保 OpenAI chunk 形状（object/created/id/model）
                    out_line = filtered
                    if filtered.startswith("data:") and "[DONE]" not in filtered:
                        try:
                            obj = json.loads(filtered[5:].strip())
                            obj.setdefault("object", "chat.completion.chunk")
                            obj.setdefault("id", chunk_id)
                            obj.setdefault("created", created)
                            obj.setdefault("model", model)
                            delta = (obj.get("choices") or [{}])[0].get("delta") or {}
                            collected.append(str(delta.get("content") or ""))
                            out_line = "data: " + json.dumps(obj, ensure_ascii=False)
                        except (json.JSONDecodeError, IndexError):
                            pass
                    yield (out_line + "\n\n").encode("utf-8")
            sink["text"] = "".join(collected)
        finally:
            u = sink.get("usage") or {"in_miss": 0, "in_hit": 0, "out": 0}
            br = compute(u, model, time.time(), cfg)
            await asyncio.to_thread(
                ledger.settle, request_id, br["charged_milli"],
                in_miss_tok=u.get("in_miss", 0), in_hit_tok=u.get("in_hit", 0),
                out_tok=u.get("out", 0), upstream_cost_milli=br["upstream_cost_milli"],
                ttfb_ms=sink.get("ttfb_ms"),
                total_ms=int((time.perf_counter() - t0) * 1000),
                status="ok")

    return StreamingResponse(gen(), media_type="text/event-stream",
                             headers={"Cache-Control": "no-cache",
                                      "X-Request-Id": request_id,
                                      **_budget_headers(budget_mode)})


async def _json_response(body, entry, sink, request_id, ctx, cfg, ledger,
                         model, est, t0, cache=None, cacheable=False,
                         headers=None, budget_mode: str = "normal",
                         chain: Optional[List[Dict[str, Any]]] = None):
    parts = []
    async for raw_bytes in upstream_stream(body, entry, sink,
                                           chain if chain is not None else [entry]):
        line = raw_bytes.decode("utf-8").strip()
        if not line:
            continue
        for sub in line.split("\n\n"):
            if sub.startswith("data:"):
                payload = sub[5:].strip()
                if payload == "[DONE]":
                    continue
                try:
                    obj = json.loads(payload)
                except json.JSONDecodeError:
                    continue
                delta = (obj.get("choices") or [{}])[0].get("delta") or {}
                if delta.get("content"):
                    parts.append(str(delta["content"]))
                if obj.get("finish_reason") or (obj.get("choices") or [{}])[0].get("finish_reason"):
                    sink.setdefault("finish_reason",
                                    (obj.get("choices") or [{}])[0].get("finish_reason"))
    usage = sink.get("usage") or {"in_miss": 0, "in_hit": 0,
                                  "out": len("".join(parts)) // 4}
    br = compute(usage, model, time.time(), cfg)
    total_ms = int((time.perf_counter() - t0) * 1000)
    await asyncio.to_thread(
        ledger.settle, request_id, br["charged_milli"],
        in_miss_tok=usage.get("in_miss", 0), in_hit_tok=usage.get("in_hit", 0),
        out_tok=usage.get("out", 0), upstream_cost_milli=br["upstream_cost_milli"],
        ttfb_ms=sink.get("ttfb_ms"), total_ms=total_ms, status="ok")
    # 缓存准入（D-P5）：stop 且单轮且未命中来的 -> 写缓存
    if cache is not None and cacheable and sink.get("finish_reason", "stop") == "stop":
        try:
            from gateway.proxy.normalizer import canonical_hash
            bucket_cfg = cfg.bucket(str(headers.get("x-campus-bucket")
                                        or "default")) if headers else cfg.bucket("default")
            ckey = canonical_hash(bucket_cfg.name, bucket_cfg.doc_version, body)
            norm_text = json.dumps(body.get("messages") or [], ensure_ascii=False,
                                   sort_keys=True)
            cache.put(ckey, norm_text, "".join(parts), model,
                      doc_version=bucket_cfg.doc_version,
                      ttl_hours=bucket_cfg.ttl_hours)
        except Exception:
            pass
    fb = sink.get("upstream_fallback") or {}
    fb_headers: Dict[str, str] = {}
    if fb.get("used"):
        fb_headers = {"X-Upstream-Fallback": "1",
                      "X-Upstream-Original": str(fb.get("original") or ""),
                      "X-Upstream-Used": str(fb.get("used_model") or ""),
                      "X-Upstream-Reason": str(fb.get("reason") or "")[:200]}
    return JSONResponse({
        "id": f"chatcmpl-{request_id}",
        "object": "chat.completion",
        "created": int(time.time()),
        "model": model,
        "choices": [{"index": 0,
                     "message": {"role": "assistant",
                                 "content": "".join(parts)},
                     "finish_reason": sink.get("finish_reason") or "stop"}],
        "usage": {"prompt_tokens": usage.get("in_miss", 0) + usage.get("in_hit", 0),
                  "completion_tokens": usage.get("out", 0),
                  "total_tokens": usage.get("in_miss", 0) + usage.get("in_hit", 0)
                  + usage.get("out", 0)},
    }, headers={"X-Request-Id": request_id,
                **_budget_headers(budget_mode), **fb_headers})