345cc73965
P0-1 SqlCredentialStore/save_draft 由自提交改 flush，端点/服务统一 commit （新增 CredentialStore.commit() 统一提交点；token 刷新落库显式提交）； 补多凭据一请求中途失败整体回滚集成测试。 P0-2 启动校验 _fernet(enc_key) 快速失败 + catch-all Exception → ErrorEnvelope； credential_enc_key 改 SecretStr。 P0-3 run_job 异常分类：AppError 存 code+message，其余存通用文案不泄 str(exc)。 P0-4 评审/正文 SSE 失败先发 error 事件，尾部 commit 包 try/except。 P1-4 max_version 加 FOR UPDATE 行锁消除 TOCTOU。 P1-5 scan_overdue 谓词下推 + 批量 UPDATE RETURNING。 P1-10 移除 OAuth user_code 日志。 P2 provider_deps 改调网关 build_adapter；accept_service Committable Protocol； CORS 白名单收窄；request_id 安全字符集白名单；stdlib 日志接管；读端点 404 校验； httpx timeout；测试用合法 Fernet key；类型化响应模型(JobResponse/DimensionEntry/ ReviewConflictView/selling_points)+路由 ErrorEnvelope responses（供 codegen）。
432 lines
18 KiB
Python
432 lines
18 KiB
Python
"""项目/章节端点的依赖装配（运行时实现）。
|
||
|
||
- `get_project_repo` / `get_chapter_repo`：把请求 session 装配成 SQLAlchemy repo。
|
||
- `get_writer_gateway`：据 writer 档位路由从已存凭据解密 → 建 OpenAI 兼容适配器
|
||
→ `Gateway`（注入 `SqlAlchemyLedgerSink` + `resolve_route`）。这是 **draft SSE 的可注入缝**——
|
||
测试经 `app.dependency_overrides[get_writer_gateway]` 注入 mock 网关（产 `Delta`，绝不联网）。
|
||
- `seed_stub_user`：幂等 seed 单用户 stub（owner_id FK 依赖它，见 memory/gotchas）。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Annotated
|
||
|
||
import httpx
|
||
from fastapi import Depends
|
||
from sqlalchemy import select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
from ww_config import get_settings
|
||
from ww_core.domain import ForeshadowLedgerRepo, SqlForeshadowLedgerRepo
|
||
from ww_core.domain.chapter_repo import ChapterRepo, SqlChapterRepo
|
||
from ww_core.domain.character_repo import CharacterWriteRepo, SqlCharacterWriteRepo
|
||
from ww_core.domain.digest_repo import DigestAppendRepo, SqlDigestAppendRepo
|
||
from ww_core.domain.injection_repo import InjectionOverrideRepo, SqlInjectionOverrideRepo
|
||
from ww_core.domain.job_repo import JobRepo, SqlJobRepo
|
||
from ww_core.domain.outline_write_repo import OutlineWriteRepo, SqlOutlineWriteRepo
|
||
from ww_core.domain.project_repo import ProjectRepo, SqlProjectRepo
|
||
from ww_core.domain.repositories import MemoryRepos, OutlineRepo, RulesRepo
|
||
from ww_core.domain.review_repo import ReviewRepo, SqlReviewRepo
|
||
from ww_core.domain.rule_repo import RuleWriteRepo, SqlRuleWriteRepo
|
||
from ww_core.domain.style_repo import SqlStyleFingerprintWriteRepo, StyleFingerprintWriteRepo
|
||
from ww_core.domain.world_entity_repo import SqlWorldEntityWriteRepo, WorldEntityWriteRepo
|
||
from ww_core.memory.sql_repositories import SqlOutlineRepo, SqlRulesRepo, sql_memory_repos
|
||
from ww_db import get_session, get_sessionmaker
|
||
from ww_db.models import User
|
||
from ww_llm_gateway import (
|
||
Gateway,
|
||
ProviderAdapter,
|
||
Route,
|
||
SqlAlchemyLedgerSink,
|
||
build_adapter,
|
||
chain_from_routing,
|
||
resolve_route,
|
||
)
|
||
from ww_llm_gateway.adapters.kimi_code import KIMI_CODE_PROVIDER
|
||
from ww_llm_gateway.types import Tier
|
||
from ww_shared import AppError, ErrorCode
|
||
from ww_skills import SkillRegistry, SqlSkillRepo
|
||
|
||
from ww_api.security.credentials import (
|
||
CredentialKeyError,
|
||
decrypt_api_key,
|
||
)
|
||
from ww_api.services.credentials import (
|
||
AUTH_TYPE_OAUTH,
|
||
STUB_OWNER_ID,
|
||
CredentialStore,
|
||
SqlCredentialStore,
|
||
StoredCredential,
|
||
)
|
||
from ww_api.services.foreshadow_scan import SessionFactory
|
||
from ww_api.services.kimi_oauth import (
|
||
decrypt_oauth_bundle,
|
||
encrypt_oauth_bundle,
|
||
needs_refresh,
|
||
)
|
||
from ww_api.services.kimi_oauth import refresh as kimi_refresh
|
||
from ww_api.services.provider_deps import _PROVIDER_BASE_URLS
|
||
|
||
# 单用户 stub 的占位邮箱（多租户化时由真实主体替换）。
|
||
_STUB_USER_EMAIL = "stub@local"
|
||
|
||
|
||
async def seed_stub_user(session: AsyncSession) -> None:
|
||
"""幂等 seed 单用户 stub 行——所有 owner_id FK（projects/usage_ledger/...）依赖它。
|
||
|
||
无该行时插入；已存在则跳过。在 app lifespan 启动时调用一次。
|
||
"""
|
||
existing = (
|
||
await session.execute(select(User).where(User.id == STUB_OWNER_ID))
|
||
).scalar_one_or_none()
|
||
if existing is not None:
|
||
return
|
||
session.add(User(id=STUB_OWNER_ID, email=_STUB_USER_EMAIL, display_name="stub"))
|
||
await session.commit()
|
||
|
||
|
||
def get_project_repo(
|
||
session: Annotated[AsyncSession, Depends(get_session)],
|
||
) -> ProjectRepo:
|
||
return SqlProjectRepo(session)
|
||
|
||
|
||
def get_chapter_repo(
|
||
session: Annotated[AsyncSession, Depends(get_session)],
|
||
) -> ChapterRepo:
|
||
return SqlChapterRepo(session)
|
||
|
||
|
||
def get_memory_repos(
|
||
session: Annotated[AsyncSession, Depends(get_session)],
|
||
) -> MemoryRepos:
|
||
"""记忆组装的 7-repo 捆绑（draft SSE 用）。测试覆盖此依赖注入内存 fake。"""
|
||
return sql_memory_repos(session)
|
||
|
||
|
||
def get_injection_repo(
|
||
session: Annotated[AsyncSession, Depends(get_session)],
|
||
) -> InjectionOverrideRepo:
|
||
"""本章注入覆盖读/写 repo（B0 可控版 GET/PUT injection + draft 同读）。
|
||
|
||
upsert 只 flush，端点提交（与其它写侧一致）。测试经 `app.dependency_overrides` 注 fake。
|
||
"""
|
||
return SqlInjectionOverrideRepo(session)
|
||
|
||
|
||
def get_review_repo(
|
||
session: Annotated[AsyncSession, Depends(get_session)],
|
||
) -> ReviewRepo:
|
||
"""审稿留痕 repo（review SSE collect / 历史 / accept 裁决）。"""
|
||
return SqlReviewRepo(session)
|
||
|
||
|
||
def get_digest_append_repo(
|
||
session: Annotated[AsyncSession, Depends(get_session)],
|
||
) -> DigestAppendRepo:
|
||
"""章节摘要写侧 repo（验收事务追加终稿 digest）。"""
|
||
return SqlDigestAppendRepo(session)
|
||
|
||
|
||
def get_foreshadow_repo(
|
||
session: Annotated[AsyncSession, Depends(get_session)],
|
||
) -> ForeshadowLedgerRepo:
|
||
"""伏笔账本写侧 repo（登记/状态变更端点；只 flush，端点提交）。测试经 override 注 fake。"""
|
||
return SqlForeshadowLedgerRepo(session)
|
||
|
||
|
||
def get_job_repo(
|
||
session: Annotated[AsyncSession, Depends(get_session)],
|
||
) -> JobRepo:
|
||
"""长任务写侧 repo（创建/进度/完成/失败；状态写只 flush，提交归 run_job/端点）。
|
||
|
||
测试经 `app.dependency_overrides[get_job_repo]` 注入 fake（避免真连 DB）。
|
||
"""
|
||
return SqlJobRepo(session)
|
||
|
||
|
||
def get_outline_write_repo(
|
||
session: Annotated[AsyncSession, Depends(get_session)],
|
||
) -> OutlineWriteRepo:
|
||
"""大纲写侧 repo（大纲生成端点逐章 upsert；只 flush，端点提交）。测试经 override 注 fake。"""
|
||
return SqlOutlineWriteRepo(session)
|
||
|
||
|
||
def get_rule_write_repo(
|
||
session: Annotated[AsyncSession, Depends(get_session)],
|
||
) -> RuleWriteRepo:
|
||
"""规则写侧 repo（POST /rules：作者显式加规则；只 flush，端点提交）。测试经 override 注。"""
|
||
return SqlRuleWriteRepo(session)
|
||
|
||
|
||
async def get_skill_registry(
|
||
session: Annotated[AsyncSession, Depends(get_session)],
|
||
) -> SkillRegistry:
|
||
"""从 `skills` 表加载声明式 skill registry（ARCH §5.6；越权声明 → VALIDATION）。
|
||
|
||
每请求按 session 加载（registry 不可变快照）。测试经 `app.dependency_overrides` 注 fake repo
|
||
或直接注 `SkillRegistry`。技能库 UI（T5.6）经此读 builtin/custom/community。
|
||
"""
|
||
return await SkillRegistry.load(SqlSkillRepo(session))
|
||
|
||
|
||
def get_style_write_repo(
|
||
session: Annotated[AsyncSession, Depends(get_session)],
|
||
) -> StyleFingerprintWriteRepo:
|
||
"""文风指纹写侧 repo（`GET /style` 读最新 + 学文风后台任务 append 版本化指纹）。
|
||
|
||
注：学文风的 `work` 在 `run_job` 自建的独立 session 上自造 repo（请求 session 已关闭），
|
||
故本依赖只服务于 `GET /style` 读侧。测试经 `app.dependency_overrides` 注 fake。
|
||
"""
|
||
return SqlStyleFingerprintWriteRepo(session)
|
||
|
||
|
||
def get_character_write_repo(
|
||
session: Annotated[AsyncSession, Depends(get_session)],
|
||
) -> CharacterWriteRepo:
|
||
"""角色写侧 repo（POST /characters 入库：schema→DB 形变；只 flush，端点提交）。
|
||
|
||
测试经 `app.dependency_overrides` 注 fake。
|
||
"""
|
||
return SqlCharacterWriteRepo(session)
|
||
|
||
|
||
def get_world_entity_write_repo(
|
||
session: Annotated[AsyncSession, Depends(get_session)],
|
||
) -> WorldEntityWriteRepo:
|
||
"""世界观实体写侧 repo（预留对称入库；当前生成端点只用其形变能力）。"""
|
||
return SqlWorldEntityWriteRepo(session)
|
||
|
||
|
||
def get_rules_read_repo(
|
||
session: Annotated[AsyncSession, Depends(get_session)],
|
||
) -> RulesRepo:
|
||
"""规则读侧 repo（GET /rules 列表，复用 C5 assemble 读侧；测试经 override 注 fake）。"""
|
||
return SqlRulesRepo(session)
|
||
|
||
|
||
def get_outline_read_repo(
|
||
session: Annotated[AsyncSession, Depends(get_session)],
|
||
) -> OutlineRepo:
|
||
"""大纲读侧 repo（GET /outline 列表，复用 C5 assemble 读侧；测试经 override 注 fake）。"""
|
||
return SqlOutlineRepo(session)
|
||
|
||
|
||
async def get_worldbuilder_gateway(
|
||
session: Annotated[AsyncSession, Depends(get_session)],
|
||
) -> Gateway:
|
||
"""世界观生成（writer 档位）的可注入网关缝。测试经 override 注 mock（产 WorldGenResult）。"""
|
||
store = SqlCredentialStore(session)
|
||
return await build_gateway_for_tier(session, store, "writer")
|
||
|
||
|
||
async def get_character_gen_gateway(
|
||
session: Annotated[AsyncSession, Depends(get_session)],
|
||
) -> Gateway:
|
||
"""角色生成（writer 档位）的可注入网关缝。测试经 override 注 mock（产 CharacterGenResult）。"""
|
||
store = SqlCredentialStore(session)
|
||
return await build_gateway_for_tier(session, store, "writer")
|
||
|
||
|
||
async def get_precheck_gateway(
|
||
session: Annotated[AsyncSession, Depends(get_session)],
|
||
) -> Gateway:
|
||
"""入库前 continuity 预检（analyst 档位）的可注入网关缝。测试注 mock（产 ContinuityReview）。"""
|
||
store = SqlCredentialStore(session)
|
||
return await build_gateway_for_tier(session, store, "analyst")
|
||
|
||
|
||
def get_session_factory() -> SessionFactory:
|
||
"""验收后到期扫描的**独立 session 工厂**缝。
|
||
|
||
BackgroundTask 在请求 session 关闭后才跑，必须自建 session（不复用 `get_session`）。
|
||
返回的工厂 `()` → `async with` 得一个新 `AsyncSession`。测试经 `app.dependency_overrides`
|
||
注入 fake 工厂（避免真起后台线程/真连 DB）。
|
||
"""
|
||
return get_sessionmaker()
|
||
|
||
|
||
async def _build_provider_adapter(store: CredentialStore, provider: str) -> ProviderAdapter | None:
|
||
"""据 provider 解密凭据 → 经 `build_adapter` 工厂建对应 provider 适配器（T5.4 follow-up）。
|
||
|
||
`build_adapter(provider, *, api_key, base_url=None)` 按 provider 选适配器类：
|
||
OpenAI 兼容（deepseek/kimi/qwen/glm/openai）经 `base_url` 走 OpenAI 兼容适配器；
|
||
Anthropic/Gemini 走各自原生适配器（无需 `base_url`）；`kimi-code` 走 OAuth bearer +
|
||
coding base + 伪造头（**access token 经 `_resolve_kimi_code_token` 按需刷新**，K1.3）。
|
||
|
||
返回 `None` 表示该 provider 未配置凭据——回退链里缺位时网关会跳到下一个，故宽容返回
|
||
None（不直接抛）。
|
||
"""
|
||
cred = await store.get_credential(STUB_OWNER_ID, provider)
|
||
if cred is None:
|
||
return None
|
||
settings = get_settings()
|
||
|
||
if cred.auth_type == AUTH_TYPE_OAUTH or provider == KIMI_CODE_PROVIDER:
|
||
# OAuth 凭据（Kimi Code）：解密 token 包 → 临近过期则刷新并持久化 → access token
|
||
# 当 api_key 喂工厂（工厂为 kimi-code 构建带伪造头 + coding base 的客户端）。
|
||
access_token = await _resolve_kimi_code_token(
|
||
store, cred, settings.credential_enc_key.get_secret_value()
|
||
)
|
||
return build_adapter(
|
||
provider, api_key=access_token, base_url=_PROVIDER_BASE_URLS.get(provider)
|
||
)
|
||
|
||
if cred.api_key_enc is None:
|
||
# api_key 凭据但密文缺失（数据不一致）——视作未配置，回退链跳过。
|
||
return None
|
||
try:
|
||
api_key = decrypt_api_key(
|
||
cred.api_key_enc, key=settings.credential_enc_key.get_secret_value()
|
||
)
|
||
except CredentialKeyError as exc:
|
||
raise AppError(ErrorCode.INTERNAL, str(exc)) from exc
|
||
# OpenAI 兼容 provider 需 base_url；Anthropic/Gemini 走原生 SDK（base_url=None）。
|
||
base_url = _PROVIDER_BASE_URLS.get(provider)
|
||
return build_adapter(provider, api_key=api_key, base_url=base_url)
|
||
|
||
|
||
async def _resolve_kimi_code_token(
|
||
store: CredentialStore, cred: StoredCredential, enc_key: str
|
||
) -> str:
|
||
"""解密 Kimi Code OAuth token 包 → 临近过期时刷新并持久化 → 返回当前 access token。
|
||
|
||
刷新经一个**临时 httpx 客户端**（与 token 端点交互）；新 token 包经
|
||
`store.upsert_oauth_credential` 持久化（下次建网关复用刷新结果）。明文 token 绝不进
|
||
日志/响应。无 `oauth_enc` → `LLM_UNAVAILABLE`（未连接 Kimi Code）。
|
||
"""
|
||
if cred.oauth_enc is None:
|
||
raise AppError(
|
||
ErrorCode.LLM_UNAVAILABLE,
|
||
f"{KIMI_CODE_PROVIDER} 未连接（无 OAuth 凭据），请先在设置中连接 Kimi Code",
|
||
{"provider": KIMI_CODE_PROVIDER},
|
||
)
|
||
try:
|
||
token = decrypt_oauth_bundle(cred.oauth_enc, key=enc_key)
|
||
except CredentialKeyError as exc:
|
||
raise AppError(ErrorCode.INTERNAL, str(exc)) from exc
|
||
|
||
if not needs_refresh(token):
|
||
return token.access_token
|
||
|
||
# 临近过期 → 刷新并持久化新包。
|
||
async with httpx.AsyncClient(timeout=30.0) as http:
|
||
refreshed = await kimi_refresh(http, token.refresh_token)
|
||
new_blob = encrypt_oauth_bundle(refreshed, key=enc_key)
|
||
await store.upsert_oauth_credential(STUB_OWNER_ID, KIMI_CODE_PROVIDER, new_blob)
|
||
# token 刷新是独立可持久的副作用（下次建网关复用），须立即提交，不依赖请求后续是否提交。
|
||
await store.commit()
|
||
return refreshed.access_token
|
||
|
||
|
||
async def build_gateway_for_tier(
|
||
session: AsyncSession, store: CredentialStore, tier: Tier
|
||
) -> Gateway:
|
||
"""据指定档位路由 + DB `tier_routing.fallback` 装配**多 provider 回退链**网关（T5.4 接线）。
|
||
|
||
流程（§4.3 三级解析 / §4.5 回退链）：
|
||
1. 读 DB `tier_routing` 取该 tier 的 primary `provider:model` + fallback 列表（缺则退回
|
||
全局 `resolve_route`，单 provider，向后兼容）。
|
||
2. 为 primary + fallback 里**每个能建出适配器**的 provider 预备 OpenAI 兼容适配器
|
||
（未知 base_url / 未配凭据的 provider 跳过——回退链自然绕过它）。
|
||
3. 至少要有一个可用适配器，否则 `LLM_UNAVAILABLE`（无任何凭据可用）。
|
||
4. 注入 `chain_resolver=chain_from_routing(...)`（多元素链，启用回退）；无 DB 路由时
|
||
用 `resolver=resolve_route`（单路由，M1 行为不变）。
|
||
|
||
单 provider 配置仍走单元素链（网关把它当无回退处理），**不破既有行为**。
|
||
"""
|
||
ledger = SqlAlchemyLedgerSink(session)
|
||
|
||
stored = next((r for r in await store.list_routing() if r.tier == tier), None)
|
||
if stored is None:
|
||
# 无 DB 路由行：退回全局默认（单 provider，M1 兼容）。
|
||
route = resolve_route(tier)
|
||
adapter = await _build_provider_adapter(store, route.provider)
|
||
if adapter is None:
|
||
raise AppError(
|
||
ErrorCode.LLM_UNAVAILABLE,
|
||
f"{tier} 档位 provider {route.provider} 未配置凭据，请先在设置中配置",
|
||
{"provider": route.provider, "tier": tier},
|
||
)
|
||
return Gateway(adapters={route.provider: adapter}, ledger=ledger, resolver=resolve_route)
|
||
|
||
# DB 路由：primary + fallback 构链；为每个可建的 provider 预备适配器。
|
||
primary_spec = f"{stored.provider}:{stored.model}"
|
||
chain: list[Route] = chain_from_routing(tier, primary_spec, list(stored.fallback))
|
||
adapters: dict[str, ProviderAdapter] = {}
|
||
for route in chain:
|
||
if route.provider in adapters:
|
||
continue
|
||
built = await _build_provider_adapter(store, route.provider)
|
||
if built is not None:
|
||
adapters[route.provider] = built
|
||
if not adapters:
|
||
raise AppError(
|
||
ErrorCode.LLM_UNAVAILABLE,
|
||
f"{tier} 档位无任何已配置凭据的 provider，请先在设置中配置",
|
||
{"providers": [r.provider for r in chain], "tier": tier},
|
||
)
|
||
|
||
def _resolver(_tier: Tier) -> list[Route]:
|
||
return chain_from_routing(_tier, primary_spec, list(stored.fallback))
|
||
|
||
return Gateway(adapters=adapters, ledger=ledger, chain_resolver=_resolver)
|
||
|
||
|
||
async def build_writer_gateway(session: AsyncSession, store: CredentialStore) -> Gateway:
|
||
"""据 writer 档位路由解密对应 provider 凭据 → 建网关。"""
|
||
return await build_gateway_for_tier(session, store, "writer")
|
||
|
||
|
||
async def get_writer_gateway(
|
||
session: Annotated[AsyncSession, Depends(get_session)],
|
||
) -> Gateway:
|
||
"""draft SSE 的可注入网关缝。测试覆盖此依赖注入 mock（产 `Delta`，绝不联网）。"""
|
||
store = SqlCredentialStore(session)
|
||
return await build_writer_gateway(session, store)
|
||
|
||
|
||
async def get_review_gateway(
|
||
session: Annotated[AsyncSession, Depends(get_session)],
|
||
) -> Gateway:
|
||
"""续审（analyst 档位）的可注入网关缝。测试经 override 注 mock（产 `parsed`，绝不联网）。"""
|
||
store = SqlCredentialStore(session)
|
||
return await build_gateway_for_tier(session, store, "analyst")
|
||
|
||
|
||
async def get_digest_gateway(
|
||
session: Annotated[AsyncSession, Depends(get_session)],
|
||
) -> Gateway:
|
||
"""验收终稿 digest 提炼（light 档位）的可注入网关缝。测试经 override 注 mock。"""
|
||
store = SqlCredentialStore(session)
|
||
return await build_gateway_for_tier(session, store, "light")
|
||
|
||
|
||
async def get_outline_gateway(
|
||
session: Annotated[AsyncSession, Depends(get_session)],
|
||
) -> Gateway:
|
||
"""大纲生成（analyst 档位）的可注入网关缝。测试经 override 注 mock（产 OutlineResult）。"""
|
||
store = SqlCredentialStore(session)
|
||
return await build_gateway_for_tier(session, store, "analyst")
|
||
|
||
|
||
async def get_style_extract_gateway(
|
||
session: Annotated[AsyncSession, Depends(get_session)],
|
||
) -> Gateway:
|
||
"""学文风提取（analyst 档位）的可注入网关缝。
|
||
|
||
`POST /style` 在 dep 解析阶段构建网关 → 无凭据时这里抛 `LLM_UNAVAILABLE`（503，
|
||
调度 job 之前拦下，避免凭空写一行注定失败的 job）。提取本体在 BackgroundTask 里
|
||
用 `run_job` 自建的独立 session 重新构网关跑（请求 session 已关闭）；本依赖确保
|
||
凭据探测在请求阶段发生。测试经 override 注 mock（产 StyleFingerprintResult）。
|
||
"""
|
||
store = SqlCredentialStore(session)
|
||
return await build_gateway_for_tier(session, store, "analyst")
|
||
|
||
|
||
async def get_refine_gateway(
|
||
session: Annotated[AsyncSession, Depends(get_session)],
|
||
) -> Gateway:
|
||
"""回炉（writer 档位）的可注入网关缝。测试经 override 注 mock（产纯文本重写段）。"""
|
||
store = SqlCredentialStore(session)
|
||
return await build_gateway_for_tier(session, store, "writer")
|