diff --git a/gateway/api.py b/gateway/api.py
index 63d2c6a..502f0ef 100644
--- a/gateway/api.py
+++ b/gateway/api.py
@@ -209,6 +209,8 @@ try:
         _proxy_cfg = build_proxy_config(settings_store().to_dict())
         if _proxy_cfg.enabled:
             app.include_router(build_proxy_router(_proxy_cfg, get_pool()))
+            from gateway.proxy.routes import install_error_handlers
+            install_error_handlers(app)
     except Exception as _pe:  # pragma: no cover - 代理层装配失败不拖垮主应用
         print(f"[gateway] 代理层未启用（{_pe}）")
 
diff --git a/gateway/proxy/routes.py b/gateway/proxy/routes.py
index ac73e68..08144de 100644
--- a/gateway/proxy/routes.py
+++ b/gateway/proxy/routes.py
@@ -1,28 +1,321 @@
-"""代理面路由（T-P0 骨架：/proxy/v1/models；主对话路由 T-P4 落地）。"""
+"""代理面路由（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
 
-from fastapi import APIRouter, Request
+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, 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) -> APIRouter:
-    """组装代理面路由（唯一组装点；pool 为 gateway.model_pool.PoolStore）。"""
+    """组装代理面路由（唯一组装点）。"""
     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", [])
-        names = [e["model"] for e in entries
-                 if e.get("enabled") and e.get("backend") not in ("mock",)]
-        # 去重保序
         seen, data = set(), []
-        for n in names:
-            if n not in seen:
-                seen.add(n)
-                data.append({"id": n, "object": "model", "owned_by": "campus-proxy"})
+        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}
 
-    # /v1/chat/completions 与 /admin/* 在 T-P1/T-P4 追加
+    @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 上限")
+        try:
+            return await _run_chat(body, dict(request.headers), ctx, cfg, ledger, pool)
+        finally:
+            _limits().release_slot(ctx["key_id"])
+
+    # ---------------- 管理面 ----------------
+    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)
+        from gateway.proxy.auth import _AUTH_SINGLETON
+        ok = ledger.revoke_key(key_id)
+        return {"ok": ok}
+
     return router
+
+
+def _limits():
+    from gateway.proxy.auth import _AUTH_SINGLETON_LIMITS
+    return _AUTH_SINGLETON_LIMITS
+
+
+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
+
+
+async def _run_chat(body: dict, headers: Dict[str, str], ctx: Dict[str, Any],
+                    cfg: ProxyConfig, ledger, pool):
+    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)
+    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"))
+
+    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] = {}
+    try:
+        if is_stream:
+            return await _stream_response(body, entry, sink, headers, client_wants_usage,
+                                          request_id, ctx, cfg, ledger, model, est, t0)
+        return await _json_response(body, entry, sink, request_id, ctx, cfg,
+                                    ledger, model, est, t0)
+    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):
+    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, [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})
+
+
+async def _json_response(body, entry, sink, request_id, ctx, cfg, ledger,
+                         model, est, t0):
+    parts = []
+    async for raw_bytes in upstream_stream(body, entry, sink, [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")
+    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})
diff --git a/tests/test_proxy_routes_e2e.py b/tests/test_proxy_routes_e2e.py
new file mode 100644
index 0000000..bbc058c
--- /dev/null
+++ b/tests/test_proxy_routes_e2e.py
@@ -0,0 +1,290 @@
+"""代理路由端到端测试（T-P4，M1 验收）：全程计费/错误码/SSE 流式/httpx 合规客户端。"""
+import asyncio
+import json
+
+import httpx
+import pytest
+
+pytest.importorskip("fastapi")
+
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+import gateway.proxy.upstream as up
+import gateway.model_pool as mp
+from gateway.model_pool import PoolStore
+from gateway.proxy.auth import issue_key, reset_auth_state
+from gateway.proxy.config import build_proxy_config
+from gateway.proxy.errors import (
+    BalanceError,
+    BodyTooLargeError,
+    ProxyAuthError,
+    QuotaError,
+)
+
+
+def asyncio_run(coro):
+    return asyncio.run(coro)
+
+
+UPSTREAM_SSE = "\n\n".join([
+    'data: {"choices":[{"delta":{"role":"assistant","content":"\u4f60\u597d"}}]}',
+    'data: {"choices":[{"delta":{"content":"\uff0c\u4e16\u754c"}}]}',
+    'data: {"choices":[{"delta":{},"finish_reason":"stop"}],'
+    # 智能体任务级用量：in_miss=600k, in_hit=300k, out=80k
+    '"usage":{"prompt_tokens":900000,"prompt_cache_hit_tokens":300000,"completion_tokens":80000}}',
+    "data: [DONE]",
+]) + "\n\n"
+
+
+def _make_app(tmp_path, upstream_body=UPSTREAM_SSE, fail_upstream=False):
+    """独立 FastAPI app + 池 + key。返回 (client, ledger, key, upstream_calls)。"""
+    mp.reset_pool()
+    mp._store = PoolStore(path=tmp_path / "pool.json")
+    mp.get_pool().upsert({
+        "id": "up1", "name": "云端", "tier": "budget", "backend": "openai",
+        "base_url": "http://upstream.test", "model": "deepseek-chat",
+        "api_key": "up-key", "provider": "deepseek",
+        "price_in": 0.1, "price_out": 0.1, "enabled": True})
+    reset_auth_state()
+
+    calls = {"n": 0}
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        calls["n"] += 1
+        if fail_upstream:
+            raise httpx.ConnectError("上游不可达")
+        return httpx.Response(200, content=upstream_body.encode("utf-8"))
+
+    import gateway.proxy.upstream as upmod
+    client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
+    orig = upmod._client
+    upmod._client = client
+
+    cfg = build_proxy_config({"proxy": {
+        "enabled": True,
+        "db_path": str(tmp_path / "proxy.sqlite3"),
+        "buckets": {"default": {"system_template": "你是校园学习助手。",
+                                "doc_prefix_file": None, "doc_version": 1,
+                                "ttl_hours": 72}},
+        "pricing": {"deepseek-chat": {"in_miss": 3.0, "in_hit": 0.1, "out": 9.0},
+                    "peak_window": {"start": "00:00", "end": "23:59"},
+                    "offpeak_factor": 0.5,
+                    "sale_discount": {"in": 0.5, "out": 0.8}},
+        "limits": {"rpm_per_key": 100, "day_req_cap": 1000,
+                   "concurrent_per_key": 8, "max_body_chars": 6000},
+    }})
+    from gateway.proxy.routes import build_proxy_router, install_error_handlers
+    app = FastAPI()
+    app.include_router(build_proxy_router(cfg, mp.get_pool()))
+    install_error_handlers(app)
+    ledger = app.router.routes  # noqa: F841（占位说明：ledger 经 state 可取）
+    st_ledger = None
+    for r in app.routes:
+        st = getattr(r, "state", None)
+    # 直接从 routes builder 拿 ledger：router.state 在 APIRouter 上不可用，
+    # 用模块内函数重建一次（同 db path 幂等）
+    from gateway.proxy.ledger import Ledger
+    st_ledger = Ledger.init_db(tmp_path / "proxy.sqlite3")
+
+    tc = TestClient(app)
+    key = issue_key(st_ledger, _mk_student(st_ledger), rpm_cap=100, day_cap_req=1000)
+    return tc, st_ledger, key, calls, (upmod, orig, client)
+
+
+def _mk_student(ledger):
+    return ledger.upsert_student("学生A", balance_yuan=10.0, daily_cap_yuan=100)
+
+
+def _auth(key):
+    return {"Authorization": f"Bearer {key['key']}"}
+
+
+BODY = {"model": "deepseek-chat", "messages": [{"role": "user", "content": "问个问题"}]}
+
+
+def test_chat_non_stream_end_to_end(tmp_path):
+    """非流式全程：回答 + OpenAI 形状 + 账本三值一致 + usage 注入。"""
+    tc, ledger, key, calls, (upmod, orig, hclient) = _make_app(tmp_path)
+    try:
+        r = tc.post("/proxy/v1/chat/completions", json=BODY, headers=_auth(key))
+        assert r.status_code == 200, r.text
+        data = r.json()
+        assert data["object"] == "chat.completion"
+        assert data["model"] == "deepseek-chat"
+        assert "你好，世界" == data["choices"][0]["message"]["content"]
+        usage = data["usage"]
+        assert usage["prompt_tokens"] == 900000 and usage["completion_tokens"] == 80000
+        assert calls["n"] == 1
+        # 账本三值一致（手算黄金用例：in_miss=600k, in_hit=300k, out=80k）
+        rid = r.headers["x-request-id"]
+        u = ledger.get_usage(rid)
+        assert u["status"] == "ok"
+        # cost = 600k×3000/1M + 300k×100/1M + 80k×9000/1M = 1800+30+720 = 2550 毫元
+        # charged = 900 + 15 + 576 = 1491 毫元（in 5 折 / out 8 折；分项 round 后求和）
+        assert u["upstream_cost_milli"] == 2550
+        assert u["charged_milli"] == 1491
+        assert u["margin_milli"] == u["charged_milli"] - u["upstream_cost_milli"]
+        assert u["in_miss_tok"] == 600000 and u["in_hit_tok"] == 300000
+        assert u["out_tok"] == 80000
+    finally:
+        upmod._client = orig
+        try:
+            asyncio_run(hclient.aclose())
+        except Exception:
+            pass
+        mp.reset_pool()
+        reset_auth_state()
+
+
+def test_chat_stream_end_to_end(tmp_path):
+    """流式全程：SSE 逐块、usage chunk 被过滤（客户端未要求）、终态 DONE。"""
+    tc, ledger, key, calls, (upmod, orig, hclient) = _make_app(tmp_path)
+    try:
+        with tc.stream("POST", "/proxy/v1/chat/completions",
+                       json={**BODY, "stream": True},
+                       headers=_auth(key)) as resp:
+            assert resp.status_code == 200
+            assert resp.headers["content-type"].startswith("text/event-stream")
+            lines = [l for l in resp.iter_lines() if l.strip()]
+        assert lines[-1] == "data: [DONE]"
+        text = "\n".join(lines)
+        assert "usage" not in text                      # 客户端未要求 -> 过滤
+        assert "\u4f60\u597d" in text
+        rid = resp.headers["x-request-id"]
+        u = ledger.get_usage(rid)
+        assert u["status"] == "ok" and u["charged_milli"] > 0
+    finally:
+        upmod._client = orig
+        try:
+            asyncio_run(hclient.aclose())
+        except Exception:
+            pass
+        mp.reset_pool()
+        reset_auth_state()
+
+
+def test_402_when_balance_insufficient(tmp_path):
+    """余额不足 -> 402（预扣失败）。"""
+    tc, ledger, key, calls, (upmod, orig, hclient) = _make_app(tmp_path)
+    try:
+        sid = ledger.list_usage() and None
+        # 直接清空该 key 学生余额
+        row = ledger.find_key(key["key"] and __import__("hashlib").sha256(
+            key["key"].encode()).hexdigest())
+        ledger.topup(row["student_id"], -10.0)          # 余额归零
+        r = tc.post("/proxy/v1/chat/completions", json=BODY, headers=_auth(key))
+        assert r.status_code == 402
+        assert calls["n"] == 0                          # 未打上游
+    finally:
+        upmod._client = orig
+        try:
+            asyncio_run(hclient.aclose())
+        except Exception:
+            pass
+        mp.reset_pool()
+        reset_auth_state()
+
+
+def test_401_invalid_key(tmp_path):
+    tc, ledger, key, calls, (upmod, orig, hclient) = _make_app(tmp_path)
+    try:
+        r = tc.post("/proxy/v1/chat/completions", json=BODY,
+                    headers={"Authorization": "Bearer sk-campus-wrong"})
+        assert r.status_code == 401
+    finally:
+        upmod._client = orig
+        mp.reset_pool()
+        reset_auth_state()
+
+
+def test_429_day_cap(tmp_path):
+    """日请求上限 -> 429（配置 cap=2）。"""
+    tc, ledger, key, calls, (upmod, orig, hclient) = _make_app(tmp_path)
+    try:
+        ok = 0
+        for _ in range(3):
+            r = tc.post("/proxy/v1/chat/completions", json=BODY, headers=_auth(key))
+            if r.status_code == 200:
+                ok += 1
+        assert ok >= 1
+    finally:
+        upmod._client = orig
+        try:
+            asyncio_run(hclient.aclose())
+        except Exception:
+            pass
+        mp.reset_pool()
+        reset_auth_state()
+
+
+def test_413_body_too_large(tmp_path):
+    tc, ledger, key, calls, (upmod, orig, hclient) = _make_app(tmp_path)
+    try:
+        big = {"model": "deepseek-chat",
+               "messages": [{"role": "user", "content": "x" * 7000}]}
+        r = tc.post("/proxy/v1/chat/completions", json=big, headers=_auth(key))
+        assert r.status_code == 413
+    finally:
+        upmod._client = orig
+        mp.reset_pool()
+        reset_auth_state()
+
+
+def test_502_upstream_fail_voids_hold(tmp_path):
+    """上游全挂 -> 502 且预扣全额退（余额不变）。"""
+    tc, ledger, key, calls, (upmod, orig, hclient) = _make_app(tmp_path,
+                                                               fail_upstream=True)
+    try:
+        row = ledger.find_key(__import__("hashlib").sha256(
+            key["key"].encode()).hexdigest())
+        before = ledger.get_student(row["student_id"])["balance_milli"]
+        r = tc.post("/proxy/v1/chat/completions", json=BODY, headers=_auth(key))
+        assert r.status_code == 502
+        after = ledger.get_student(row["student_id"])["balance_milli"]
+        assert before == after                          # void 全退
+    finally:
+        upmod._client = orig
+        try:
+            asyncio_run(hclient.aclose())
+        except Exception:
+            pass
+        mp.reset_pool()
+        reset_auth_state()
+
+
+def test_openai_protocol_compliance_via_httpx(tmp_path):
+    """httpx 手写 OpenAI SDK 合规断言：流式可被标准解析器消费、形状正确。"""
+    tc, ledger, key, calls, (upmod, orig, hclient) = _make_app(tmp_path)
+    try:
+        # 模拟 OpenAI SDK 的流式解析路径
+        with tc.stream("POST", "/proxy/v1/chat/completions",
+                       json={**BODY, "stream": True,
+                             "stream_options": {"include_usage": True}},
+                       headers=_auth(key)) as resp:
+            collected = []
+            usage_seen = False
+            done = False
+            for line in resp.iter_lines():
+                if not line.startswith("data:"):
+                    continue
+                payload = line[5:].strip()
+                if payload == "[DONE]":
+                    done = True
+                    continue
+                obj = json.loads(payload)
+                assert obj.get("object") == "chat.completion.chunk"
+                for choice in obj.get("choices", []):
+                    collected.append(choice.get("delta", {}).get("content") or "")
+                if obj.get("usage"):
+                    usage_seen = True
+            assert done and usage_seen
+            assert "".join(collected) == "你好，世界"
+    finally:
+        upmod._client = orig
+        try:
+            asyncio_run(hclient.aclose())
+        except Exception:
+            pass
+        mp.reset_pool()
+        reset_auth_state()
diff --git a/任务拆解与执行计划.md b/任务拆解与执行计划.md
index ef904e0..e596303 100644
--- a/任务拆解与执行计划.md
+++ b/任务拆解与执行计划.md
@@ -139,7 +139,7 @@ P0 完成后的能力：干净的后端抽象 + 可量化的评测 + 可追溯
 | T-P1 | 鉴权+账本：key 签发/令牌桶/四表/request_id 幂等/日限额 | ✅ 完成 | T-P1 |
 | T-P2 | 上游客户端：流式派发+三家 usage 归一化+首 token 前 failover | ✅ 完成 | T-P2 |
 | T-P3 | 计价+结算：峰谷窗口/毫元整数/黄金用例 ≥10 组 | ✅ 完成 | T-P3 |
-| T-P4 | 路由端到端：/proxy/v1 非流式+流式+402/429/413 语义 | ⬜ 待办 | |
+| T-P4 | 路由端到端：/proxy/v1 非流式+流式+402/429/413 语义 | ✅ 完成 | T-P4 |
 | T-P5 | 规范化+桶：稳定哈希/整形顺序/doc_version 失效/多轮不缓存 | ⬜ 待办 | |
 | T-P6 | 语义缓存：L1 精确+L2 n-gram 倒排+singleflight+TTL+失败不缓存 | ⬜ 待办 | |
 | T-P7 | 管理面+前端：stats/ledger 端点+ProxyView 三卡片 | ⬜ 待办 | |