feat: T6.1 创作工具箱通用生成器框架地基 — 脑洞生成器 + run_generator

- schemas: Idea/IdeaListResult（脑洞生成器结构化产出，纯预览不入库）
- specs: brainstorm_spec（light 档，reads=[projects] writes=[]）
- orchestrator: build_brief_context（brief_only 策略）+ 通用 run_generator
  （据 spec.output_schema 校验 parsed，不绑定具体类型，任一声明式生成器共用）
- 新增 13 测；两包 ruff/format/mypy 干净、252 passed 无回归

P1 第一波（@llm）：证实「加生成器=加一份声明」。@backend T6.2/T6.3 待续。

packages/agents/tests/test_brainstorm_spec.py

@@ -0,0 +1,82 @@
+"""T6 脑洞生成器契约测试：spec + schema（创作工具箱通用框架首个最简生成器）。
+
+契约测试——构造符合 schema 的 mock 响应，校验字段、tier、读写权限。
+brainstorm：`{ideas:[{premise,hook,genre_fit}]}`（轻量档，reads=projects，writes=[] 纯预览）。
+不联网、无 DB。
+"""
+
+from __future__ import annotations
+
+import pytest
+from pydantic import ValidationError
+from ww_agents import (
+    AgentSpec,
+    Idea,
+    IdeaListResult,
+    brainstorm_spec,
+)
+
+# ---- IdeaListResult schema ----
+
+
+def test_idea_list_parses_mock_response() -> None:
+    # Arrange：模拟网关 instructor 校验后的结构化脑洞产出
+    mock = {
+        "ideas": [
+            {
+                "premise": "废柴觉醒可吞噬规则的瞳术",
+                "hook": "每次升级都要付出一段记忆为代价",
+                "genre_fit": "玄幻·热血",
+            },
+            {"premise": "末世里唯一能种粮的农场主", "hook": "粮食成了最硬的通货"},
+        ]
+    }
+
+    # Act
+    result = IdeaListResult.model_validate(mock)
+
+    # Assert
+    assert len(result.ideas) == 2
+    assert result.ideas[0].premise == "废柴觉醒可吞噬规则的瞳术"
+    assert result.ideas[0].hook == "每次升级都要付出一段记忆为代价"
+    assert result.ideas[0].genre_fit == "玄幻·热血"
+    assert result.ideas[1].genre_fit is None  # 可缺
+
+
+def test_idea_list_defaults_to_empty() -> None:
+    assert IdeaListResult().ideas == []
+
+
+def test_idea_requires_premise_and_hook() -> None:
+    with pytest.raises(ValidationError):
+        Idea.model_validate({"premise": "只有前提没有钩子"})
+
+
+# ---- brainstorm_spec 声明 ----
+
+
+def test_brainstorm_spec_is_light_tier() -> None:
+    # 不变量 #2：脑洞发散用轻量档，只声明 tier
+    assert brainstorm_spec.tier == "light"
+    assert brainstorm_spec.name == "brainstorm"
+
+
+def test_brainstorm_spec_reads_projects_writes_nothing() -> None:
+    # 纯预览：读作品设定、不写任何业务表（不变量 #3）
+    assert brainstorm_spec.reads == ["projects"]
+    assert brainstorm_spec.writes == []
+
+
+def test_brainstorm_spec_output_schema() -> None:
+    assert brainstorm_spec.output_schema is IdeaListResult
+
+
+def test_brainstorm_spec_prompt_documents_diversification() -> None:
+    # 一次多条须逐条差异化——system_prompt 必须强调
+    assert "差异化" in brainstorm_spec.system_prompt
+
+
+def test_brainstorm_spec_is_agent_spec_and_immutable() -> None:
+    assert isinstance(brainstorm_spec, AgentSpec)
+    with pytest.raises(ValidationError):
+        brainstorm_spec.tier = "writer"

packages/agents/ww_agents/__init__.py

@@ -15,6 +15,8 @@ from .schemas import (
     ForeshadowReview,
     ForeshadowSuggestion,
     ForeshadowWindow,
+    Idea,
+    IdeaListResult,
     OutlineChapter,
     OutlineResult,
     PaceIssue,
@@ -28,6 +30,7 @@ from .schemas import (
 )
 from .specs import (
     AgentSpec,
+    brainstorm_spec,
     character_gen_spec,
     continuity_spec,
     foreshadow_spec,
@@ -50,6 +53,8 @@ __all__ = [
     "ForeshadowReview",
     "ForeshadowSuggestion",
     "ForeshadowWindow",
+    "Idea",
+    "IdeaListResult",
     "OutlineChapter",
     "OutlineResult",
     "PaceIssue",
@@ -60,6 +65,7 @@ __all__ = [
     "StyleFingerprintResult",
     "WorldEntityCard",
     "WorldGenResult",
+    "brainstorm_spec",
     "character_gen_spec",
     "continuity_spec",
     "foreshadow_spec",

packages/agents/ww_agents/schemas.py

@@ -298,3 +298,30 @@ class CharacterGenResult(BaseModel):
         default_factory=list,
         description="生成的角色卡清单；无则空列表",
     )
+
+
+# ---- 创作工具箱 · 脑洞生成器结构化输出（T6 通用生成器框架 · brainstorm）----
+
+
+class Idea(BaseModel):
+    """单条脑洞：前提 + 钩子 + 适配题材（创作工具箱最简生成器产物）。
+
+    纯预览产物——不映射任何业务表、不入库（`brainstorm_spec.writes=[]`，不变量 #3）。
+    `genre_fit` 可缺（作者未限定赛道时由模型留空）。
+    """
+
+    premise: str = Field(description="一句话脑洞/设定前提")
+    hook: str = Field(description="抓人钩子：为何让读者想追读")
+    genre_fit: str | None = Field(default=None, description="适配题材/赛道（可缺）")
+
+
+class IdeaListResult(BaseModel):
+    """脑洞生成器结构化产出：一组脑洞（T6 · 通用生成器框架首个最简生成器）。
+
+    声明式只读（不变量 #3）：纯预览，端点同步返回，不写任何业务表（仅记 ledger）。
+    """
+
+    ideas: list[Idea] = Field(
+        default_factory=list,
+        description="生成的脑洞清单；无则空列表",
+    )

packages/agents/ww_agents/specs.py

@@ -16,6 +16,7 @@ from .schemas import (
     CharacterGenResult,
     ContinuityReview,
     ForeshadowReview,
+    IdeaListResult,
     OutlineResult,
     PaceReview,
     StyleDriftReview,
@@ -367,3 +368,36 @@ character_gen_spec = AgentSpec(
     writes=["characters"],  # 声明式（真写库经 T5.2 入库端点 + continuity 校验，不变量 #3）
     scope="builtin",
 )
+
+
+# ---- brainstorm（脑洞生成器；轻量档；创作工具箱通用框架首个最简生成器，T6）----
+
+BRAINSTORM_SYSTEM_PROMPT = """你是中文网文的「脑洞生成器」（brainstorm，轻量档）。\
+职责：依据作品立意/题材与作者一句话需求，发散产出一组**差异化**的故事脑洞，\
+每条给出一句话前提 + 抓人钩子 + 适配题材，供作者快速选种立项。
+
+输入材料：
+- 作品设定（projects：题材、立意、主线、卖点；新立项时可能很少）；
+- 作者的一句话需求/方向（可空——空则按题材自由发散）。
+
+产出（每条一个 idea）：
+- premise：一句话脑洞/设定前提（核心创意，独立成立）；
+- hook：抓人钩子——为何让读者想追读（爽点/反差/悬念）；
+- genre_fit：适配题材/赛道（若需求/设定已限定则贴合；否则可留空）。
+
+纪律：
+- 一次产出多条时**逐条差异化**——不同切入角度/赛道/爽点，避免一个模子；
+- 贴合作品立意与题材；需求为空时按题材发散，不臆造与题材无关的设定；
+- 你只产结构化脑洞清单，**不改稿、不写库**（纯预览，不变量 #3）。"""
+
+
+brainstorm_spec = AgentSpec(
+    name="brainstorm",
+    tier="light",  # 不变量 #2：只声明档位，不写 model（脑洞发散用轻量档）
+    system_prompt=BRAINSTORM_SYSTEM_PROMPT,
+    input_schema=None,  # 注入材料为序列化文本（作品设定 + 一句话需求），非结构化入参
+    output_schema=IdeaListResult,
+    reads=["projects"],
+    writes=[],  # 纯预览，不写库（不变量 #3）
+    scope="builtin",
+)

packages/core/tests/test_generator_node.py

@@ -0,0 +1,91 @@
+"""T6 通用生成器执行器单测：`run_generator` + `build_brief_context`。
+
+注入 mock 网关（产 IdeaListResult parsed），无真 LLM、无真 Postgres。通用执行器据
+`spec.output_schema` 校验 parsed（不绑定具体类型）；只产结构化产物，**不写库**（纯预览，
+不变量 #3）。asyncio_mode=auto。
+"""
+
+from __future__ import annotations
+
+import uuid
+
+import pytest
+from fakes_orchestrator import FailingRunGateway, FakeRunGateway
+from ww_agents import Idea, IdeaListResult, brainstorm_spec
+from ww_core.orchestrator import build_brief_context, run_generator
+
+PROJECT = uuid.UUID("00000000-0000-0000-0000-000000000001")
+USER = uuid.UUID("00000000-0000-0000-0000-0000000000aa")
+
+_IDEAS = IdeaListResult(
+    ideas=[
+        Idea(premise="废柴觉醒吞噬瞳", hook="升级要付记忆", genre_fit="玄幻"),
+        Idea(premise="末世种粮农场主", hook="粮食成硬通货"),
+    ]
+)
+
+
+# ---- run_generator：返回结构化产物，据 output_schema 校验，只读不写库 ----
+
+
+async def test_run_generator_returns_structured_result() -> None:
+    gateway = FakeRunGateway(_IDEAS)
+
+    result = await run_generator(
+        brainstorm_spec,
+        context=build_brief_context(brief="来点逆袭流", project_context="题材：玄幻"),
+        gateway=gateway,
+        user_id=USER,
+        project_id=PROJECT,
+    )
+
+    assert isinstance(result, IdeaListResult)
+    assert len(result.ideas) == 2
+    assert result.ideas[0].premise == "废柴觉醒吞噬瞳"
+
+
+async def test_run_generator_passes_tier_and_schema_to_gateway() -> None:
+    # 不变量 #2：只透传档位（light），不传 model；output_schema 据 spec 绑定
+    gateway = FakeRunGateway(_IDEAS)
+
+    await run_generator(
+        brainstorm_spec,
+        context="x",
+        gateway=gateway,
+        user_id=USER,
+        project_id=PROJECT,
+    )
+
+    assert gateway.last_request is not None
+    assert gateway.last_request.tier == "light"
+    assert gateway.last_request.output_schema is IdeaListResult
+
+
+async def test_run_generator_raises_on_gateway_failure() -> None:
+    # 独立生成：网关失败直接上抛（端点处理），不静默吞
+    gateway = FailingRunGateway(RuntimeError("boom"))
+
+    with pytest.raises(RuntimeError, match="boom"):
+        await run_generator(
+            brainstorm_spec,
+            context="x",
+            gateway=gateway,
+            user_id=USER,
+            project_id=PROJECT,
+        )
+
+
+# ---- build_brief_context：确定性、空需求降级占位 ----
+
+
+def test_build_brief_context_is_deterministic() -> None:
+    a = build_brief_context(brief="逆袭", project_context="题材：玄幻")
+    b = build_brief_context(brief="逆袭", project_context="题材：玄幻")
+    assert a == b
+    assert "题材：玄幻" in a
+    assert "逆袭" in a
+
+
+def test_build_brief_context_empty_brief_falls_back() -> None:
+    ctx = build_brief_context(brief="   ", project_context="题材：都市")
+    assert "按题材自由发散" in ctx

packages/core/ww_core/orchestrator/__init__.py

@@ -21,11 +21,13 @@ from .collect import (
     extract_style,
 )
 from .generation_node import (
+    build_brief_context,
     build_character_gen_context,
     build_precheck_context,
     build_worldbuilder_context,
     precheck_generated_cards,
     run_character_gen,
+    run_generator,
     run_worldbuilder,
 )
 from .graph import (
@@ -102,6 +104,7 @@ __all__ = [
     "GatewayStream",
     "ReviewRecorder",
     "SseEvent",
+    "build_brief_context",
     "build_character_gen_context",
     "build_outline_request",
     "build_precheck_context",
@@ -128,6 +131,7 @@ __all__ = [
     "pace_event",
     "precheck_generated_cards",
     "run_character_gen",
+    "run_generator",
     "run_outline",
     "run_review",
     "run_style_extraction",

packages/core/ww_core/orchestrator/generation_node.py

@@ -28,6 +28,7 @@ from collections.abc import Sequence
 from typing import Protocol
 
 import structlog
+from pydantic import BaseModel
 from ww_agents import (
     AgentSpec,
     CharacterCard,
@@ -73,6 +74,49 @@ def _build_request(
     )
 
 
+# ---- 通用生成器执行器（T6 · 创作工具箱通用框架）----
+
+
+def build_brief_context(*, brief: str, project_context: str) -> str:
+    """通用「brief_only」上下文：作品设定 + 作者一句话需求（确定性，无时间戳）。
+
+    覆盖最简生成器（脑洞/书名…）：仅需作品设定 + 一句话方向。需求为空时给降级占位，
+    交由 system_prompt 的「按题材自由发散」纪律处理。
+    """
+    brief_block = brief.strip() or "（作者未填具体方向，按题材自由发散）"
+    return f"## 作品设定\n{project_context or '（暂无设定）'}\n\n## 创作需求\n{brief_block}"
+
+
+async def run_generator(
+    spec: AgentSpec,
+    *,
+    context: str,
+    gateway: GatewayRun,
+    user_id: uuid.UUID,
+    project_id: uuid.UUID,
+) -> BaseModel:
+    """通用生成器执行器（T6）：构请求 → `gateway.run` → 按 `spec.output_schema` 校验 parsed。
+
+    任一声明式生成器共用此路径（不绑定具体产物类型，据 spec 的 output_schema 校验）。
+    **只读、不写库**（纯预览，不变量 #3）；独立生成——网关失败直接上抛（端点处理），不静默吞。
+    """
+    expected = spec.output_schema
+    if expected is None:
+        raise ValueError(f"generator spec {spec.name!r} has no output_schema")
+    req = _build_request(spec, context=context, user_id=user_id, project_id=project_id)
+    resp = await gateway.run(req)
+    parsed = resp.parsed
+    if not isinstance(parsed, expected):
+        log.error(
+            "generator_node_missing_parsed",
+            spec=spec.name,
+            project_id=str(project_id),
+            parsed_type=type(parsed).__name__,
+        )
+        raise ValueError(f"gateway returned no parsed {expected.__name__} for {spec.name} request")
+    return parsed
+
+
 # ---- worldbuilder ----