f43ccd293f
通用执行路径驱动全部生成器（"加生成器=加一份声明"）：
- @llm: ww_agents +7 输出 schema + 7 spec（book-title/blurb/name/golden-finger/
glossary/opening/fine-outline，只声明 tier）+ build_outline_chapter_context
- @backend: ww_skills GeneratorTool 描述符 + TOOLBOX(11) + get_tool；3 通用端点
GET /skills/toolbox · POST .../skills/{tool_key}/generate(预览不写库,仅记账) ·
POST .../ingest(复用 continuity 409 + partition_writes 白名单)；纯 context 派发
- @frontend: 工具箱落地页 RSC + 声明驱动 GeneratorRunner + lib/toolbox 纯函数
+ LeftNav「工具箱」+ ⌘K nav-toolbox/action-gen-*；legacy 3 跳现页
- @qa: tests/test_t6_toolbox_e2e.py 5 用例真 pg + mock 网关零 token，无端点 bug
- P2 收尾: 限流→decisions.md 记延后(单用户原型)；noopener/Committable 早已修
守不变量 #2(只声明 tier)/#3(预览不写库,入库经验收 gate)/#9(缓存前缀)。无 DB 迁移。
门禁绿: 后端 ruff/format/mypy 195/alembic 无漂移/pytest 583；前端 lint/tsc/vitest 279/build。
spec 回写 PRODUCT_SPEC §7 + ARCHITECTURE §7.2 端点表。
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
515 lines
22 KiB
Python
515 lines
22 KiB
Python
"""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()
|
||
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
|