"""创作工具箱通用端点（C3 扩 T6 / 通用生成器框架；不变量 #1/#2/#3/#9）。 声明驱动、一条执行路径——「加一个生成器」= 在 `ww_skills.TOOLBOX` 加一份声明，无需新端点： - `GET /skills/toolbox` 列出全部生成器描述符（legacy + 新工具）。 - `POST /projects/:id/skills/:tool_key/generate` 按 context_strategy 组材料 → 网关结构化预览。 - `POST /projects/:id/skills/:tool_key/ingest` 仅可入库工具：过 continuity 预检 + 白名单 → 写库。 legacy 工具（spec=None）无通用执行路径——generate/ingest 对其返 404（前端走 legacy_route 跳现页）。 未知 tool_key→404；项目不存在→404；无凭据→503 LLM_UNAVAILABLE；缺章/大纲→空节拍不报错（预览）。 提交边界：网关 ledger + 写侧均只 flush → 端点末尾一次 `commit()`（含冲突 409 路径也 commit 落 precheck usage，同 characters 入库纪律，不变量 #3）。 """ from __future__ import annotations import uuid from typing import Annotated, Any from fastapi import APIRouter, Depends, Request from sqlalchemy.ext.asyncio import AsyncSession from ww_agents import AgentSpec, Conflict, ContinuityReview, continuity_spec from ww_core.domain.chapter_repo import ChapterRepo from ww_core.domain.outline_write_repo import OutlineWriteRepo from ww_core.domain.project_repo import ProjectRepo from ww_core.domain.repositories import MemoryRepos, OutlineRepo from ww_core.domain.rule_repo import RuleWriteRepo from ww_core.domain.world_entity_repo import WorldEntityFields, WorldEntityWriteRepo from ww_core.orchestrator import run_generator from ww_db import get_session from ww_llm_gateway import Gateway from ww_llm_gateway.types import Block, LlmRequest, Scope from ww_shared import AppError, ErrorCode, ErrorEnvelope from ww_skills import GeneratorTool, get_tool, partition_writes from ww_api.logging_config import get_logger from ww_api.routers.generation import _project_context, _world_context from ww_api.schemas.generation import IngestConflictView, WorldEntityCardView from ww_api.schemas.toolbox import ( OutlineSceneIngestView, TeardownIngestView, ToolboxListResponse, ToolDescriptorView, ToolGeneratePreviewResponse, ToolGenerateRequest, ToolIngestRequest, ToolIngestResponse, ToolInputFieldView, ) from ww_api.services.credentials import STUB_OWNER_ID from ww_api.services.project_deps import ( TierGatewayBuilder, get_chapter_repo, get_memory_repos, get_outline_read_repo, get_outline_write_repo, get_project_repo, get_rule_write_repo, get_tier_gateway_builder, get_world_entity_write_repo, ) from ww_api.services.toolbox_context import build_toolbox_context log = get_logger("ww.api.toolbox") router = APIRouter(tags=["toolbox"]) # OpenAPI 错误响应声明（类型化错误形，§7.1）。 _TOOL_ERRORS: dict[int | str, dict[str, Any]] = { 404: {"model": ErrorEnvelope, "description": "工具或项目不存在 / legacy 工具无通用执行"}, 422: {"model": ErrorEnvelope, "description": "工具不支持该操作（如不可入库）"}, 503: {"model": ErrorEnvelope, "description": "LLM 不可用"}, } _INGEST_ERRORS: dict[int | str, dict[str, Any]] = { **_TOOL_ERRORS, 409: {"model": ErrorEnvelope, "description": "存在未裁决 continuity 冲突"}, } ProjectRepoDep = Annotated[ProjectRepo, Depends(get_project_repo)] MemoryReposDep = Annotated[MemoryRepos, Depends(get_memory_repos)] OutlineReadRepoDep = Annotated[OutlineRepo, Depends(get_outline_read_repo)] ChapterRepoDep = Annotated[ChapterRepo, Depends(get_chapter_repo)] WorldWriteRepoDep = Annotated[WorldEntityWriteRepo, Depends(get_world_entity_write_repo)] OutlineWriteRepoDep = Annotated[OutlineWriteRepo, Depends(get_outline_write_repo)] RuleWriteRepoDep = Annotated[RuleWriteRepo, Depends(get_rule_write_repo)] GatewayBuilderDep = Annotated[TierGatewayBuilder, Depends(get_tier_gateway_builder)] SessionDep = Annotated[AsyncSession, Depends(get_session)] def _descriptor_view(tool: GeneratorTool) -> ToolDescriptorView: return ToolDescriptorView( key=tool.key, title=tool.title, subtitle=tool.subtitle, genre=tool.genre, is_legacy=tool.spec is None, ingestable=tool.ingest is not None, input_fields=[ ToolInputFieldView( name=f.name, label=f.label, type=f.type, required=f.required, default=f.default, help=f.help, ) for f in tool.input_fields ], legacy_route=tool.legacy_route, ) @router.get("/skills/toolbox") async def list_toolbox() -> ToolboxListResponse: """创作工具箱全量描述符（legacy + 新工具；前端据此渲染卡片栅格 + 表单 + 路由）。""" from ww_skills import TOOLBOX # 确定性顺序（按 key 升序），便于前端稳定渲染与测试断言。 tools = [_descriptor_view(TOOLBOX[key]) for key in sorted(TOOLBOX)] return ToolboxListResponse(tools=tools) def _require_real_tool(tool_key: str) -> GeneratorTool: """解析工具：未知 → 404；legacy（无 spec）→ 404（前端走 legacy_route）。""" tool = get_tool(tool_key) if tool is None: raise AppError(ErrorCode.NOT_FOUND, f"未知工具：{tool_key}", {"tool_key": tool_key}) if tool.spec is None: raise AppError( ErrorCode.NOT_FOUND, f"工具 {tool_key} 无通用执行路径，请使用其专属页面（legacy_route）", {"tool_key": tool_key, "legacy_route": tool.legacy_route}, ) return tool async def _chapter_beats( outline_repo: OutlineRepo, project_id: uuid.UUID, chapter_no: int ) -> list[str]: """读指定章的大纲粗节拍（缺章/缺大纲 → 空 list，不报错；DB JSONB dict→list 解包）。""" view = await outline_repo.get(project_id, chapter_no) if view is None: return [] raw = view.beats or {} return list(raw.get("beats", [])) if isinstance(raw, dict) else [] async def _prior_chapter_text( chapter_repo: ChapterRepo, project_id: uuid.UUID, chapter_no: int ) -> str: """读续写承接的前文正文：优先该章最新 accepted，无则草稿，皆无 → 空串（端点降级）。 core 的 builder 不 import apps/api——正文在端点读 chapter 领域表后经参数注入 （仿链 accept_op 的注入模式），守不变量 #1（DB 为真相源）。 """ accepted = await chapter_repo.latest_accepted(project_id, chapter_no) if accepted is not None and accepted.content.strip(): return accepted.content draft = await chapter_repo.get_draft(project_id, chapter_no) if draft is not None and draft.content.strip(): return draft.content return "" @router.post("/projects/{project_id}/skills/{tool_key}/generate", responses=_TOOL_ERRORS) async def generate_with_tool( project_id: uuid.UUID, tool_key: str, body: ToolGenerateRequest, request: Request, project_repo: ProjectRepoDep, memory: MemoryReposDep, outline_repo: OutlineReadRepoDep, chapter_repo: ChapterRepoDep, build_gateway: GatewayBuilderDep, session: SessionDep, ) -> ToolGeneratePreviewResponse: """通用生成（结构化预览，不入库）。未知/ legacy 工具→404；项目不存在→404；无凭据→503。""" request_id = getattr(request.state, "request_id", None) tool = _require_real_tool(tool_key) spec = tool.spec assert spec is not None # _require_real_tool 已保证（mypy 收窄） project = await project_repo.get(STUB_OWNER_ID, project_id) if project is None: raise AppError(ErrorCode.NOT_FOUND, f"project not found: {project_id}") # brief_only 给精简设定、其余给完整设定（确定性序列化，无时间戳/UUID）。 if tool.context_strategy == "brief_only": project_context = _project_context(project.title, project.genre, None, None) else: project_context = _project_context( project.title, project.genre, project.premise, project.theme ) world_context = "" beats: list[str] = [] prior_text = "" text = body.text or "" if tool.context_strategy == "with_world": world_context = await _world_context(memory, project_id) elif tool.context_strategy == "with_outline_chapter": chapter_no = body.chapter_no or 1 beats = await _chapter_beats(outline_repo, project_id, chapter_no) elif tool.context_strategy == "with_prior_chapter": # 续写：端点先读该章最新 accepted/draft 正文（DB 为真相源，#1）+ 该章节拍，注入 builder。 chapter_no = body.chapter_no or 1 prior_text = await _prior_chapter_text(chapter_repo, project_id, chapter_no) beats = await _chapter_beats(outline_repo, project_id, chapter_no) elif tool.context_strategy == "text_input": # text_input 工具（扩写/降AI/拆书）必须有原文样本——空原文会让 LLM 空跑（守输入校验纪律）。 if not text.strip(): raise AppError( ErrorCode.VALIDATION, f"工具 {tool_key} 需要原文输入（text 不可为空）", {"tool_key": tool_key}, ) # 拆书的书名（body.kind）前置入原文，供拆解对照（teardown system prompt 期待书名）。 kind = (body.kind or "").strip() if kind: text = f"书名：{kind}\n\n{text}" context = build_toolbox_context( tool.context_strategy, brief=body.brief, project_context=project_context, world_context=world_context, chapter_no=body.chapter_no, beats=beats, prior_text=prior_text, text=text, ) gateway = await build_gateway(spec.tier) parsed = await run_generator( spec, context=context, gateway=gateway, user_id=STUB_OWNER_ID, project_id=project_id, ) # 提交边界：预览不写业务表，但网关 ledger 只 flush → 末尾 commit 落 usage（否则丢失）。 await session.commit() log.info( "toolbox_generate_done", tool_key=tool_key, project_id=str(project_id), request_id=request_id, output_kind=type(parsed).__name__, ) return ToolGeneratePreviewResponse( tool_key=tool_key, output_kind=type(parsed).__name__, preview=parsed.model_dump(), ) async def _precheck_world_entities( spec: AgentSpec, *, items: list[WorldEntityCardView], world_context: str, gateway: Gateway, project_id: uuid.UUID, ) -> list[Conflict]: """入库前 continuity 预检（world_entities）：把待入库实体 vs 既有世界观真相源比对。 仿 `precheck_generated_cards` 但服务于世界观实体（金手指/词条）。复用 `continuity_spec` （analyst 档，只读，不变量 #3）——直接构 LlmRequest 走网关，节点逻辑无 LLM 非确定性。 """ lines = [ f"- [{e.type}] {e.name}：{('；'.join(e.rules) if e.rules else '（无硬规则）')}" for e in items ] context = ( "## 待校验：本次生成的世界观实体（尚未入库）\n" f"{chr(10).join(lines) or '（无）'}\n\n" "## 真相源：既有世界观硬规则\n" f"{world_context or '（暂无世界观设定）'}" ) req = LlmRequest( tier=spec.tier, system=[Block(text=spec.system_prompt, cache=True)], input=context, output_schema=spec.output_schema, scope=Scope(user_id=STUB_OWNER_ID, project_id=project_id), ) resp = await gateway.run(req) parsed = resp.parsed if not isinstance(parsed, ContinuityReview): raise ValueError("gateway returned no parsed ContinuityReview for toolbox precheck") return list(parsed.conflicts) def _conflict_409(conflicts: list[Conflict]) -> AppError: return AppError( ErrorCode.CONFLICT_UNRESOLVED, "生成内容与现有设定存在 continuity 冲突，请裁决后确认入库", { "conflicts": [ IngestConflictView( type=c.type, where=c.where, refs=list(c.refs), suggestion=c.suggestion ).model_dump() for c in conflicts ], "conflict_count": len(conflicts), }, ) @router.post( "/projects/{project_id}/skills/{tool_key}/ingest", status_code=201, responses=_INGEST_ERRORS ) async def ingest_with_tool( project_id: uuid.UUID, tool_key: str, body: ToolIngestRequest, request: Request, project_repo: ProjectRepoDep, memory: MemoryReposDep, world_write_repo: WorldWriteRepoDep, outline_write_repo: OutlineWriteRepoDep, rule_write_repo: RuleWriteRepoDep, build_gateway: GatewayBuilderDep, session: SessionDep, ) -> ToolIngestResponse: """通用入库（仅可入库工具）：过 continuity 预检 + 白名单 → 写 ingest.table。 未知/legacy 工具→404；不可入库工具→422；有冲突且未确认→409；越权写表丢弃+审计。 """ request_id = getattr(request.state, "request_id", None) tool = _require_real_tool(tool_key) spec = tool.spec assert spec is not None if tool.ingest is None: raise AppError( ErrorCode.VALIDATION, f"工具 {tool_key} 不支持入库（纯预览）", {"tool_key": tool_key}, ) project = await project_repo.get(STUB_OWNER_ID, project_id) if project is None: raise AppError(ErrorCode.NOT_FOUND, f"project not found: {project_id}") table = tool.ingest.table if table == "world_entities": return await _ingest_world_entities( spec, body=body, project_id=project_id, memory=memory, world_write_repo=world_write_repo, session=session, build_gateway=build_gateway, request_id=request_id, tool_key=tool_key, ) if table == "outline": return await _ingest_outline( spec, body=body, project_id=project_id, outline_write_repo=outline_write_repo, session=session, request_id=request_id, tool_key=tool_key, ) if table == "rules": return await _ingest_rules( spec, body=body, project_id=project_id, rule_write_repo=rule_write_repo, session=session, request_id=request_id, tool_key=tool_key, ) # 描述符的 IngestSpec.table 已限白名单；其余表本期通用入库未实现。 raise AppError(ErrorCode.VALIDATION, f"入库表 {table} 暂未支持通用入库", {"table": table}) async def _ingest_world_entities( spec: AgentSpec, *, body: ToolIngestRequest, project_id: uuid.UUID, memory: MemoryRepos, world_write_repo: WorldEntityWriteRepo, session: AsyncSession, build_gateway: TierGatewayBuilder, request_id: str | None, tool_key: str, ) -> ToolIngestResponse: """world_entities 入库（金手指/词条）：continuity 预检 + 白名单 → 写 world_entities。""" items = body.world_entities # gate 1：入库前 continuity 预检（analyst 网关；编排器追加的检查，守不变量 #1/#3）。 world_context = await _world_context(memory, project_id) precheck_gateway = await build_gateway("analyst") conflicts = await _precheck_world_entities( continuity_spec, items=items, world_context=world_context, gateway=precheck_gateway, project_id=project_id, ) if conflicts and not body.acknowledge_conflicts: await session.commit() # 落 precheck 网关 usage；不写业务表。 log.info( "toolbox_ingest_blocked_by_conflicts", tool_key=tool_key, project_id=str(project_id), request_id=request_id, conflict_count=len(conflicts), ) raise _conflict_409(conflicts) # gate 2：权限白名单——按 spec 声明的 writes 过滤（越权表丢弃 + 审计）。 allowed, rejected = partition_writes(spec, {"world_entities": items}) if rejected: log.warning( "toolbox_ingest_rejected_over_permission", tool_key=tool_key, project_id=str(project_id), request_id=request_id, rejected_tables=rejected, ) fields = [ WorldEntityFields(type=entity.type, name=entity.name, rules=list(entity.rules)) for entity in allowed.get("world_entities", []) ] # 批量插入（一次 flush 落全部），保序返回逐条 View。 created = [view.name for view in await world_write_repo.create_many(project_id, fields)] await session.commit() log.info( "toolbox_ingest_done", tool_key=tool_key, project_id=str(project_id), request_id=request_id, table="world_entities", created_count=len(created), ) return ToolIngestResponse(table="world_entities", created=created, rejected_tables=rejected) async def _ingest_outline( spec: AgentSpec, *, body: ToolIngestRequest, project_id: uuid.UUID, outline_write_repo: OutlineWriteRepo, session: AsyncSession, request_id: str | None, tool_key: str, ) -> ToolIngestResponse: """outline 入库（细纲）：白名单 → 把场景节拍 upsert 到目标章的 outline 行。 细纲不做 continuity 预检（场景是粗节拍的展开，非与世界观真相源比对的设定/角色）； 仍过 partition_writes 白名单。无章号 → VALIDATION。 """ if body.chapter_no is None: raise AppError(ErrorCode.VALIDATION, "细纲入库需指定 chapter_no", {"tool_key": tool_key}) # gate：权限白名单（细纲 spec 声明 writes=["outline"]）。 scenes: list[OutlineSceneIngestView] = body.scenes allowed, rejected = partition_writes(spec, {"outline": scenes}) if rejected: log.warning( "toolbox_ingest_rejected_over_permission", tool_key=tool_key, project_id=str(project_id), request_id=request_id, rejected_tables=rejected, ) allowed_scenes: list[OutlineSceneIngestView] = allowed.get("outline", []) # 场景节拍序列化为该章 beats（按 idx 升序拼装，确定性）。 ordered = sorted(allowed_scenes, key=lambda s: s.idx) beats = [s.beat for s in ordered] created: list[str] = [] if "outline" in allowed: view = await outline_write_repo.upsert_chapter( project_id, volume=1, chapter_no=body.chapter_no, beats=beats, foreshadow_windows=[], ) created = [str(s.idx) for s in ordered] log.info( "toolbox_outline_upserted", tool_key=tool_key, project_id=str(project_id), chapter_no=view.chapter_no, scene_count=len(ordered), ) await session.commit() log.info( "toolbox_ingest_done", tool_key=tool_key, project_id=str(project_id), request_id=request_id, table="outline", created_count=len(created), ) return ToolIngestResponse(table="outline", created=created, rejected_tables=rejected) # 拆书入库的 rules 级别（meta 规则非设定卡，落项目级；单用户原型 owner stub）。 _TEARDOWN_RULE_LEVEL = "project" def _flatten_teardown_rules(teardown: TeardownIngestView) -> list[tuple[str, str]]: """把拆书结论拍平成可读 rules 条目：返回 (label, content) 列表（确定性顺序，跳过空字段）。 主题/原型/钩子各一条（清单拼成可读文本）；结构一条。空清单/空串的字段跳过，不落空规则。 """ entries: list[tuple[str, str]] = [] if teardown.themes: entries.append(("themes", f"核心主题：{'；'.join(teardown.themes)}")) if teardown.archetypes: entries.append(("archetypes", f"人物原型：{'；'.join(teardown.archetypes)}")) if teardown.structure.strip(): entries.append(("structure", f"叙事结构：{teardown.structure.strip()}")) if teardown.hooks: entries.append(("hooks", f"钩子套路：{'；'.join(teardown.hooks)}")) return entries async def _ingest_rules( spec: AgentSpec, *, body: ToolIngestRequest, project_id: uuid.UUID, rule_write_repo: RuleWriteRepo, session: AsyncSession, request_id: str | None, tool_key: str, ) -> ToolIngestResponse: """rules 入库（拆书）：白名单 → 把拆书结论拍平为可读 rules 条目写入项目级 rules。 meta 规则非设定卡，无需 continuity 预检（仿 outline 无 409 路径）；仍过 partition_writes 白名单（teardown spec 声明 writes=["rules"]）。无 teardown 产物 → VALIDATION。 """ if body.teardown is None: raise AppError(ErrorCode.VALIDATION, "拆书入库需提供 teardown 产物", {"tool_key": tool_key}) entries = _flatten_teardown_rules(body.teardown) # gate：权限白名单（teardown spec 声明 writes=["rules"]）。 allowed, rejected = partition_writes(spec, {"rules": entries}) if rejected: log.warning( "toolbox_ingest_rejected_over_permission", tool_key=tool_key, project_id=str(project_id), request_id=request_id, rejected_tables=rejected, ) created: list[str] = [] for label, content in allowed.get("rules", []): await rule_write_repo.create(project_id, level=_TEARDOWN_RULE_LEVEL, content=content) created.append(label) await session.commit() log.info( "toolbox_ingest_done", tool_key=tool_key, project_id=str(project_id), request_id=request_id, table="rules", created_count=len(created), ) return ToolIngestResponse(table="rules", created=created, rejected_tables=rejected)