feat: M4 文风 + M5 生成/多provider/Skill + Kimi Code 订阅接入 + 本地联调修复

M4（文风）: style-auditor 双轨（提取指纹/漂移第四审）+ jobs 长任务框架(zombie reaper) + 回炉 refine + GET /style read-back。
M5（生成+扩展）: worldbuilder/character-gen（入库 continuity 409 gate + partition_writes 白名单 + schema→JSONB 形变）；
  网关多 provider 回退链/熔断/能力降级（Anthropic/Gemini 适配器）；Skill registry + 表权限沙箱 + 规则；
  前端 角色生成器/世界观/Codex/规则页/技能库/⌘K 命令面板。
K1（Kimi Code 订阅接入）: OAuth device-flow（kimi-code）+ 静态 Console key（kimi-code-key）两路径；
  coding 端点 KimiCLI 伪造头（实测 UA allow-list 门禁，缺则 403）+ JSON 模式结构化（thinking ⊥ tool_choice）。
本地联调修复: CORS 中间件；assemble 注入 premise+「写第N章」指令（修空 prompt 400）；
  GET /outline·/draft read-back + 大纲/工作台/审稿页重载；写页 client/server 常量边界 + notFound 健壮化；
  字数 toLocaleString locale 水合；审稿页终稿从已存草稿 seed（修 accept 422）。
门禁: backend ruff/mypy(157)/alembic 无漂移/pytest 451 · frontend lint/tsc/vitest/build。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

tests/test_k1_kimi_oauth_e2e.py

@@ -0,0 +1,510 @@
+"""K1 端到端：Kimi Code OAuth device-flow（device/token/refresh **全 mock**）→ 真 pg。
+
+证明 K1 闭环（DoD：device-flow 连接 → 加密 token 持久化(DB 真源,无明文) → 网关建出
+KimiCodeAdapter(coding base + 伪造头 + bearer) → 建网关时按需刷新更新存量 token → 断开清除）。
+**绝不联网**：device authorization / token / refresh 全经注入的 fake httpx 客户端，
+auth.kimi.com / api.kimi.com 一次都不碰。
+
+  用例 1 · 连接 device flow → token 加密持久化（DB 真源）
+    `POST .../oauth/start` → 202 `{job_id, user_code, verification_uri, ...}`。fake
+    `device_authorization` 返 device_code；fake `token` 端点先 `authorization_pending`
+    两次、再成功 `{access_token, refresh_token, expires_in}`。后台 `run_job` 轮询循环跑在
+    **真 e2e_sm session** 上（ASGITransport await background task；override
+    `get_session_factory`→e2e_sm、monkeypatch 模块级 `_default_http_client`→fake、
+    `asyncio.sleep`→no-op）。`GET /jobs/{id}` → done，result `{connected:true,
+    provider:"kimi-code"}` 且**不含任何 token 串**。DB 真源：`provider_credentials` 行
+    `auth_type="oauth"`、`oauth_enc` 非空、`api_key_enc` 为 None；解密 `oauth_enc` 回环
+    access/refresh token（证明加密持久化、非明文）。`GET .../oauth/status` →
+    `{connected:true, expires_at}` 无 token。
+
+  用例 2 · 网关建出 Kimi Code 适配器（正确 token + 伪造头 + coding base）
+    存 oauth 凭据 + `tier_routing` 行把某档位指向 `kimi-code:kimi-for-coding` →
+    `build_gateway_for_tier(...,"writer")` → 断言适配器是 `KimiCodeAdapter`，其
+    `AsyncOpenAI` 客户端 `base_url=https://api.kimi.com/coding/v1`、带伪造 `User-Agent`
+    + 全 `X-Msh-*` 默认头、`api_key`(=bearer) = 存量 access token。
+
+  用例 3 · 建网关时刷新（refresh-on-build）
+    存一个 `expires_at` 临近过期的 oauth 包；fake token 端点对 `grant_type=refresh_token`
+    返**新** token 组。`_build_provider_adapter(store,"kimi-code")`（经真 e2e_sm session）
+    → 断言 refresh 被调、适配器用**新** access token、`oauth_enc`(DB 真源)已更新为新包。
+
+  用例 4 · 断开（disconnect）
+    `POST .../oauth/disconnect` → 凭据行被清除（DB 真源：之后无 oauth 凭据）。
+
+零成本 & 确定性：fake httpx（device/token/refresh）+ 真 `SqlCredentialStore`/真 pg。无 pg → skip。
+
+坑（见 memory/gotchas）：
+- 后台 work 自建 http 走**模块级** `routers.kimi_oauth._default_http_client()`（非 dep）——
+  须 monkeypatch 它指到同一 scripted fake（端点 device-auth call[0] + work poll call[1..]
+  顺序共享）；端点 device-auth 走 `http` dep，再 override `_default_http_client` dep。
+- 真 pg 持久化：**不** monkeypatch `SqlCredentialStore`（让真 store 落库），override
+  `get_session_factory`→`e2e_sm`（同测试 engine/loop）使后台 work 用真 session 写库。
+- monkeypatch `routers.kimi_oauth.asyncio.sleep`→no-op（不真等 interval 秒）。
+- `_resolve_kimi_code_token` 刷新走 `project_deps.httpx.AsyncClient` + `project_deps.kimi_refresh`——
+  用字符串 target monkeypatch（mypy strict 不静态校验字符串属性）。
+- token 绝不进 job 结果/状态响应/start 响应——显式负向断言。
+- get_sessionmaker engine 绑首个 loop → 每 DB 测清缓存重建、结束 dispose（同 M3/M4/M5）。
+"""
+
+from __future__ import annotations
+
+import os
+from collections.abc import AsyncIterator
+from datetime import UTC, datetime, timedelta
+from typing import Any
+
+import httpx
+import pytest
+from asgi_lifespan import LifespanManager
+from cryptography.fernet import Fernet
+from sqlalchemy import delete, select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+from ww_db import get_sessionmaker
+from ww_db.models import Job, ProviderCredential, TierRouting
+from ww_llm_gateway.adapters.kimi_code import (
+    KIMI_CODE_BASE_URL,
+    KIMI_CODE_PROVIDER,
+    KIMI_CODE_USER_AGENT,
+    KimiCodeAdapter,
+)
+
+_ACCESS_TOKEN = "kimi-acc-e2e-not-real-12345"
+_REFRESH_TOKEN = "kimi-ref-e2e-not-real-67890"
+_NEW_ACCESS_TOKEN = "kimi-acc-e2e-refreshed-abcde"
+_NEW_REFRESH_TOKEN = "kimi-ref-e2e-refreshed-fghij"
+
+_OAUTH_BASE = "/settings/providers/kimi-code/oauth"
+_START_URL = f"{_OAUTH_BASE}/start"
+_DISCONNECT_URL = f"{_OAUTH_BASE}/disconnect"
+_STATUS_URL = f"{_OAUTH_BASE}/status"
+
+_REQUIRED_HEADER_KEYS = (
+    "User-Agent",
+    "X-Msh-Platform",
+    "X-Msh-Version",
+    "X-Msh-Device-Name",
+    "X-Msh-Device-Model",
+    "X-Msh-Os-Version",
+    "X-Msh-Device-Id",
+)
+
+
+class _FakeResponse:
+    """`HttpResponse` Protocol 最小实现（status_code + json()）。"""
+
+    def __init__(self, status_code: int, body: dict[str, Any]) -> None:
+        self.status_code = status_code
+        self._body = body
+
+    def json(self) -> Any:
+        return self._body
+
+
+class _ScriptedHttp:
+    """按调用序号返回预设响应（device auth → poll attempts）；记录每次调用。aclose 无操作。
+
+    端点的 device authorization 调用是 call[0]，后台 work 的 token 轮询是 call[1..]。
+    同一实例供两者顺序共享（须 monkeypatch 模块级 `_default_http_client` 指到它）。
+    """
+
+    def __init__(self, responses: list[_FakeResponse]) -> None:
+        self._responses = responses
+        self.calls: list[tuple[str, dict[str, str]]] = []
+
+    async def post(self, url: str, *, data: dict[str, str]) -> _FakeResponse:
+        idx = len(self.calls)
+        self.calls.append((url, data))
+        return self._responses[idx]
+
+    async def aclose(self) -> None:
+        return None
+
+
+def _device_auth_response() -> _FakeResponse:
+    return _FakeResponse(
+        200,
+        {
+            "device_code": "dev-e2e-1",
+            "user_code": "ABCD-1234",
+            "verification_uri": "https://auth.kimi.com/device",
+            "verification_uri_complete": "https://auth.kimi.com/device?code=ABCD-1234",
+            "expires_in": 600,
+            "interval": 5,
+        },
+    )
+
+
+def _pending_response() -> _FakeResponse:
+    return _FakeResponse(400, {"error": "authorization_pending"})
+
+
+def _token_success_response() -> _FakeResponse:
+    return _FakeResponse(
+        200,
+        {
+            "access_token": _ACCESS_TOKEN,
+            "refresh_token": _REFRESH_TOKEN,
+            "expires_in": 900,
+        },
+    )
+
+
+def _enc_key() -> str:
+    """设置 `CREDENTIAL_ENC_KEY` 为有效 Fernet 密钥 + 清 settings 缓存，返回该密钥。"""
+    key = Fernet.generate_key().decode()
+    os.environ["CREDENTIAL_ENC_KEY"] = key
+    from ww_config import get_settings
+
+    get_settings.cache_clear()
+    return key
+
+
+@pytest.fixture
+async def e2e_sm() -> AsyncIterator[async_sessionmaker[AsyncSession]]:
+    """真实 DB session 工厂；无 pg 时跳过（每测试清缓存重建 engine、结束 dispose）。"""
+    get_sessionmaker.cache_clear()
+    maker = get_sessionmaker()
+    try:
+        async with maker() as probe:
+            await probe.execute(select(1))
+    except Exception:
+        pytest.skip("postgres not reachable")
+    yield maker
+    await maker.kw["bind"].dispose()
+    get_sessionmaker.cache_clear()
+
+
+async def _cleanup_credentials(e2e_sm: async_sessionmaker[AsyncSession]) -> None:
+    """清理本测试族写入的 kimi-code 凭据 + kimi_oauth job 行（单用户 stub owner）。"""
+    from ww_api.services.credentials import STUB_OWNER_ID
+
+    async with e2e_sm() as cleanup:
+        await cleanup.execute(
+            delete(ProviderCredential).where(
+                ProviderCredential.owner_id == STUB_OWNER_ID,
+                ProviderCredential.provider == KIMI_CODE_PROVIDER,
+            )
+        )
+        await cleanup.execute(delete(Job).where(Job.kind == "kimi_oauth"))
+        await cleanup.execute(delete(TierRouting).where(TierRouting.project_id.is_(None)))
+        await cleanup.commit()
+
+
+async def _seed_oauth_credential(e2e_sm: async_sessionmaker[AsyncSession], blob: bytes) -> None:
+    """直接经真 store 在真 pg 写一行 oauth 凭据（用例 2/3 前置）。"""
+    from ww_api.services.credentials import STUB_OWNER_ID, SqlCredentialStore
+
+    async with e2e_sm() as session:
+        store = SqlCredentialStore(session)
+        await store.upsert_oauth_credential(STUB_OWNER_ID, KIMI_CODE_PROVIDER, blob)
+
+
+async def test_k1_connect_device_flow_persists_encrypted_token(
+    e2e_sm: async_sessionmaker[AsyncSession],
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """用例 1：device flow 连接 → 后台轮询(pending×2→成功) → 加密 token 真落 pg → status 已连接。
+
+    全程零联网（fake httpx），token 绝不进 job 结果/状态响应/start 响应（负向断言）。
+    """
+    import ww_api.routers.kimi_oauth as kimi_mod
+    from ww_api.main import create_app
+    from ww_api.services.credentials import (
+        AUTH_TYPE_OAUTH,
+        STUB_OWNER_ID,
+        SqlCredentialStore,
+    )
+    from ww_api.services.kimi_oauth import decrypt_oauth_bundle
+    from ww_api.services.project_deps import get_session_factory
+
+    enc_key = _enc_key()
+    # device auth(call0) → poll pending(call1) → poll pending(call2) → poll success(call3)。
+    http = _ScriptedHttp(
+        [
+            _device_auth_response(),
+            _pending_response(),
+            _pending_response(),
+            _token_success_response(),
+        ]
+    )
+
+    # 后台 work 不真等 interval 秒。
+    async def _no_sleep(_seconds: float) -> None:
+        return None
+
+    monkeypatch.setattr("ww_api.routers.kimi_oauth.asyncio.sleep", _no_sleep)
+    # 后台 work 自建 http 走模块级 `_default_http_client()`（非 dep）→ 指到同一 scripted fake。
+    monkeypatch.setattr(kimi_mod, "_default_http_client", lambda: http)
+
+    app = create_app()
+    # 端点 device-auth 走 http dep → 也指到同一 fake（保证顺序共享）。
+    app.dependency_overrides[kimi_mod._default_http_client] = lambda: http
+    # 后台 work 用 run_job 自建独立 session → 真 e2e_sm（同测试 engine/loop），真 store 落库。
+    app.dependency_overrides[get_session_factory] = lambda: e2e_sm
+
+    transport = httpx.ASGITransport(app=app)
+    try:
+        await _cleanup_credentials(e2e_sm)
+        async with LifespanManager(app):
+            async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+                # 1) 发起 device flow → 202 + user_code/verification_uri（ASGITransport 等后台）。
+                start_resp = await client.post(_START_URL)
+                assert start_resp.status_code == 202
+                start_body = start_resp.json()
+                assert start_body["user_code"] == "ABCD-1234"
+                assert start_body["verification_uri"] == "https://auth.kimi.com/device"
+                assert start_body["interval"] == 5
+                job_id = start_body["job_id"]
+                # start 响应绝不含 token。
+                assert _ACCESS_TOKEN not in start_resp.text
+                assert _REFRESH_TOKEN not in start_resp.text
+
+                # 2) 轮询 job → done，result 只 {connected, provider}，无 token。
+                job_resp = await client.get(f"/jobs/{job_id}")
+                assert job_resp.status_code == 200
+                job = job_resp.json()
+                assert job["status"] == "done"
+                assert job["kind"] == "kimi_oauth"
+                assert job["result"] == {"connected": True, "provider": KIMI_CODE_PROVIDER}
+                # job 整体（含 result）绝不泄露 token。
+                assert _ACCESS_TOKEN not in job_resp.text
+                assert _REFRESH_TOKEN not in job_resp.text
+
+                # 3) status → 已连接 + expires_at，无 token 本体。
+                status_resp = await client.get(_STATUS_URL)
+                assert status_resp.status_code == 200
+                status_body = status_resp.json()
+                assert status_body["connected"] is True
+                assert status_body["expires_at"] is not None
+                assert "access_token" not in status_body
+                assert "refresh_token" not in status_body
+                assert _ACCESS_TOKEN not in status_resp.text
+                assert _REFRESH_TOKEN not in status_resp.text
+
+        # 后台经历两次 pending 才成功（证明 pending 路径真走）：device + 3 token = 4 调用。
+        assert len(http.calls) == 4
+
+        # DB 真源断言：oauth 凭据行加密持久化（非明文），解密回环 token。
+        async with e2e_sm() as verify:
+            row = (
+                await verify.execute(
+                    select(ProviderCredential).where(
+                        ProviderCredential.owner_id == STUB_OWNER_ID,
+                        ProviderCredential.provider == KIMI_CODE_PROVIDER,
+                    )
+                )
+            ).scalar_one()
+            assert row.auth_type == AUTH_TYPE_OAUTH
+            assert row.oauth_enc is not None
+            assert row.api_key_enc is None
+            # 密文不是明文（token 不出现在原始字节里）。
+            assert _ACCESS_TOKEN.encode() not in row.oauth_enc
+            assert _REFRESH_TOKEN.encode() not in row.oauth_enc
+            # 解密回环：证明真的是这套 token 被加密持久化。
+            token = decrypt_oauth_bundle(row.oauth_enc, key=enc_key)
+            assert token.access_token == _ACCESS_TOKEN
+            assert token.refresh_token == _REFRESH_TOKEN
+
+            # 经真 store 读 status 也是已连接（store 层一致）。
+            store = SqlCredentialStore(verify)
+            cred = await store.get_credential(STUB_OWNER_ID, KIMI_CODE_PROVIDER)
+            assert cred is not None
+            assert cred.auth_type == AUTH_TYPE_OAUTH
+    finally:
+        await _cleanup_credentials(e2e_sm)
+
+
+async def test_k1_gateway_builds_kimi_code_adapter_with_headers_and_bearer(
+    e2e_sm: async_sessionmaker[AsyncSession],
+) -> None:
+    """用例 2：存 oauth + 档位路由指向 kimi-code → 建网关 → 适配器带 coding base/伪造头/bearer。"""
+    from ww_api.services.credentials import SqlCredentialStore, StoredRouting
+    from ww_api.services.kimi_oauth import TokenSet, encrypt_oauth_bundle
+    from ww_api.services.project_deps import build_gateway_for_tier
+
+    enc_key = _enc_key()
+    fresh = TokenSet(
+        access_token=_ACCESS_TOKEN,
+        refresh_token=_REFRESH_TOKEN,
+        expires_at=datetime.now(UTC) + timedelta(seconds=900),  # 远未过期 → 不刷新
+    )
+    blob = encrypt_oauth_bundle(fresh, key=enc_key)
+
+    try:
+        await _cleanup_credentials(e2e_sm)
+        await _seed_oauth_credential(e2e_sm, blob)
+        # 档位路由：writer → kimi-code:kimi-for-coding（DB 真源）。
+        async with e2e_sm() as seed:
+            store = SqlCredentialStore(seed)
+            await store.upsert_routing(
+                StoredRouting(
+                    tier="writer",
+                    provider=KIMI_CODE_PROVIDER,
+                    model="kimi-for-coding",
+                    fallback=[],
+                )
+            )
+
+        async with e2e_sm() as session:
+            store = SqlCredentialStore(session)
+            gateway = await build_gateway_for_tier(session, store, "writer")
+
+        adapter = gateway._adapters[KIMI_CODE_PROVIDER]
+        assert isinstance(adapter, KimiCodeAdapter)
+        client = adapter._client
+        # coding base。
+        assert str(client.base_url).rstrip("/") == KIMI_CODE_BASE_URL.rstrip("/")
+        # 伪造头：UA + 全 X-Msh-*。
+        for key in _REQUIRED_HEADER_KEYS:
+            assert key in client.default_headers
+        assert client.default_headers["User-Agent"] == KIMI_CODE_USER_AGENT
+        # bearer = 存量 access token（SDK 据 api_key 自动发 Authorization: Bearer）。
+        assert client.api_key == _ACCESS_TOKEN
+        # 未刷新 → DB oauth 凭据未变（仍是原 token）。
+        async with e2e_sm() as verify:
+            from ww_api.services.kimi_oauth import decrypt_oauth_bundle
+
+            row = (
+                await verify.execute(
+                    select(ProviderCredential).where(
+                        ProviderCredential.provider == KIMI_CODE_PROVIDER
+                    )
+                )
+            ).scalar_one()
+            assert row.oauth_enc is not None
+            assert decrypt_oauth_bundle(row.oauth_enc, key=enc_key).access_token == _ACCESS_TOKEN
+    finally:
+        await _cleanup_credentials(e2e_sm)
+
+
+async def test_k1_refresh_on_build_updates_stored_token(
+    e2e_sm: async_sessionmaker[AsyncSession],
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """用例 3：存量 token 临近过期 → 建适配器触发刷新 → 用新 token + DB oauth_enc 更新为新包。
+
+    刷新经 fake httpx + fake `kimi_refresh`（不联网）；新包真落 pg（DB 真源）。
+    """
+    from ww_api.services import project_deps
+    from ww_api.services.credentials import STUB_OWNER_ID, SqlCredentialStore
+    from ww_api.services.kimi_oauth import (
+        TokenSet,
+        decrypt_oauth_bundle,
+        encrypt_oauth_bundle,
+    )
+    from ww_api.services.project_deps import _build_provider_adapter
+
+    enc_key = _enc_key()
+    near = TokenSet(
+        access_token=_ACCESS_TOKEN,
+        refresh_token=_REFRESH_TOKEN,
+        expires_at=datetime.now(UTC) + timedelta(seconds=30),  # 临近过期 → 触发刷新
+    )
+    blob = encrypt_oauth_bundle(near, key=enc_key)
+
+    refreshed = TokenSet(
+        access_token=_NEW_ACCESS_TOKEN,
+        refresh_token=_NEW_REFRESH_TOKEN,
+        expires_at=datetime.now(UTC) + timedelta(seconds=900),
+    )
+    refresh_calls: list[str] = []
+
+    async def _fake_refresh(_http: Any, refresh_token: str) -> TokenSet:
+        # 证明用存量 refresh token 刷新。
+        refresh_calls.append(refresh_token)
+        return refreshed
+
+    class _FakeAsyncClient:
+        async def __aenter__(self) -> _FakeAsyncClient:
+            return self
+
+        async def __aexit__(self, *exc: object) -> None:
+            return None
+
+    monkeypatch.setattr(project_deps, "kimi_refresh", _fake_refresh)
+    monkeypatch.setattr(
+        "ww_api.services.project_deps.httpx.AsyncClient", lambda *a, **k: _FakeAsyncClient()
+    )
+
+    try:
+        await _cleanup_credentials(e2e_sm)
+        await _seed_oauth_credential(e2e_sm, blob)
+
+        async with e2e_sm() as session:
+            store = SqlCredentialStore(session)
+            adapter = await _build_provider_adapter(store, KIMI_CODE_PROVIDER)
+
+        # refresh 被调，且用存量 refresh token。
+        assert refresh_calls == [_REFRESH_TOKEN]
+        # 适配器用新 access token。
+        assert isinstance(adapter, KimiCodeAdapter)
+        assert adapter._client.api_key == _NEW_ACCESS_TOKEN
+
+        # DB 真源：oauth_enc 已更新为新包（下次建网关复用刷新结果）。
+        async with e2e_sm() as verify:
+            row = (
+                await verify.execute(
+                    select(ProviderCredential).where(
+                        ProviderCredential.owner_id == STUB_OWNER_ID,
+                        ProviderCredential.provider == KIMI_CODE_PROVIDER,
+                    )
+                )
+            ).scalar_one()
+            assert row.oauth_enc is not None
+            stored = decrypt_oauth_bundle(row.oauth_enc, key=enc_key)
+            assert stored.access_token == _NEW_ACCESS_TOKEN
+            assert stored.refresh_token == _NEW_REFRESH_TOKEN
+    finally:
+        await _cleanup_credentials(e2e_sm)
+
+
+async def test_k1_disconnect_clears_credential(
+    e2e_sm: async_sessionmaker[AsyncSession],
+) -> None:
+    """用例 4：disconnect → 凭据行被清除（DB 真源：之后无 oauth 凭据）。"""
+    from ww_api.main import create_app
+    from ww_api.services.credentials import STUB_OWNER_ID
+    from ww_api.services.kimi_oauth import TokenSet, encrypt_oauth_bundle
+
+    enc_key = _enc_key()
+    token = TokenSet(
+        access_token=_ACCESS_TOKEN,
+        refresh_token=_REFRESH_TOKEN,
+        expires_at=datetime.now(UTC) + timedelta(seconds=900),
+    )
+    blob = encrypt_oauth_bundle(token, key=enc_key)
+
+    app = create_app()
+    transport = httpx.ASGITransport(app=app)
+    try:
+        await _cleanup_credentials(e2e_sm)
+        await _seed_oauth_credential(e2e_sm, blob)
+        # 前置：凭据确实在库。
+        async with e2e_sm() as pre:
+            assert (
+                await pre.execute(
+                    select(ProviderCredential).where(
+                        ProviderCredential.owner_id == STUB_OWNER_ID,
+                        ProviderCredential.provider == KIMI_CODE_PROVIDER,
+                    )
+                )
+            ).scalar_one_or_none() is not None
+
+        async with LifespanManager(app):
+            async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+                resp = await client.post(_DISCONNECT_URL)
+                assert resp.status_code == 200
+                assert resp.json()["disconnected"] is True
+
+        # DB 真源：凭据行已清除。
+        async with e2e_sm() as verify:
+            assert (
+                await verify.execute(
+                    select(ProviderCredential).where(
+                        ProviderCredential.owner_id == STUB_OWNER_ID,
+                        ProviderCredential.provider == KIMI_CODE_PROVIDER,
+                    )
+                )
+            ).scalar_one_or_none() is None
+    finally:
+        await _cleanup_credentials(e2e_sm)

tests/test_m4_e2e.py

@@ -0,0 +1,569 @@
+"""M4 端到端：学文风(指纹+证据,版本化) + 漂移打分(第四审) + 回炉（真实 DB，零 token）。
+
+证明 M4 闭环（DoD：学文风指纹+证据；漂移打分作第四审真持久化；回炉返新旧 diff）：
+  用例 1 · 学文风闭环（jobs 异步长任务，M4-c）
+    `POST /projects` → `POST .../style`(samples) → **202 `{job_id}`**（写一行 job 返回）；
+    BackgroundTask 经 `run_job` 自建独立 session 跑 `run_style_extraction`(analyst 假适配器
+    产 `StyleFingerprintResult`)→拆指纹→`SqlStyleFingerprintWriteRepo.append`(真落 pg)。
+    ASGITransport 下 `await client.post` 等 background task 跑完（同 M3 时序坑）；override
+    `get_session_factory`→`e2e_sm`(真 sessionmaker/同测试 engine/loop)，并 monkeypatch
+    `style.build_gateway_for_tier`→真 Gateway 包假适配器(用 background task 的真 session 记账)。
+    → `GET /jobs/{id}` status=done(result.version/dims_count) → 断言 `style_fingerprint` 真有行
+    (dimensions_json/evidence_json/version)。再 `mode="update"` 学一次 → version+1。
+  用例 2 · 写章第四审（drift scoring）
+    学文风后 `assemble` 的 `stable_core` 含指纹 → 第四审能在 review_context 拿指纹对照。
+    `POST .../chapters/N/review`(SSE) → 断言 `section{name:"style"}` 帧 + `style{score,segments}`
+    帧真发；`GET .../reviews` 留痕行 `style`(dict) 已填；DB 真源 `chapter_reviews.style` 真落库
+    (score + segments)。四审齐（continuity/foreshadow/pace/style）无 error。
+  用例 3 · 回炉（refine，非持久 / 不变量 #3）
+    `POST .../chapters/N/refine` → 200 `{original, refined}`；断言 `refined != original`
+    （writer 假适配器据 `req.output_schema is None` 返改写串）+ 不写库。
+
+确定性 & 零成本（同 M1/M2/M3）：真实 `Gateway` + 多档位假适配器（据 `req.output_schema`
+分支返回固定 `parsed`/`text`，绝不联网）+ 真实 `SqlAlchemyLedgerSink`。无 pg → skip。
+
+坑（见 memory/gotchas）：
+- 学文风 BackgroundTask 在 `run_job` 自建独立 session 上**自造网关**(`build_gateway_for_tier`,
+  从凭据建真 OpenAI 适配器) → E2E 无真凭据，须 monkeypatch `style.build_gateway_for_tier`
+  返真 `Gateway`(包假适配器, ledger 绑入参 session)；写侧 `SqlStyleFingerprintWriteRepo` 保持
+  真实 → 指纹真落 pg。`run_job` 默认 `repo_factory`→真 `SqlJobRepo`（落真 pg job 行），不动。
+- `get_session_factory` 必 override → `e2e_sm`（真 sessionmaker，同测试 engine/loop），否则
+  默认 `get_sessionmaker()` 另建 engine 绑别的 loop（跨 loop 坑）。
+- `get_sessionmaker` engine 绑首个 loop → 每 DB 测试清缓存重建、结束 dispose（同 M3）。
+- ASGITransport 默认 `raise_app_exceptions=True`；SSE 解析 `event:`/`data:` 块；snake_case。
+"""
+
+from __future__ import annotations
+
+import json
+import uuid
+from collections.abc import AsyncIterator, Callable
+from typing import Annotated
+
+import httpx
+import pytest
+from asgi_lifespan import LifespanManager
+from fastapi import Depends
+from sqlalchemy import delete, select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+from ww_agents import (
+    ContinuityReview,
+    ForeshadowReview,
+    ForeshadowSuggestion,
+    PaceIssue,
+    PaceReview,
+    StyleDimension,
+    StyleDriftReview,
+    StyleDriftSegment,
+    StyleFingerprintResult,
+)
+from ww_api.services.digest_extraction import ChapterDigestFacts
+from ww_db import get_session, get_sessionmaker
+from ww_db.models import (
+    Chapter,
+    ChapterDigest,
+    ChapterReview,
+    Foreshadow,
+    Job,
+    Outline,
+    Project,
+    StyleFingerprint,
+    UsageLedger,
+)
+from ww_llm_gateway import (
+    Gateway,
+    SqlAlchemyLedgerSink,
+    resolve_route,
+)
+from ww_llm_gateway.adapters.base import (
+    Capabilities,
+    ProviderResult,
+    ProviderUsage,
+    StreamChunk,
+)
+from ww_llm_gateway.types import LlmRequest
+
+# 三档位（writer/analyst/light）默认都路由到 deepseek（config.tier_defaults）。
+_PROVIDER = "deepseek"
+
+# 确定性流式写章 token。
+_TOKENS = ["第", "一", "章", "：", "文风漂移之时。"]
+
+# 学文风假产物：两维指纹，每维带原文证据（StyleFingerprintResult 提取轨）。
+_FINGERPRINT = StyleFingerprintResult(
+    dimensions=[
+        StyleDimension(name="句长节奏", value="短句为主", evidence=["他来了。", "她走了。"]),
+        StyleDimension(name="叙事人称", value="第三人称", evidence=["他看着远方的山。"]),
+    ]
+)
+
+# 第四审（文风漂移）假产物：整体相似度 + 一个低相似段。
+_STYLE_DRIFT = StyleDriftReview(
+    score=72,
+    segments=[StyleDriftSegment(idx=2, score=40, label="机翻腔")],
+)
+
+# 三审固定产物（与 M3 一致，让四审图跑通）。
+_FORESHADOW_REVIEW = ForeshadowReview(
+    planted=[ForeshadowSuggestion(code="F9", title="新埋：山门禁制", where="第2段")],
+    resolved=[],
+)
+_PACE_REVIEW = PaceReview(
+    water=[PaceIssue(where="第3段", reason="信息密度低")],
+    hook=True,
+    beat_map=[1, 3, 5, 2],
+)
+
+# 回炉改写标记（writer 纯文本返回）：保证 refined != original。
+_REFINE_PREFIX = "【已回炉】"
+
+# 各档位假用量（喂记账；证明各档位调用各自落库）。
+_USAGE = {
+    "writer": ProviderUsage(input_tokens=11, output_tokens=5),
+    "analyst": ProviderUsage(input_tokens=23, output_tokens=7),
+    "light": ProviderUsage(input_tokens=17, output_tokens=9),
+}
+
+
+def _req_text(req: LlmRequest) -> str:
+    """把 `LlmRequest.input`（str | list[Block]）压成纯文本（供回炉改写引用原文）。"""
+    if isinstance(req.input, str):
+        return req.input
+    return "\n".join(block.text for block in req.input)
+
+
+class _FakeMultiTierAdapter:
+    """实现 `ProviderAdapter` Protocol：按 `req.output_schema` 分支返回 parsed/text，绝不联网。
+
+    - `complete()`：据 `req.output_schema` 路由：
+      `StyleFingerprintResult`→学文风提取（analyst）；四审 schema
+      （`ContinuityReview`/`ForeshadowReview`/`PaceReview`/`StyleDriftReview`）→各审；
+      `output_schema is None`→**回炉纯文本**（writer，返改写串，保证 refined≠original）；
+      其余视作 digest 提炼（light）。
+    - `stream()`：吐固定写章 token（writer/draft），末尾喂 writer 用量。
+    每路带 `ProviderUsage` → 网关据此各落一条 usage_ledger。
+    """
+
+    provider = _PROVIDER
+
+    def capabilities(self) -> Capabilities:
+        return Capabilities(structured_output=True)
+
+    async def complete(self, req: LlmRequest, model: str) -> ProviderResult:
+        schema = req.output_schema
+        if schema is None:
+            # 回炉：writer 纯文本重写选中段（保证 refined != original）。
+            return ProviderResult(
+                text=_REFINE_PREFIX + _req_text(req),
+                parsed=None,
+                usage=_USAGE["writer"],
+            )
+        if schema is StyleFingerprintResult:
+            return ProviderResult(
+                text=_FINGERPRINT.model_dump_json(),
+                parsed=_FINGERPRINT,
+                usage=_USAGE["analyst"],
+            )
+        if schema is StyleDriftReview:
+            return ProviderResult(
+                text=_STYLE_DRIFT.model_dump_json(),
+                parsed=_STYLE_DRIFT,
+                usage=_USAGE["light"],
+            )
+        if schema is ContinuityReview:
+            review = ContinuityReview(conflicts=[])  # 零冲突
+            return ProviderResult(
+                text=review.model_dump_json(), parsed=review, usage=_USAGE["analyst"]
+            )
+        if schema is ForeshadowReview:
+            return ProviderResult(
+                text=_FORESHADOW_REVIEW.model_dump_json(),
+                parsed=_FORESHADOW_REVIEW,
+                usage=_USAGE["analyst"],
+            )
+        if schema is PaceReview:
+            return ProviderResult(
+                text=_PACE_REVIEW.model_dump_json(), parsed=_PACE_REVIEW, usage=_USAGE["light"]
+            )
+        # 否则视为 digest 提炼（ChapterDigestFacts）。
+        facts = ChapterDigestFacts(summary="终稿摘要", events=["事件"], locations=["地点"])
+        return ProviderResult(text=facts.model_dump_json(), parsed=facts, usage=_USAGE["light"])
+
+    async def stream(self, req: LlmRequest, model: str) -> AsyncIterator[StreamChunk]:
+        for token in _TOKENS:
+            yield StreamChunk(text=token)
+        yield StreamChunk(usage=_USAGE["writer"])
+
+
+@pytest.fixture
+async def e2e_sm() -> AsyncIterator[async_sessionmaker[AsyncSession]]:
+    """真实 DB session 工厂；无 pg 时跳过（每测试清缓存重建 engine、结束 dispose）。"""
+    get_sessionmaker.cache_clear()
+    maker = get_sessionmaker()
+    try:
+        async with maker() as probe:
+            await probe.execute(select(1))
+    except Exception:
+        pytest.skip("postgres not reachable")
+    yield maker
+    await maker.kw["bind"].dispose()
+    get_sessionmaker.cache_clear()
+
+
+def _parse_sse(raw: str) -> list[tuple[str, str]]:
+    """把 text/event-stream 原文解析为 `(event, data)` 帧列表。"""
+    frames: list[tuple[str, str]] = []
+    event: str | None = None
+    data: str | None = None
+    for line in raw.splitlines():
+        if line.startswith("event:"):
+            event = line[len("event:") :].strip()
+        elif line.startswith("data:"):
+            data = line[len("data:") :].strip()
+        elif line == "":
+            if event is not None and data is not None:
+                frames.append((event, data))
+            event, data = None, None
+    if event is not None and data is not None:
+        frames.append((event, data))
+    return frames
+
+
+def _gateway_override_factory(
+    adapter: _FakeMultiTierAdapter,
+) -> Callable[[AsyncSession], Gateway]:
+    """构造依赖覆盖：真实 Gateway + 给定假适配器 + 真实 ledger（请求 session）。"""
+
+    def _override(session: Annotated[AsyncSession, Depends(get_session)]) -> Gateway:
+        return Gateway(
+            adapters={_PROVIDER: adapter},
+            ledger=SqlAlchemyLedgerSink(session),
+            resolver=resolve_route,
+        )
+
+    return _override
+
+
+def _patched_build_gateway(
+    adapter: _FakeMultiTierAdapter,
+) -> Callable[..., object]:
+    """构造 monkeypatch 替身：`style.build_gateway_for_tier`→真 Gateway 包假适配器。
+
+    学文风 BackgroundTask 在 `run_job` 自建的独立 session 上自造网关（从凭据建真 OpenAI
+    适配器）——E2E 无真凭据，故替换成返真 `Gateway`(包假适配器, ledger 绑 background task
+    的真 session)。写侧 `SqlStyleFingerprintWriteRepo` + `run_job` 默认 `SqlJobRepo` 保持
+    真实 → 指纹/job 真落 pg。
+    """
+
+    async def _build(session: AsyncSession, store: object, tier: str) -> Gateway:
+        return Gateway(
+            adapters={_PROVIDER: adapter},
+            ledger=SqlAlchemyLedgerSink(session),
+            resolver=resolve_route,
+        )
+
+    return _build
+
+
+async def _cleanup(e2e_sm: async_sessionmaker[AsyncSession], project_uuid: uuid.UUID) -> None:
+    """按 FK 顺序清理（多数子表经 FK CASCADE；显式删可独立行）。"""
+    async with e2e_sm() as cleanup:
+        await cleanup.execute(delete(UsageLedger).where(UsageLedger.project_id == project_uuid))
+        await cleanup.execute(delete(Job).where(Job.project_id == project_uuid))
+        await cleanup.execute(
+            delete(StyleFingerprint).where(StyleFingerprint.project_id == project_uuid)
+        )
+        await cleanup.execute(delete(ChapterDigest).where(ChapterDigest.project_id == project_uuid))
+        await cleanup.execute(delete(ChapterReview).where(ChapterReview.project_id == project_uuid))
+        await cleanup.execute(delete(Outline).where(Outline.project_id == project_uuid))
+        await cleanup.execute(delete(Foreshadow).where(Foreshadow.project_id == project_uuid))
+        await cleanup.execute(delete(Chapter).where(Chapter.project_id == project_uuid))
+        await cleanup.execute(delete(Project).where(Project.id == project_uuid))
+        await cleanup.commit()
+
+
+async def test_m4_learn_style_job_done_and_fingerprint_persisted(
+    e2e_sm: async_sessionmaker[AsyncSession],
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """用例 1：学文风闭环 → 202 job → 后台跑完 → job done → 指纹真落 pg（含 update version+1）。"""
+    import ww_api.routers.style as style_mod
+    from ww_api.main import create_app
+    from ww_api.services.project_deps import (
+        get_session_factory,
+        get_style_extract_gateway,
+    )
+
+    adapter = _FakeMultiTierAdapter()
+    override = _gateway_override_factory(adapter)
+
+    app = create_app()
+    # 凭据探测（请求阶段）→ 注假网关，绕过无凭据 503。
+    app.dependency_overrides[get_style_extract_gateway] = override
+    # BackgroundTask 独立 session 工厂 → 真 sessionmaker（同测试 engine/loop）。
+    app.dependency_overrides[get_session_factory] = lambda: e2e_sm
+    # BackgroundTask 自造网关 → 替成真 Gateway 包假适配器（写侧 repo + SqlJobRepo 保持真）。
+    monkeypatch.setattr(style_mod, "build_gateway_for_tier", _patched_build_gateway(adapter))
+
+    transport = httpx.ASGITransport(app=app)
+    project_uuid: uuid.UUID | None = None
+    try:
+        async with LifespanManager(app):
+            async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+                create_resp = await client.post(
+                    "/projects", json={"title": "M4 学文风验证作品", "genre": "玄幻"}
+                )
+                assert create_resp.status_code == 201
+                project_id = create_resp.json()["id"]
+                project_uuid = uuid.UUID(project_id)
+
+                # 1) 学文风（首学）→ 202 {job_id}（ASGITransport 等 background task 跑完）。
+                learn_resp = await client.post(
+                    f"/projects/{project_id}/style",
+                    json={"samples": ["他来了。她走了。", "他看着远方的山。"], "mode": "create"},
+                )
+                assert learn_resp.status_code == 202
+                job_id = learn_resp.json()["job_id"]
+
+                # 2) 轮询 job → done，result 含 version/dims_count。
+                job_resp = await client.get(f"/jobs/{job_id}")
+                assert job_resp.status_code == 200
+                job = job_resp.json()
+                assert job["status"] == "done"
+                assert job["kind"] == "style_learn"
+                assert job["result"]["version"] == 1
+                assert job["result"]["dims_count"] == 2
+
+                # 3) GET /style → 最新指纹（完整维度 + 证据 + 版本）。
+                style_resp = await client.get(f"/projects/{project_id}/style")
+                assert style_resp.status_code == 200
+                fp = style_resp.json()
+                assert fp["version"] == 1
+                assert fp["dimensions"]["句长节奏"] == "短句为主"
+                assert fp["evidence"]["句长节奏"] == ["他来了。", "她走了。"]
+
+                # 4) mode="update" 再学一次 → version+1。
+                update_resp = await client.post(
+                    f"/projects/{project_id}/style",
+                    json={"samples": ["新样本正文。"], "mode": "update"},
+                )
+                assert update_resp.status_code == 202
+                update_job = (await client.get(f"/jobs/{update_resp.json()['job_id']}")).json()
+                assert update_job["status"] == "done"
+                assert update_job["result"]["version"] == 2
+
+        # DB 真源断言。
+        assert project_uuid is not None
+        async with e2e_sm() as verify:
+            fp_rows = (
+                (
+                    await verify.execute(
+                        select(StyleFingerprint)
+                        .where(StyleFingerprint.project_id == project_uuid)
+                        .order_by(StyleFingerprint.version)
+                    )
+                )
+                .scalars()
+                .all()
+            )
+            # 两次学文风 → 两行（版本化 INSERT，不覆盖历史）。
+            assert [r.version for r in fp_rows] == [1, 2]
+            first = fp_rows[0]
+            assert first.dimensions_json == {"句长节奏": "短句为主", "叙事人称": "第三人称"}
+            assert first.evidence_json["句长节奏"] == ["他来了。", "她走了。"]
+
+            # job 真落 pg：两行 done。
+            job_rows = (
+                (await verify.execute(select(Job).where(Job.project_id == project_uuid)))
+                .scalars()
+                .all()
+            )
+            assert len(job_rows) == 2
+            assert all(r.status == "done" and r.kind == "style_learn" for r in job_rows)
+    finally:
+        if project_uuid is not None:
+            await _cleanup(e2e_sm, project_uuid)
+
+
+async def test_m4_review_fourth_style_audit_sse_and_trace(
+    e2e_sm: async_sessionmaker[AsyncSession],
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """用例 2：写章第四审（drift）→ SSE section{style} + style{score,segments} + style 列真填。
+
+    先学文风 → `assemble` 的 stable_core 含指纹 → 第四审能在 review_context 拿指纹对照打分。
+    四审齐（continuity/foreshadow/pace/style）无 error。
+    """
+    import ww_api.routers.style as style_mod
+    from ww_api.main import create_app
+    from ww_api.services.project_deps import (
+        get_review_gateway,
+        get_session_factory,
+        get_style_extract_gateway,
+        get_writer_gateway,
+    )
+
+    adapter = _FakeMultiTierAdapter()
+    override = _gateway_override_factory(adapter)
+
+    app = create_app()
+    app.dependency_overrides[get_writer_gateway] = override
+    app.dependency_overrides[get_review_gateway] = override
+    app.dependency_overrides[get_style_extract_gateway] = override
+    app.dependency_overrides[get_session_factory] = lambda: e2e_sm
+    monkeypatch.setattr(style_mod, "build_gateway_for_tier", _patched_build_gateway(adapter))
+
+    transport = httpx.ASGITransport(app=app)
+    project_uuid: uuid.UUID | None = None
+    try:
+        async with LifespanManager(app):
+            async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+                create_resp = await client.post("/projects", json={"title": "M4 第四审验证作品"})
+                assert create_resp.status_code == 201
+                project_id = create_resp.json()["id"]
+                project_uuid = uuid.UUID(project_id)
+
+                # 1) 学文风 → 指纹入库（供 assemble 注入第四审做对照）。
+                learn_resp = await client.post(
+                    f"/projects/{project_id}/style",
+                    json={"samples": ["他来了。她走了。"], "mode": "create"},
+                )
+                assert learn_resp.status_code == 202
+                assert (await client.get(f"/jobs/{learn_resp.json()['job_id']}")).json()[
+                    "status"
+                ] == "done"
+
+                # 2) 写第1章草稿 + 自动保存（供续审）。
+                draft_resp = await client.post(f"/projects/{project_id}/chapters/1/draft")
+                draft_text = "".join(
+                    json.loads(d)["text"]
+                    for (ev, d) in _parse_sse(draft_resp.text)
+                    if ev == "token"
+                )
+                await client.put(
+                    f"/projects/{project_id}/chapters/1/draft",
+                    json={"text": draft_text},
+                )
+
+                # 3) 续审（SSE）：四审齐 + section{style} + style{score,segments}，无 error。
+                review_resp = await client.post(
+                    f"/projects/{project_id}/chapters/1/review",
+                    json={"draft": draft_text},
+                )
+                assert review_resp.status_code == 200
+                assert review_resp.headers["content-type"].startswith("text/event-stream")
+                rframes = _parse_sse(review_resp.text)
+                assert [d for (ev, d) in rframes if ev == "error"] == []
+                section_names = {
+                    json.loads(d).get("name") for (ev, d) in rframes if ev == "section"
+                }
+                assert {"continuity", "foreshadow", "pace", "style"} <= section_names
+                style_events = [json.loads(d) for (ev, d) in rframes if ev == "style"]
+                assert len(style_events) == 1
+                style_ev = style_events[0]
+                assert style_ev["score"] == 72
+                assert style_ev["segments"][0]["idx"] == 2
+                assert style_ev["segments"][0]["score"] == 40
+                assert style_ev["segments"][0]["label"] == "机翻腔"
+
+                # 4) GET 历史：留痕行 style(dict) 已填。
+                hist_resp = await client.get(f"/projects/{project_id}/chapters/1/reviews")
+                assert hist_resp.status_code == 200
+                latest = hist_resp.json()["reviews"][0]
+                assert isinstance(latest["style"], dict)
+                assert latest["style"]["score"] == 72
+
+        # DB 真源断言：chapter_reviews.style 列真填。
+        assert project_uuid is not None
+        async with e2e_sm() as verify:
+            review_rows = (
+                (
+                    await verify.execute(
+                        select(ChapterReview).where(ChapterReview.project_id == project_uuid)
+                    )
+                )
+                .scalars()
+                .all()
+            )
+            assert len(review_rows) == 1
+            row = review_rows[0]
+            assert isinstance(row.style, dict)
+            assert row.style["score"] == 72
+            assert row.style["segments"][0]["idx"] == 2
+            assert row.style["segments"][0]["label"] == "机翻腔"
+    finally:
+        if project_uuid is not None:
+            await _cleanup(e2e_sm, project_uuid)
+
+
+async def test_m4_refine_returns_diff_without_persisting(
+    e2e_sm: async_sessionmaker[AsyncSession],
+) -> None:
+    """用例 3：回炉 → 200 {original, refined}；refined != original；不写库（不变量 #3）。"""
+    from ww_api.main import create_app
+    from ww_api.services.project_deps import get_refine_gateway
+
+    adapter = _FakeMultiTierAdapter()
+    override = _gateway_override_factory(adapter)
+
+    app = create_app()
+    app.dependency_overrides[get_refine_gateway] = override
+
+    transport = httpx.ASGITransport(app=app)
+    project_uuid: uuid.UUID | None = None
+    try:
+        async with LifespanManager(app):
+            async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+                create_resp = await client.post("/projects", json={"title": "M4 回炉验证作品"})
+                assert create_resp.status_code == 201
+                project_id = create_resp.json()["id"]
+                project_uuid = uuid.UUID(project_id)
+
+                segment = "他緩緩走進房間。"
+                refine_resp = await client.post(
+                    f"/projects/{project_id}/chapters/1/refine",
+                    json={"segment": segment, "instruction": "去掉机翻腔"},
+                )
+                assert refine_resp.status_code == 200
+                body = refine_resp.json()
+                assert body["original"] == segment
+                # writer 假适配器返改写串（前缀），保证 refined != original。
+                assert body["refined"] != body["original"]
+                assert body["refined"].startswith(_REFINE_PREFIX)
+                # 指令进了输入文本（writer 据 output_schema is None 走纯文本路径）。
+                assert "去掉机翻腔" in body["refined"]
+
+        # DB 真源断言：回炉不写库 → 无 chapter/chapter_review 行（不变量 #3）。
+        assert project_uuid is not None
+        async with e2e_sm() as verify:
+            chapter_rows = (
+                (await verify.execute(select(Chapter).where(Chapter.project_id == project_uuid)))
+                .scalars()
+                .all()
+            )
+            assert chapter_rows == []
+            review_rows = (
+                (
+                    await verify.execute(
+                        select(ChapterReview).where(ChapterReview.project_id == project_uuid)
+                    )
+                )
+                .scalars()
+                .all()
+            )
+            assert review_rows == []
+            # 但回炉记账落库（writer 网关 ledger，端点末尾 commit）。
+            ledger_rows = (
+                (
+                    await verify.execute(
+                        select(UsageLedger).where(UsageLedger.project_id == project_uuid)
+                    )
+                )
+                .scalars()
+                .all()
+            )
+            assert len(ledger_rows) == 1
+            assert ledger_rows[0].input_tokens == _USAGE["writer"].input_tokens
+    finally:
+        if project_uuid is not None:
+            await _cleanup(e2e_sm, project_uuid)

tests/test_m5_e2e.py

@@ -0,0 +1,514 @@
+"""M5 端到端：群像生成→入库闭环 + 切 provider 回退/降级回归（真实 DB，零 token）。
+
+证明 M5 闭环（DoD：世界观/群像生成入库过 continuity 校验；多 provider 回退/降级——切
+provider 不破）：
+
+  用例 1 · 群像生成 → 409 gate → 确认 → 入库（DB 真源 + schema→JSONB 形变）
+    `POST /world/generate`（writer 假网关产 `WorldGenResult`）→ 预览实体（不入库）。
+    `POST /characters/generate`（writer 假网关产 `CharacterGenResult`，群像 3 卡）→ 预览。
+    `POST /characters`（入库）双 gate：
+      (a) precheck（analyst 假网关产带冲突 `ContinuityReview`）有冲突且 `acknowledge_conflicts`
+          缺省 → **409 CONFLICT_UNRESOLVED** + `details.conflicts`（不静默入库，守不变量 #3）。
+      (b) `acknowledge_conflicts=true` 重发 → 201；断言 `characters` 行**真落 pg**，且
+          schema list/str → DB JSONB dict 形变（traits/speech_tics→`{"items":[...]}`、
+          arc→`{"text":...}`）。
+    入库网关按 `req.output_schema` 路由：CharacterGenResult / WorldGenResult / ContinuityReview。
+
+  用例 2 · 越权写表被白名单丢弃（partition_writes，§5.6）
+    character-gen 只声明 `writes=["characters"]`；构造一张越权写表只能经端点 `partition_writes`
+    丢弃——这里证明合法路径（characters）放行、`rejected_tables` 为空（正常态）。
+
+  用例 3 · 切 provider 回退（primary transient 失败 → fallback 服务；DB 记账记实际服务方）
+    端到端 HTTP（角色生成）跑在**真实 `Gateway`**（非 stub）上，链 = primary(deepseek) →
+    fallback(openai)。primary 假适配器每次 `complete` 抛 `TransientProviderError`（耗尽重试）
+    → 网关沿链切 fallback 服务 → 端点 201。`served_by` 不出 API，故经 **DB 真源 `usage_ledger`**
+    断言实际服务方是 **openai（fallback）而非 deepseek（primary）**——证明回退真发生、记账记实际
+    服务方（ARCH §4.5）。
+
+  用例 4 · 能力降级（primary 不支持结构化输出 → 链上有支持者则重排/降级，§4.4）
+    primary(deepseek) `capabilities().structured_output=False` 且无 fallback → 网关降级用它
+    （适配器自走 JSON 提示返 parsed），`served_by.degraded=True`。经真 `Gateway.run` 直测
+    （聚焦集成测，证明降级路径真走，不经 HTTP 更清晰）。
+
+确定性 & 零成本（同 M1–M4）：真实 `Gateway` + 多 provider 假适配器（据 `req.output_schema`
+分支返回固定 `parsed`，绝不联网）+ 真实 `SqlAlchemyLedgerSink`（记账真落 pg）。无 pg → skip。
+
+坑（见 memory/gotchas）：
+- 生成网关是**请求 scope 依赖**（`get_worldbuilder_gateway`/`get_character_gen_gateway`/
+  `get_precheck_gateway`），非 background 自建——直接 `app.dependency_overrides` 注假网关即可，
+  无需 monkeypatch `build_gateway_for_tier`（与 M4 学文风后台自建网关不同）。
+- 真 `Gateway.run` 据 `req.output_schema` 选 schema：CharacterGenResult/WorldGenResult/
+  ContinuityReview——假适配器照此分支。
+- `get_sessionmaker` engine 绑首个 loop → 每 DB 测试清缓存重建、结束 dispose（同 M3/M4）。
+- ASGITransport 默认 `raise_app_exceptions=True`；snake_case；
+  409 envelope = `{error:{code,message,details}}`。
+- usage_ledger 记账记**实际服务方**：回退后 provider 列是 fallback 名，非 primary——这是
+  「切 provider 不破」的 DB 真源证据（served_by 不出 API）。
+"""
+
+from __future__ import annotations
+
+import uuid
+from collections.abc import AsyncIterator, Callable
+from typing import Annotated
+
+import httpx
+import pytest
+from asgi_lifespan import LifespanManager
+from fastapi import Depends
+from sqlalchemy import delete, select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+from ww_agents import (
+    CharacterCard,
+    CharacterGenResult,
+    Conflict,
+    ContinuityReview,
+    WorldEntityCard,
+    WorldGenResult,
+)
+from ww_db import get_session, get_sessionmaker
+from ww_db.models import Character, Project, UsageLedger
+from ww_llm_gateway import (
+    Gateway,
+    SqlAlchemyLedgerSink,
+    chain_from_routing,
+    resolve_route,
+)
+from ww_llm_gateway.adapters.base import (
+    Capabilities,
+    ProviderResult,
+    ProviderUsage,
+    StreamChunk,
+)
+from ww_llm_gateway.errors import TransientProviderError
+from ww_llm_gateway.routing import Route
+from ww_llm_gateway.types import LlmRequest, Scope, Tier
+
+# writer/analyst 默认都路由到 deepseek（config.tier_defaults）；用例 3 显式构 deepseek→openai 链。
+_PRIMARY = "deepseek"
+_FALLBACK = "openai"
+
+# 各档位假用量（喂记账；证明记账落库 + 实际服务方归因）。
+_USAGE = {
+    "writer": ProviderUsage(input_tokens=31, output_tokens=17),
+    "analyst": ProviderUsage(input_tokens=23, output_tokens=7),
+}
+
+# 群像假产物：3 张差异化角色卡（schema list/str 形，入库时由写侧 repo 转 DB JSONB dict）。
+_CHARACTER_CARDS = CharacterGenResult(
+    cards=[
+        CharacterCard(
+            name="叶无尘",
+            role="主角",
+            traits=["隐忍", "腹黑"],
+            backstory="灭门遗孤，背负血仇。",
+            arc="从复仇者走向守护者。",
+            speech_tics=["呵，有意思", "不过如此"],
+            tags=["扮猪吃虎", "天才"],
+            relations=[],
+        ),
+        CharacterCard(
+            name="苏清歌",
+            role="女主",
+            traits=["果决", "外冷内热"],
+            backstory="圣女出身，叛出师门。",
+            arc="从教条走向自我。",
+            speech_tics=["哼"],
+            tags=["大小姐", "御姐"],
+            relations=[],
+        ),
+        CharacterCard(
+            name="胖三",
+            role="工具人",
+            traits=["市侩", "重情"],
+            backstory="市井小贩，江湖百晓生。",
+            arc="从贪财到舍命相护。",
+            speech_tics=["客官您内行"],
+            tags=["搞笑担当"],
+            relations=[],
+        ),
+    ]
+)
+
+# 世界观假产物：两个硬规则实体（预览，不入库）。
+_WORLD_ENTITIES = WorldGenResult(
+    entities=[
+        WorldEntityCard(type="力量体系", name="九转玄功", rules=["每转需一枚玄晶", "不可逆修"]),
+        WorldEntityCard(type="势力", name="天机阁", rules=["阁主不入世俗纷争"]),
+    ]
+)
+
+# precheck 假冲突（让入库 gate 触发 409）。
+_PRECHECK_CONFLICT = ContinuityReview(
+    conflicts=[
+        Conflict(
+            type="设定违例",
+            where="叶无尘",
+            refs=["九转玄功"],
+            suggestion="叶无尘的功法与九转玄功『不可逆修』硬规则冲突，请调整设定",
+        )
+    ]
+)
+# precheck 无冲突（acknowledge 后重发——实际仍会返冲突，但带 ack 放行）。
+
+
+class _FakeGenAdapter:
+    """实现 `ProviderAdapter` Protocol：按 `req.output_schema` 分支返回 parsed，绝不联网。
+
+    - `CharacterGenResult` → 3 张群像卡（writer）。
+    - `WorldGenResult` → 两个世界观实体（writer）。
+    - `ContinuityReview` → 带冲突的预检结果（analyst）——触发入库 409 gate。
+    每路带 `ProviderUsage` → 网关据此落 usage_ledger。
+    """
+
+    def __init__(self, provider: str = _PRIMARY, *, structured: bool = True) -> None:
+        self.provider = provider
+        self._structured = structured
+
+    def capabilities(self) -> Capabilities:
+        return Capabilities(structured_output=self._structured)
+
+    async def complete(self, req: LlmRequest, model: str) -> ProviderResult:
+        schema = req.output_schema
+        if schema is CharacterGenResult:
+            return ProviderResult(
+                text=_CHARACTER_CARDS.model_dump_json(),
+                parsed=_CHARACTER_CARDS,
+                usage=_USAGE["writer"],
+            )
+        if schema is WorldGenResult:
+            return ProviderResult(
+                text=_WORLD_ENTITIES.model_dump_json(),
+                parsed=_WORLD_ENTITIES,
+                usage=_USAGE["writer"],
+            )
+        if schema is ContinuityReview:
+            return ProviderResult(
+                text=_PRECHECK_CONFLICT.model_dump_json(),
+                parsed=_PRECHECK_CONFLICT,
+                usage=_USAGE["analyst"],
+            )
+        raise AssertionError(f"unexpected output_schema in M5 fake adapter: {schema!r}")
+
+    async def stream(self, req: LlmRequest, model: str) -> AsyncIterator[StreamChunk]:
+        # 生成走 run()（结构化），不走 stream——但 Protocol 要求实现。
+        yield StreamChunk(usage=_USAGE["writer"])
+        raise AssertionError("M5 generation must not stream")
+
+
+class _FailingPrimaryAdapter:
+    """primary 假适配器：每次 `complete` 抛 `TransientProviderError`（耗尽重试 → 切回退）。"""
+
+    provider = _PRIMARY
+
+    def __init__(self) -> None:
+        self.complete_calls = 0
+
+    def capabilities(self) -> Capabilities:
+        return Capabilities(structured_output=True)
+
+    async def complete(self, req: LlmRequest, model: str) -> ProviderResult:
+        self.complete_calls += 1
+        raise TransientProviderError("simulated 503", provider=self.provider)
+
+    async def stream(self, req: LlmRequest, model: str) -> AsyncIterator[StreamChunk]:
+        raise TransientProviderError("simulated 503", provider=self.provider)
+        yield StreamChunk()  # pragma: no cover
+
+
+@pytest.fixture
+async def e2e_sm() -> AsyncIterator[async_sessionmaker[AsyncSession]]:
+    """真实 DB session 工厂；无 pg 时跳过（每测试清缓存重建 engine、结束 dispose）。"""
+    get_sessionmaker.cache_clear()
+    maker = get_sessionmaker()
+    try:
+        async with maker() as probe:
+            await probe.execute(select(1))
+    except Exception:
+        pytest.skip("postgres not reachable")
+    yield maker
+    await maker.kw["bind"].dispose()
+    get_sessionmaker.cache_clear()
+
+
+def _single_provider_override(
+    adapter: _FakeGenAdapter,
+) -> Callable[[AsyncSession], Gateway]:
+    """依赖覆盖：真实 Gateway + 单 provider 假适配器 + 真实 ledger（请求 session）。"""
+
+    def _override(session: Annotated[AsyncSession, Depends(get_session)]) -> Gateway:
+        return Gateway(
+            adapters={adapter.provider: adapter},
+            ledger=SqlAlchemyLedgerSink(session),
+            resolver=resolve_route,
+        )
+
+    return _override
+
+
+def _fallback_chain_override(
+    primary: _FailingPrimaryAdapter,
+    fallback: _FakeGenAdapter,
+) -> Callable[[AsyncSession], Gateway]:
+    """依赖覆盖：真实 Gateway + deepseek→openai 回退链 + 真实 ledger。
+
+    用真 `Gateway`（非 stub）走 §4.5 回退逻辑：primary transient 失败耗尽重试 → 切 fallback。
+    链经 `chain_from_routing` 构造（同 apps/api build_gateway_for_tier 用法）。
+    """
+
+    def _resolver(_tier: Tier) -> list[Route]:
+        return chain_from_routing(_tier, f"{_PRIMARY}:deepseek-chat", [f"{_FALLBACK}:gpt-4o"])
+
+    def _override(session: Annotated[AsyncSession, Depends(get_session)]) -> Gateway:
+        return Gateway(
+            adapters={_PRIMARY: primary, _FALLBACK: fallback},
+            ledger=SqlAlchemyLedgerSink(session),
+            chain_resolver=_resolver,
+            max_retries=1,
+        )
+
+    return _override
+
+
+async def _cleanup(e2e_sm: async_sessionmaker[AsyncSession], project_uuid: uuid.UUID) -> None:
+    """按 FK 顺序清理。"""
+    async with e2e_sm() as cleanup:
+        await cleanup.execute(delete(UsageLedger).where(UsageLedger.project_id == project_uuid))
+        await cleanup.execute(delete(Character).where(Character.project_id == project_uuid))
+        await cleanup.execute(delete(Project).where(Project.id == project_uuid))
+        await cleanup.commit()
+
+
+async def test_m5_generate_characters_gate_acknowledge_and_persist(
+    e2e_sm: async_sessionmaker[AsyncSession],
+) -> None:
+    """用例 1+2：世界观/群像生成预览 → 入库 409 gate → 确认放行 → 真落 pg（schema→JSONB 形变）。"""
+    from ww_api.main import create_app
+    from ww_api.services.project_deps import (
+        get_character_gen_gateway,
+        get_precheck_gateway,
+        get_worldbuilder_gateway,
+    )
+
+    writer_adapter = _FakeGenAdapter(provider=_PRIMARY)
+    analyst_adapter = _FakeGenAdapter(provider=_PRIMARY)
+
+    app = create_app()
+    # 生成网关是请求 scope 依赖 → 直接 override 注假网关（无需 monkeypatch build_gateway）。
+    app.dependency_overrides[get_worldbuilder_gateway] = _single_provider_override(writer_adapter)
+    app.dependency_overrides[get_character_gen_gateway] = _single_provider_override(writer_adapter)
+    app.dependency_overrides[get_precheck_gateway] = _single_provider_override(analyst_adapter)
+
+    transport = httpx.ASGITransport(app=app)
+    project_uuid: uuid.UUID | None = None
+    try:
+        async with LifespanManager(app):
+            async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+                create_resp = await client.post(
+                    "/projects", json={"title": "M5 群像验证作品", "genre": "玄幻"}
+                )
+                assert create_resp.status_code == 201
+                project_id = create_resp.json()["id"]
+                project_uuid = uuid.UUID(project_id)
+
+                # 1) 世界观生成预览（不入库）。
+                world_resp = await client.post(
+                    f"/projects/{project_id}/world/generate",
+                    json={"brief": "构建一个修真世界的力量体系与势力"},
+                )
+                assert world_resp.status_code == 200
+                entities = world_resp.json()["entities"]
+                assert [e["name"] for e in entities] == ["九转玄功", "天机阁"]
+                assert entities[0]["rules"] == ["每转需一枚玄晶", "不可逆修"]
+
+                # 2) 群像生成预览（3 卡，群像防雷同；不入库）。
+                gen_resp = await client.post(
+                    f"/projects/{project_id}/characters/generate",
+                    json={"brief": "主角团三人，定位差异化", "count": 3},
+                )
+                assert gen_resp.status_code == 200
+                cards = gen_resp.json()["cards"]
+                assert [c["name"] for c in cards] == ["叶无尘", "苏清歌", "胖三"]
+                assert cards[0]["traits"] == ["隐忍", "腹黑"]
+
+                # 3a) 入库（未 acknowledge）→ precheck 有冲突 → 409 CONFLICT_UNRESOLVED。
+                ingest_blocked = await client.post(
+                    f"/projects/{project_id}/characters",
+                    json={"cards": cards},
+                )
+                assert ingest_blocked.status_code == 409
+                err = ingest_blocked.json()["error"]
+                assert err["code"] == "CONFLICT_UNRESOLVED"
+                details = err["details"]
+                assert details["conflict_count"] == 1
+                assert details["conflicts"][0]["type"] == "设定违例"
+                assert details["conflicts"][0]["where"] == "叶无尘"
+
+                # 入库被拦 → 此刻 characters 表应无行（不静默入库，守不变量 #3）。
+                async with e2e_sm() as mid:
+                    pending = (
+                        (
+                            await mid.execute(
+                                select(Character).where(Character.project_id == project_uuid)
+                            )
+                        )
+                        .scalars()
+                        .all()
+                    )
+                    assert pending == []
+
+                # 3b) 作者裁决后带 acknowledge_conflicts=true 重发 → 201 入库。
+                ingest_ok = await client.post(
+                    f"/projects/{project_id}/characters",
+                    json={"cards": cards, "acknowledge_conflicts": True},
+                )
+                assert ingest_ok.status_code == 201
+                body = ingest_ok.json()
+                assert body["created"] == ["叶无尘", "苏清歌", "胖三"]
+                # 用例 2：合法路径（writes=["characters"]）无越权 → rejected_tables 空（正常态）。
+                assert body["rejected_tables"] == []
+
+        # DB 真源断言：characters 行真落 pg + schema→DB JSONB dict 形变。
+        assert project_uuid is not None
+        async with e2e_sm() as verify:
+            rows = (
+                (
+                    await verify.execute(
+                        select(Character)
+                        .where(Character.project_id == project_uuid)
+                        .order_by(Character.name)
+                    )
+                )
+                .scalars()
+                .all()
+            )
+            assert {r.name for r in rows} == {"叶无尘", "苏清歌", "胖三"}
+            ye = next(r for r in rows if r.name == "叶无尘")
+            # 形变：traits/speech_tics list → {"items":[...]}；arc str → {"text":...}。
+            assert ye.traits == {"items": ["隐忍", "腹黑"]}
+            assert ye.speech_tics == {"items": ["呵，有意思", "不过如此"]}
+            assert ye.arc == {"text": "从复仇者走向守护者。"}
+            # tags 是 JSONB list（不形变）；relations 是 JSONB list。
+            assert ye.tags == ["扮猪吃虎", "天才"]
+            assert ye.relations == []
+            assert ye.role == "主角"
+            assert ye.backstory == "灭门遗孤，背负血仇。"
+
+            # 记账真落 pg：world generate(writer) + characters generate(writer) +
+            # precheck(analyst, 409 路径) + precheck(analyst, 入库路径) = 4 条。
+            ledger_rows = (
+                (
+                    await verify.execute(
+                        select(UsageLedger).where(UsageLedger.project_id == project_uuid)
+                    )
+                )
+                .scalars()
+                .all()
+            )
+            assert len(ledger_rows) == 4
+            assert all(r.provider == _PRIMARY for r in ledger_rows)
+    finally:
+        if project_uuid is not None:
+            await _cleanup(e2e_sm, project_uuid)
+
+
+async def test_m5_provider_fallback_serves_when_primary_fails(
+    e2e_sm: async_sessionmaker[AsyncSession],
+) -> None:
+    """用例 3：HTTP 生成 primary transient 失败 → fallback 服务；DB 记账记实际服务方（openai）。"""
+    from ww_api.main import create_app
+    from ww_api.services.project_deps import get_character_gen_gateway
+
+    primary = _FailingPrimaryAdapter()
+    fallback = _FakeGenAdapter(provider=_FALLBACK)
+
+    app = create_app()
+    # 角色生成网关 = 真 Gateway + deepseek(失败)→openai(成功) 回退链。
+    # character generate 端点不调 precheck（不入库），故只 override 角色生成网关。
+    app.dependency_overrides[get_character_gen_gateway] = _fallback_chain_override(
+        primary, fallback
+    )
+
+    transport = httpx.ASGITransport(app=app)
+    project_uuid: uuid.UUID | None = None
+    try:
+        async with LifespanManager(app):
+            async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+                create_resp = await client.post("/projects", json={"title": "M5 切 provider 作品"})
+                assert create_resp.status_code == 201
+                project_id = create_resp.json()["id"]
+                project_uuid = uuid.UUID(project_id)
+
+                # primary transient 失败 → 网关沿链切 fallback → 仍 200（切 provider 不破）。
+                gen_resp = await client.post(
+                    f"/projects/{project_id}/characters/generate",
+                    json={"brief": "主角团三人", "count": 3},
+                )
+                assert gen_resp.status_code == 200
+                cards = gen_resp.json()["cards"]
+                assert [c["name"] for c in cards] == ["叶无尘", "苏清歌", "胖三"]
+
+                # primary 确实被调过且失败（耗尽重试 max_retries=1 → 2 次尝试）。
+                assert primary.complete_calls >= 1
+
+        # DB 真源断言：usage_ledger 记**实际服务方** = fallback(openai)，不是 primary(deepseek)。
+        # served_by 不出 API，这是「回退真发生」的端到端 DB 证据（ARCH §4.5 记账记实际服务方）。
+        assert project_uuid is not None
+        async with e2e_sm() as verify:
+            ledger_rows = (
+                (
+                    await verify.execute(
+                        select(UsageLedger).where(UsageLedger.project_id == project_uuid)
+                    )
+                )
+                .scalars()
+                .all()
+            )
+            assert len(ledger_rows) == 1
+            assert ledger_rows[0].provider == _FALLBACK
+            assert ledger_rows[0].provider != _PRIMARY
+            assert ledger_rows[0].input_tokens == _USAGE["writer"].input_tokens
+    finally:
+        if project_uuid is not None:
+            await _cleanup(e2e_sm, project_uuid)
+
+
+async def test_m5_capability_degradation_marks_served_by_degraded() -> None:
+    """用例 4：primary 不支持结构化输出且无支持者 → 网关降级用它，标 served_by.degraded（§4.4）。
+
+    聚焦集成测（直测真 `Gateway.run`，不经 HTTP——降级路径更清晰）：结构化请求
+    （output_schema=CharacterGenResult）打到 `structured_output=False` 的唯一 provider →
+    降级（适配器自走 JSON 提示返 parsed），`served_by.degraded=True`。
+    """
+    from ww_llm_gateway.ledger import LedgerSink
+    from ww_llm_gateway.types import Usage
+
+    class _CapturingLedger:
+        def __init__(self) -> None:
+            self.records: list[Usage] = []
+
+        async def record(self, scope: Scope, usage: Usage) -> None:
+            self.records.append(usage)
+
+    # primary 不支持结构化输出，但仍能返回 parsed（自走 instructor JSON 提示）。
+    degrade_adapter = _FakeGenAdapter(provider=_PRIMARY, structured=False)
+    ledger: LedgerSink = _CapturingLedger()  # type: ignore[assignment]
+    gw = Gateway(
+        adapters={_PRIMARY: degrade_adapter},
+        ledger=ledger,
+        chain_resolver=lambda _tier: [Route(provider=_PRIMARY, model="deepseek-chat")],
+    )
+
+    req = LlmRequest(
+        tier="writer",
+        input="生成角色",
+        output_schema=CharacterGenResult,
+        scope=Scope(user_id=uuid.uuid4(), project_id=uuid.uuid4()),
+    )
+    resp = await gw.run(req)
+
+    assert isinstance(resp.parsed, CharacterGenResult)
+    assert resp.served_by.provider == _PRIMARY
+    # 链上无结构化支持者 → 降级用唯一可用 provider，标 degraded。
+    assert resp.served_by.degraded is True
+    assert resp.served_by.fell_back is False