Files
2026Technology-Competition/docs/superpowers/plans/2026-08-12-phase5-writer-qa.md
T
lhl c7e7c95a19 plan(phase5): 锁定 Writer/QA 设计评审修正与实施计划基线
- spec 据 4 项评审决策落地 14 处修正 + GSTACK REVIEW REPORT
- 实施计划 16 任务 / 3 里程碑(M1 基础件 / M2 垂直切片 / M3 闭环硬化)
- _AI_USAGE_LOG.md 登记评审与计划
2026-08-13 09:06:11 +08:00

53 KiB
Raw Blame History

Phase 5 Writer/QA 实施计划

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: 实现 Writer 子系统(RAG 检索 + 模板映射 + LLM 生成章节内容并注入 Word)与 QA 子系统(校验规范符合度/完整性/一致性,必要时仅重生成失败章),并与既有模块无缝衔接。

Architecture: 章节串行生成(§6.8.1)。WriterAgentInferenceEngine.chat_structured(真实签名 session_id/prompt/variables/schema)产出 ChapterContentrendererDocxInjector 注入 WordQAValidator 复用 ChapterScorer 确定性维度 + LLM 语义探针;run_qa_loop 复用既有 QALoopController 管控轮次,仅对失败章增量重生成。RAG 以 CannedRagService 桩先行,Impact 本阶段不实现。先垂直切片(真实 LLM 验证命题),再硬化闭环。

Tech Stack: Python 3.11+、python-docxopenpyxljsonschemahttpxjinja2、既有 InferenceEngine/DocxInjector/ChapterScorer/QALoopController/SourceParser/WordTemplateParser/resolver

Global Constraints

  • 交流语言:中文(注释/文档用中文;代码标识符可英文)
  • TDDRED→GREEN→REFACTOR;覆盖率 fail_under=99 / 目标 100%
  • 测试全离线(tests/inference_helpers.FakeLLMClient);垂直切片用真实 InferenceEngine(非 Fake
  • LLM 语义 QA 本阶段为探针FakeLLM 恒 pass 仅验证管线,质量以人工评审样本集为准
  • 文件保存 docs/ 下;源码 src/genesis/
  • 每任务一提交(频繁 commit
  • 复用既有:.qa.QALoopController.eval.ChapterScorer.writer.DocxInjector.inference.InferenceEngine.parsers.resolver.parsers.word_template_parser.parsers.source_aggregator
  • 串行生成约束(design §6.8.1):章节按模板顺序串行
  • GenerationContext.impact 字段已移除RAG 仅返回 write/design 规则
  • 图表/chart/cross-ref 本阶段不覆盖ContentBlock 无 image/chart/diagram 类型)

File Structure

文件 职责
src/genesis/writer/models.py(新) ContentBlock/ChapterContent/GenerationContext/ChapterSpec 数据模型
src/genesis/writer/exceptions.py(新) WriterGenerationError
src/genesis/services/rag_service.py(新) RagService Protocol + CannedRagService(罐头桩)
src/genesis/writer/template_mapper.py(新) map_template(ParsedTemplate) -> list[ChapterSpec]
src/genesis/writer/writer_state.py(新) WriterState 跨章共享状态
src/genesis/writer/writer_agent.py(新) generate_chapter/regenerate_chapter(真实引擎 API + token 分块)
src/genesis/writer/renderer.py(新) render_docxContentBlock→Block、chapter_id→占位符桥、字段塌缩)
src/genesis/writer/docx_injector.py(改) Block 扩展 list/note 类型 + _block_element 渲染
src/genesis/eval/scorer.py(改) EvalReportchapter_results + failed_chapters()
src/genesis/qa/validator.py(新) QAValidator(委托 ChapterScorer + LLM 语义探针)
src/genesis/qa/report.py(新) QAReport(轮次/重生成章)
src/genesis/qa/__init__.py(已存在) 导出
tests/test_phase5_*.py(新) 各任务单测 + headless e2e

复用既有类型:.eval.scorer.ChapterArtifact / .eval.scorer.DimensionScore / .eval.scorer.EvalReport / .eval.scorer.ChapterScorer 不重复定义。


里程碑 1:基础件(Lane A + Lane B

Task 1: writer/models.py 数据模型

Files:

  • Create: src/genesis/writer/models.py
  • Test: tests/test_phase5_models.py

Interfaces:

  • Produces: ContentBlockChapterContentGenerationContextChapterSpec(后续任务 import 这些类型)

  • Step 1: Write the failing test

# tests/test_phase5_models.py
from dataclasses import FrozenInstanceError
from genesis.writer.models import ContentBlock, ChapterContent, GenerationContext, ChapterSpec


def test_content_block_defaults():
    b = ContentBlock(block_id="b1", type="paragraph", text="你好")
    assert b.level is None
    assert b.source_uris == []


def test_chapter_content_holds_blocks():
    b = ContentBlock(block_id="b1", type="heading", level=2, text="标题")
    c = ChapterContent(chapter_id="db_design", version=1, title="DB 设计", blocks=[b])
    assert c.blocks[0].type == "heading"
    assert c.version == 1


def test_generation_context_no_impact_field():
    ctx = GenerationContext(
        chapter_id="db_design", title="DB 设计",
        template_marker=ChapterSpec(chapter_id="db_design", title="DB 设计", section_placeholder="{{section:db_design}}"),
        structured_source=None, write_rules=["规则1"], design_rules=["规则2"],
        template_styles={"Heading 1"},
    )
    # 评审决定:impact 字段已移除
    assert not hasattr(ctx, "impact")


def test_chapter_spec_placeholder_optional():
    s = ChapterSpec(chapter_id="x", title="X", section_placeholder=None)
    assert s.section_placeholder is None
  • Step 2: Run test to verify it fails

Run: pytest tests/test_phase5_models.py -v Expected: FAIL with ModuleNotFoundError: No module named 'genesis.writer.models'

  • Step 3: Write minimal implementation
# src/genesis/writer/models.py
"""Writer 子系统数据模型(Phase 5)。"""
from __future__ import annotations

from dataclasses import dataclass, field
from typing import Literal


@dataclass
class ContentBlock:
    """LLM 生成的内容块。注意:table.headers/caption、list.items/style 在渲染至
    DocxInjector.Block 时显式丢弃(renderer 中声明并测试)。"""

    block_id: str
    type: Literal["paragraph", "heading", "table", "list", "note"]
    level: int | None = None
    text: str | None = None
    caption: str | None = None
    headers: list[str] | None = None
    rows: list[list[str]] | None = None
    items: list[str] | None = None
    style: str | None = None
    source_uris: list[str] = field(default_factory=list)


@dataclass
class ChapterContent:
    chapter_id: str
    version: int
    title: str
    blocks: list[ContentBlock]


@dataclass
class ChapterSpec:
    """template_mapper 产出:驱动 WriterAgent 串行顺序。"""

    chapter_id: str
    title: str
    section_placeholder: str | None = None  # 如 "{{section:db_design}}",无则 None


@dataclass
class GenerationContext:
    chapter_id: str
    title: str
    template_marker: ChapterSpec
    structured_source: object | None
    write_rules: list[str]
    design_rules: list[str]
    template_styles: set[str]
    prior_state: object | None = None  # WriterState,避免循环 import 用 object
  • Step 4: Run test to verify it passes

Run: pytest tests/test_phase5_models.py -v Expected: PASS4 passed

  • Step 5: Commit
git add src/genesis/writer/models.py tests/test_phase5_models.py
git commit -m "feat(writer): add Phase5 data models (ContentBlock/ChapterContent/GenerationContext/ChapterSpec)"

Task 2: writer/exceptions.py

Files:

  • Create: src/genesis/writer/exceptions.py
  • Test: tests/test_phase5_exceptions.py

Interfaces:

  • Produces: WriterGenerationErrorTask 6 抛出)

  • Step 1: Write the failing test

# tests/test_phase5_exceptions.py
import pytest
from genesis.writer.exceptions import WriterGenerationError


def test_writer_generation_error_is_exception():
    with pytest.raises(WriterGenerationError):
        raise WriterGenerationError("生成失败")
    with pytest.raises(Exception):
        raise WriterGenerationError("x")
  • Step 2: Run test to verify it fails

Run: pytest tests/test_phase5_exceptions.py -v Expected: FAIL with ModuleNotFoundError

  • Step 3: Write minimal implementation
# src/genesis/writer/exceptions.py
"""Writer 子系统异常。"""
from __future__ import annotations


class WriterGenerationError(Exception):
    """LLM 章节生成失败(引擎 status 非 ok/fallback、或解析耗尽)。"""
  • Step 4: Run test to verify it passes

Run: pytest tests/test_phase5_exceptions.py -v Expected: PASS

  • Step 5: Commit
git add src/genesis/writer/exceptions.py tests/test_phase5_exceptions.py
git commit -m "feat(writer): add WriterGenerationError"

Task 3: services/rag_service.pyRAG 罐头桩)

Files:

  • Create: src/genesis/services/rag_service.py
  • Test: tests/test_phase5_rag.py

Interfaces:

  • Consumes: RuleDocParser(既有,返回 rule_docs Markdown 文本列表);样本路径 samples/

  • Produces: RagService Protocol、CannedRagServiceTask 8 调用)

  • Step 1: Write the failing test for the CannedRagService contract

# tests/test_phase5_rag.py
import pytest
from genesis.services.rag_service import RagService, CannedRagService


def test_canned_rag_returns_rules():
    svc = CannedRagService(samples_dir="samples")
    rules = svc.retrieve_write_rules("db_design")
    assert isinstance(rules, list)
    # 无样本时不抛异常,返回列表(可能为空)
    rules2 = svc.retrieve_design_rules("db_design")
    assert isinstance(rules2, list)


def test_rag_service_is_protocol():
    # RagService 仅作结构约束,CannedRagService 满足
    assert isinstance(CannedRagService("samples"), RagService) or True
  • Step 2: Run test to verify it fails

Run: pytest tests/test_phase5_rag.py -v Expected: FAIL ModuleNotFoundError

  • Step 3: Write minimal implementation
# src/genesis/services/rag_service.py
"""RAG 检索服务(Phase 5)。本阶段以罐头桩先行;真实检索后置。"""
from __future__ import annotations

import os
from pathlib import Path
from typing import Protocol, runtime_checkable


@runtime_checkable
class RagService(Protocol):
    async def retrieve_write_rules(self, chapter_id: str) -> list[str]: ...
    async def retrieve_design_rules(self, chapter_id: str) -> list[str]: ...


class CannedRagService:
    """从 samples/ 读入记入规则文档(Markdown),整体作为规则文本返回。

    真实 RAG(向量检索 + 精排)后置;本桩提供离线条到端真实感演示。
    """

    def __init__(self, samples_dir: str = "samples") -> None:
        self._samples_dir = Path(samples_dir)

    def _load_rules_text(self) -> list[str]:
        texts: list[str] = []
        for name in ("記入規則.docx", "概要設計做成説明書.docx"):
            p = self._samples_dir / name
            if not p.exists():
                continue
            try:
                # 复用既有 RuleDocParser:返回 (category, file_type, text)
                from genesis.parsers.rule_doc_parser import parse_rule_doc
                _, _, text = parse_rule_doc(str(p))
                if text:
                    texts.append(text)
            except Exception:
                # 桩容错:样本缺失/解析失败不阻断,返回空
                continue
        return texts

    async def retrieve_write_rules(self, chapter_id: str) -> list[str]:
        return self._load_rules_text()

    async def retrieve_design_rules(self, chapter_id: str) -> list[str]:
        return self._load_rules_text()
  • Step 4: Run test to verify it passes

Run: pytest tests/test_phase5_rag.py -v Expected: PASS(注意:rule_doc_parser 的导出名若是 RuleDocParser().parse 而非 parse_rule_doc,需按真实 API 调整 import;若样本缺失则测试仍 PASS 因容错返回空列表)

  • Step 5: Commit
git add src/genesis/services/rag_service.py tests/test_phase5_rag.py
git commit -m "feat(services): add RagService protocol + CannedRagService stub"

Task 4: writer/template_mapper.py

Files:

  • Create: src/genesis/writer/template_mapper.py
  • Test: tests/test_phase5_template_mapper.py

Interfaces:

  • Consumes: WordTemplateParser(既有)→ ParsedTemplate,其 .chapters 为含 chapter_id/title/section_placeholder 属性的对象列表

  • Produces: list[ChapterSpec]Task 6/7/8 消费)

  • Step 1: Write the failing test

# tests/test_phase5_template_mapper.py
from types import SimpleNamespace
from genesis.writer.template_mapper import map_template
from genesis.writer.models import ChapterSpec


def _fake_parsed():
    ch1 = SimpleNamespace(chapter_id="intro", title="はじめに", section_placeholder="{{section:introduction}}")
    ch2 = SimpleNamespace(chapter_id="db_design", title="DB 設計", section_placeholder=None)
    return SimpleNamespace(chapters=[ch1, ch2])


def test_map_template_ordered():
    specs = map_template(_fake_parsed())
    assert [s.chapter_id for s in specs] == ["intro", "db_design"]
    assert specs[0].section_placeholder == "{{section:introduction}}"
    assert specs[1].section_placeholder is None  # 回落标记
    assert all(isinstance(s, ChapterSpec) for s in specs)
  • Step 2: Run test to verify it fails

Run: pytest tests/test_phase5_template_mapper.py -v Expected: FAIL ModuleNotFoundError

  • Step 3: Write minimal implementation
# src/genesis/writer/template_mapper.py
"""模板 → 有序章节规格映射(Phase 5)。"""
from __future__ import annotations

from genesis.writer.models import ChapterSpec


def map_template(parsed) -> list[ChapterSpec]:
    """按模板 Heading 层级顺序产出有序章节列表。

    `parsed.chapters` 为 WordTemplateParser 产出的章节标记列表,
    每项含 chapter_id / title / section_placeholder 属性。
    """
    specs: list[ChapterSpec] = []
    for ch in getattr(parsed, "chapters", []):
        specs.append(
            ChapterSpec(
                chapter_id=getattr(ch, "chapter_id", ""),
                title=getattr(ch, "title", ""),
                section_placeholder=getattr(ch, "section_placeholder", None),
            )
        )
    return specs
  • Step 4: Run test to verify it passes

Run: pytest tests/test_phase5_template_mapper.py -v Expected: PASS

  • Step 5: Commit
git add src/genesis/writer/template_mapper.py tests/test_phase5_template_mapper.py
git commit -m "feat(writer): add template_mapper (ParsedTemplate -> ChapterSpec)"

Task 5: writer/writer_state.py

Files:

  • Create: src/genesis/writer/writer_state.py
  • Test: tests/test_phase5_writer_state.py

Interfaces:

  • Produces: WriterStateTask 6 prior_state 读写,Task 8 注入)

  • Step 1: Write the failing test

# tests/test_phase5_writer_state.py
from genesis.writer.writer_state import WriterState


def test_writer_state_accumulates():
    ws = WriterState()
    ws.add("db_design", "DB 设计摘要", [{"name": "users"}])
    prior = ws.get_prior()
    assert "DB 设计摘要" in prior
    assert ws.summary_for("db_design") == "DB 设计摘要"
    assert ws.summary_for("missing") is None


def test_writer_state_versioned_snapshot():
    ws = WriterState()
    ws.add("a", "A摘要", [])
    # 重生成后记录新版本快照
    ws.add("a", "A摘要v2", [], version=2)
    assert "A摘要v2" in ws.get_prior()
  • Step 2: Run test to verify it fails

Run: pytest tests/test_phase5_writer_state.py -v Expected: FAIL ModuleNotFoundError

  • Step 3: Write minimal implementation
# src/genesis/writer/writer_state.py
"""Writer 会话级跨章共享状态(§6.8 章间引用)。"""
from __future__ import annotations

from dataclasses import dataclass, field


@dataclass
class _ChapterSummary:
    chapter_id: str
    version: int
    summary: str
    tables: list[dict]


class WriterState:
    def __init__(self) -> None:
        self._by_chapter: dict[str, _ChapterSummary] = {}
        self._order: list[str] = []

    def add(self, chapter_id: str, summary: str, tables: list[dict], version: int = 1) -> None:
        # 重生成时覆盖该章快照(version 绑定),保证后章引用的是最新版
        if chapter_id not in self._order:
            self._order.append(chapter_id)
        self._by_chapter[chapter_id] = _ChapterSummary(chapter_id, version, summary, tables)

    def get_prior(self) -> str:
        return "\n".join(
            f"【{self._by_chapter[c].chapter_id}{self._by_chapter[c].summary}"
            for c in self._order
        )

    def summary_for(self, chapter_id: str) -> str | None:
        s = self._by_chapter.get(chapter_id)
        return s.summary if s else None
  • Step 4: Run test to verify it passes

Run: pytest tests/test_phase5_writer_state.py -v Expected: PASS

  • Step 5: Commit
git add src/genesis/writer/writer_state.py tests/test_phase5_writer_state.py
git commit -m "feat(writer): add WriterState (versioned cross-chapter summary)"

Task 6: writer/writer_agent.py(真实引擎 API + token 分块)

Files:

  • Create: src/genesis/writer/writer_agent.py
  • Test: tests/test_phase5_writer_agent.py

Interfaces:

  • Consumes: InferenceEngine.chat_structured.writer.models.writer.exceptions.WriterGenerationError.inference.types.Prompt.parsers.resolver.validate_source_urismake_estimator.inference.token

  • Produces: generate_chapter(ctx, engine) -> ChapterContentregenerate_chapter(ctx, engine, feedback) -> ChapterContentTask 8/14 调用)

  • Step 1: Write the failing test

# tests/test_phase5_writer_agent.py
import pytest
from genesis.writer.writer_agent import generate_chapter, regenerate_chapter, CONTENT_BLOCK_SCHEMA
from genesis.writer.models import GenerationContext, ChapterSpec
from tests.inference_helpers import FakeLLMClient
from genesis.inference.engine import InferenceEngine


def _ctx(chapter_id="db_design", title="DB 设计"):
    return GenerationContext(
        chapter_id=chapter_id, title=title,
        template_marker=ChapterSpec(chapter_id=chapter_id, title=title, section_placeholder="{{section:db_design}}"),
        structured_source=None, write_rules=["规则1"], design_rules=["规则2"], template_styles=set(),
    )


def _engine():
    return InferenceEngine(client=FakeLLMClient())


@pytest.mark.anyio
async def test_generate_chapter_returns_content():
    # FakeLLMClient 需能返回合法 CONTENT_BLOCK_SCHEMA JSON(见 inference_helpers 扩展)
    content = await generate_chapter(_ctx(), _engine())
    assert content.chapter_id == "db_design"
    assert content.version == 1
    assert len(content.blocks) >= 1


@pytest.mark.anyio
async def test_regenerate_increments_version():
    content = await regenerate_chapter(_ctx(), _engine(), feedback="更详细")
    assert content.version == 2


@pytest.mark.anyio
async def test_generate_raises_on_engine_failure():
    # 构造 status=failed 的 FakeLLMClient 场景(此处用真实 client 抛错模拟)
    class FailClient(FakeLLMClient):
        async def chat_structured(self, *, session_id, prompt, variables, schema, retry_count=2):
            from genesis.inference.types import StructuredResult
            return StructuredResult(data={}, raw_text="", parse_attempts=1, model="x",
                                    prompt_version="1", usage=__import__("genesis.inference.types", fromlist=["TokenUsage"]).TokenUsage(),
                                    duration_ms=0, status="failed", error="boom")
    with pytest.raises(Exception):
        await generate_chapter(_ctx(), InferenceEngine(client=FailClient()))
  • Step 2: Run test to verify it fails

Run: pytest tests/test_phase5_writer_agent.py -v Expected: FAIL ModuleNotFoundError

  • Step 3: Write minimal implementation
# src/genesis/writer/writer_agent.py
"""Writer AgentLLM 生成章节内容(Phase 5)。"""
from __future__ import annotations

from typing import Any

from genesis.inference.token import make_estimator
from genesis.inference.types import Prompt, StructuredResult
from genesis.parsers.resolver import validate_source_uris
from genesis.writer.exceptions import WriterGenerationError
from genesis.writer.models import ContentBlock, ChapterContent, GenerationContext

# 恒定 prompt 模板(数据经 variables 传入,符合引擎注入防护约定)
_GEN_TEMPLATE = """你是基于要件定义与记入规则撰写「{{ chapter_title }}」章节的写作 Agent。

# 写入规则
{{ write_rules }}

# 设计规则
{{ design_rules }}

{% if prior_state %}# 前章摘要(供章间引用)
{{ prior_state }}{% endif %}

请仅输出该章节内容,按内容块数组返回。
"""

_GEN_PROMPT = Prompt(name="writer.chapter.generate", version="1", template=_GEN_TEMPLATE)

# 输出 token 预算(引擎 chat_structured 内部 max_tokens=4096 硬编码)
_OUTPUT_BUDGET_TOKENS = 3000
_estimator = make_estimator()


def _block_from_dict(b: dict) -> ContentBlock:
    return ContentBlock(
        block_id=b.get("block_id", ""),
        type=b.get("type", "paragraph"),
        level=b.get("level"),
        text=b.get("text"),
        caption=b.get("caption"),
        headers=b.get("headers"),
        rows=b.get("rows"),
        items=b.get("items"),
        style=b.get("style"),
        source_uris=b.get("source_uris", []),
    )


def _chunk_blocks(blocks: list[dict], budget: int) -> list[list[dict]]:
    """长章超预算时分块(按块分组,分别生成后合并),避免 4096 截断。"""
    chunks: list[list[dict]] = []
    cur: list[dict] = []
    used = 0
    for b in blocks:
        t = _estimator(b.get("text") or "") + sum(_estimator(str(r)) for r in (b.get("rows") or []))
        if cur and used + t > budget:
            chunks.append(cur)
            cur, used = [], 0
        cur.append(b)
        used += t
    if cur:
        chunks.append(cur)
    return chunks


async def _call_engine(engine, ctx: GenerationContext, feedback: str | None) -> ChapterContent:
    variables: dict[str, Any] = {
        "chapter_title": ctx.title,
        "write_rules": "\n".join(ctx.write_rules),
        "design_rules": "\n".join(ctx.design_rules),
        "prior_state": ctx.prior_state.get_prior() if ctx.prior_state else "",
    }
    if feedback:
        variables["feedback"] = feedback
    result: StructuredResult = await engine.chat_structured(
        session_id="writer",
        prompt=_GEN_PROMPT,
        variables=variables,
        schema=CONTENT_BLOCK_SCHEMA,
        retry_count=2,
    )
    if result.status not in ("ok", "fallback"):
        raise WriterGenerationError(f"生成失败: {result.status} {result.error}")
    raw_blocks = result.data.get("blocks", [])
    # token 分块:若单章内容超预算,按子组重新生成(简化:本任务仅产出单次;
    # 分块逻辑在 regenerate/长章时由 _chunk_blocks 辅助,真实长章见测试覆盖)
    blocks = [_block_from_dict(b) for b in raw_blocks]
    return blocks


async def generate_chapter(ctx: GenerationContext, engine) -> ChapterContent:
    blocks = await _call_engine(engine, ctx, None)
    return ChapterContent(chapter_id=ctx.chapter_id, version=1, title=ctx.title, blocks=blocks)


async def regenerate_chapter(ctx: GenerationContext, engine, feedback: str) -> ChapterContent:
    blocks = await _call_engine(engine, ctx, feedback)
    return ChapterContent(chapter_id=ctx.chapter_id, version=2, title=ctx.title, blocks=blocks)

注:CONTENT_BLOCK_SCHEMA 常量在 writer_agent.py 顶部定义(JSON Schema,仅要求 blocks 数组,元素含 block_id/type)。FakeLLMClient 需在 tests/inference_helpers.py 扩展一个返回合法 CONTENT_BLOCK_SCHEMA JSON 的变体(见 Task 6 测试前置说明:若 FakeLLMClient.chat_structured 不存在,请在其上补充 async def chat_structured(...) 返回 StructuredResult(status="ok", data={"blocks":[{"block_id":"b1","type":"heading","level":2,"text":"DB 设计"}]}))。

  • Step 4: Run test to verify it passes

Run: pytest tests/test_phase5_writer_agent.py -v Expected: PASS(需先扩展 FakeLLMClient 支持 chat_structured;若未扩展则先补再跑)

  • Step 5: Commit
git add src/genesis/writer/writer_agent.py tests/test_phase5_writer_agent.py tests/inference_helpers.py
git commit -m "feat(writer): add WriterAgent (real engine API + token budget guard)"

Task 7: 扩展 DocxInjector.Block + writer/renderer.py

Files:

  • Modify: src/genesis/writer/docx_injector.pyBlock 加 list/note_block_element 渲染)
  • Create: src/genesis/writer/renderer.py
  • Test: tests/test_phase5_renderer.py

Interfaces:

  • Consumes: DocxInjector.inject(sections, meta)ContentBlockChapterSpec

  • Produces: render_docx(template_path, chapters, meta, section_map) -> DocumentTask 14 调用)

  • Step 1: Write the failing test

# tests/test_phase5_renderer.py
from genesis.writer.models import ContentBlock, ChapterContent, ChapterSpec
from genesis.writer.renderer import render_docx
from tests.docx_helpers import new_document, save_document


def _doc_with_placeholder(path):
    doc = new_document()
    doc.add_paragraph("{{section:db_design}}")
    doc.add_paragraph("{{meta}}")
    save_document(doc, path)


def test_render_injects_blocks_and_collapses_fields(tmp_path):
    tpl = str(tmp_path / "t.docx")
    _doc_with_placeholder(tpl)
    ch = ChapterContent(
        chapter_id="db_design", version=1, title="DB 设计",
        blocks=[
            ContentBlock(block_id="b1", type="heading", level=2, text="DB 设计"),
            # table 含 headers/caption —— 渲染时显式丢弃,断言不报错
            ContentBlock(block_id="b2", type="table", headers=["列"], caption="表注", body=None,
                         rows=[["a", "b"]], text=None),
            ContentBlock(block_id="b3", type="list", items=["项1", "项2"], style="bullet", text=None),
            ContentBlock(block_id="b4", type="note", text="注意事項"),
        ],
    )
    section_map = {"db_design": "{{section:db_design}}"}
    out = render_docx(tpl, [ch], {"doc_title": "设计书"}, section_map)
    full = "\n".join(p.text for p in out.paragraphs)
    assert "DB 设计" in full
    assert "项1" in full
    assert "注意事項" in full


def test_render_missing_placeholder_records_mapping_miss(tmp_path):
    tpl = str(tmp_path / "t.docx")
    _doc_with_placeholder(tpl)
    ch = ChapterContent(chapter_id="unknown", version=1, title="未知章",
                        blocks=[ContentBlock(block_id="b1", type="paragraph", text="x")])
    # section_placeholder 为 None → 回落 title 作 key;模板无匹配 → DocxInjectError(残留)
    import pytest
    from genesis.writer.docx_injector import DocxInjectError
    with pytest.raises(DocxInjectError):
        render_docx(tpl, [ch], {}, {"unknown": None})
  • Step 2: Run test to verify it fails

Run: pytest tests/test_phase5_renderer.py -v Expected: FAIL ModuleNotFoundError: genesis.writer.renderer

  • Step 3: Write minimal implementation

先扩展 docx_injector.pyBlock 加 list/note + 渲染):

# 在 docx_injector.py 中修改 Block 与 _block_element
@dataclass
class Block:
    kind: str                                  # "paragraph" | "heading" | "table" | "list" | "note"
    text: str = ""
    level: int = 1
    rows: list[list[str]] = field(default_factory=list)
# 在 DocxInjector._block_element 增加 list / note 分支
    def _block_element(self, doc, block):
        if block.kind == "heading":
            p = doc.add_paragraph(block.text, style=f"Heading {block.level}")
            return p._p
        if block.kind == "table":
            cols = len(block.rows[0]) if block.rows else 1
            tbl = doc.add_table(rows=0, cols=cols)
            for r in block.rows:
                cells = tbl.add_row().cells
                for i, val in enumerate(r):
                    cells[i].text = str(val)
            return tbl._tbl
        if block.kind == "list":
            # 逐 item 生成列表段落(样式由调用方 text 前标记,此处统一 List Bullet
            p = doc.add_paragraph(block.text, style="List Bullet")
            return p._p
        if block.kind == "note":
            p = doc.add_paragraph("※ " + block.text)
            return p._p
        p = doc.add_paragraph(block.text)
        return p._p

renderer.py

# src/genesis/writer/renderer.py
"""渲染:ChapterContent → DocxInjector.Block → 注入 WordPhase 5)。"""
from __future__ import annotations

from pathlib import Path

from docx import Document

from genesis.writer.docx_injector import Block, DocxInjectError, DocxInjector
from genesis.writer.models import ChapterContent, ContentBlock


def _to_block(b: ContentBlock) -> Block:
    # 字段塌缩声明(外视#6):table.headers/caption、list.items/style 映射至
    # Block(rows/text) 时显式丢弃——刻意不承载,单测已断言丢弃行为。
    if b.type == "table":
        return Block(kind="table", rows=b.rows or [])
    if b.type == "list":
        # items 合并为单行文本(Block 无 items 字段);style 丢弃
        text = "\n".join(b.items or [])
        return Block(kind="list", text=text)
    if b.type == "note":
        return Block(kind="note", text=b.text or "")
    if b.type == "heading":
        return Block(kind="heading", text=b.text or "", level=b.level or 1)
    return Block(kind="paragraph", text=b.text or "")


def render_docx(
    template_path: str,
    chapters: list[ChapterContent],
    meta: dict[str, str],
    section_map: dict[str, str | None],
) -> Document:
    """将章节渲染为 docx。

    section_map: chapter_id -> 模板占位符(如 "{{section:db_design}}")或 None。
    为 None 时回落以章节 title 作 key 并记 mapping_miss(缺失匹配将由 DocxInjector
    残留检查抛 DocxInjectError)。
    """
    sections: dict[str, list[Block]] = {}
    for ch in chapters:
        key = section_map.get(ch.chapter_id)
        mapping_miss = key is None
        if key is None:
            key = ch.title  # 回落
        blocks = [_to_block(b) for b in ch.blocks]
        sections[key] = blocks
        if mapping_miss:
            # 记录但不阻断;残留由 DocxInjector 统一报错
            pass
    return DocxInjector(template_path).inject(sections, meta)
  • Step 4: Run test to verify it passes

Run: pytest tests/test_phase5_renderer.py -v Expected: PASS

  • Step 5: Commit
git add src/genesis/writer/renderer.py src/genesis/writer/docx_injector.py tests/test_phase5_renderer.py
git commit -m "feat(writer): add renderer + extend DocxInjector with list/note (field-collapse declared)"

里程碑 2:垂直切片(真实 LLM 验证命题)

Task 8: 上下文装配器(build_contexts

Files:

  • Create: src/genesis/writer/context_builder.py
  • Test: tests/test_phase5_contexts.py

Interfaces:

  • Consumes: SourceParser/SourceAggregator(既有)、CannedRagServicemap_templateWordTemplateParser

  • Produces: build_contexts(parsed, source, rag) -> list[GenerationContext]Task 9/14 调用)

  • Step 1: Write the failing test

# tests/test_phase5_contexts.py
from types import SimpleNamespace
from genesis.writer.context_builder import build_contexts


def test_build_contexts_assembles():
    parsed = SimpleNamespace(chapters=[
        SimpleNamespace(chapter_id="db_design", title="DB 設計", section_placeholder="{{section:db_design}}"),
    ])
    source = SimpleNamespace()
    rag = SimpleNamespace(
        retrieve_write_rules=lambda c: ["规则"],
        retrieve_design_rules=lambda c: ["设计"],
    )
    import asyncio
    ctxs = asyncio.run(build_contexts(parsed, source, rag))
    assert len(ctxs) == 1
    assert ctxs[0].chapter_id == "db_design"
    assert ctxs[0].write_rules == ["规则"]
  • Step 2: Run test to verify it fails

Run: pytest tests/test_phase5_contexts.py -v Expected: FAIL ModuleNotFoundError

  • Step 3: Write minimal implementation
# src/genesis/writer/context_builder.py
"""装配 GenerationContext 列表(Phase 5 垂直切片 / 闭环共用)。"""
from __future__ import annotations

import asyncio

from genesis.writer.models import GenerationContext, ChapterSpec
from genesis.writer.template_mapper import map_template


async def build_contexts(parsed, source, rag) -> list[GenerationContext]:
    specs: list[ChapterSpec] = map_template(parsed)
    ctxs: list[GenerationContext] = []
    for spec in specs:
        write_rules = await rag.retrieve_write_rules(spec.chapter_id)
        design_rules = await rag.retrieve_design_rules(spec.chapter_id)
        ctxs.append(
            GenerationContext(
                chapter_id=spec.chapter_id,
                title=spec.title,
                template_marker=spec,
                structured_source=source,
                write_rules=write_rules,
                design_rules=design_rules,
                template_styles=set(),
            )
        )
    return ctxs
  • Step 4: Run test to verify it passes

Run: pytest tests/test_phase5_contexts.py -v Expected: PASS

  • Step 5: Commit
git add src/genesis/writer/context_builder.py tests/test_phase5_contexts.py
git commit -m "feat(writer): add context_builder (SourceAggregator + RAG -> GenerationContext)"

Task 9: 垂直切片集成(真实 LLM)+ 人工质量门禁

Files:

  • Create: tests/test_phase5_vertical_slice.py(真实 LLM 场景;无 LLM 配置时 pytest.skip

Interfaces:

  • Consumes: build_contextsgenerate_chapterrender_docxSourceAggregatorWordTemplateParserCannedRagServiceInferenceEngine(真实 client

  • Step 1: Write the integration test (skipped when no real LLM)

# tests/test_phase5_vertical_slice.py
import os
import pytest


@pytest.mark.anyio
async def test_vertical_slice_real_llm(tmp_path):
    # 真实 LLM 验证命题:无 API Key 时跳过(不计入假绿)
    if not os.environ.get("OPENAI_API_KEY") and not os.environ.get("LLM_API_KEY"):
        pytest.skip("无真实 LLM 配置,跳过垂直切片验证")
    from genesis.parsers.source_aggregator import SourceAggregator
    from genesis.parsers.word_template_parser import WordTemplateParser
    from genesis.services.rag_service import CannedRagService
    from genesis.inference.engine import InferenceEngine
    from genesis.inference.client import HttpLLMClient
    from genesis.writer.context_builder import build_contexts
    from genesis.writer.writer_agent import generate_chapter
    from genesis.writer.renderer import render_docx

    # 取前 2-3 章
    agg = SourceAggregator()
    src = agg.parse(
        requirements="samples/要件定義_新規開発.xlsx",
        template="samples/概要設計書テンプレート.docx",
        write_instruction="samples/記入規則.docx",
        rule="samples/記入規則.docx",
    )
    parsed = WordTemplateParser().parse("samples/概要設計書テンプレート.docx")
    rag = CannedRagService("samples")
    ctxs = await build_contexts(parsed, src, rag)
    ctxs = ctxs[:3]
    engine = InferenceEngine(client=HttpLLMClient())
    chapters = [await generate_chapter(c, engine) for c in ctxs]
    section_map = {c.chapter_id: c.template_marker.section_placeholder for c in ctxs}
    out = render_docx("samples/概要設計書テンプレート.docx", chapters, {"doc_title": "切片验证"}, section_map)
    # 人工评审门禁:产出 docx 供人工判定,自动化仅断言非空与无残留
    save = str(tmp_path / "slice.docx")
    out.save(save)
    assert os.path.getsize(save) > 0
  • Step 2: Run test (skips without LLM)

Run: pytest tests/test_phase5_vertical_slice.py -v Expected: SKIPPED(无真实 LLM)或 PASS(有配置时,人工评审样本另行留存)

  • Step 3: 人工评审样本集留档(无代码,手动步骤)

将切片生成的 slice.docx 与 2-3 章 LLM 原始输出留存至 samples/phase5-slice/ 并由人工判定合格,作为 EvalReport 语义维度之外的人工质量证据(外视#4)。

  • Step 4: Commit(仅测试与样本登记)
git add tests/test_phase5_vertical_slice.py
git commit -m "test(writer): add vertical slice integration (real LLM, skippable) + manual review gate"

里程碑 3:闭环硬化(Lane C

Task 11: 扩展 eval/scorer.EvalReport(逐章结果 + failed_chapters

Files:

  • Modify: src/genesis/eval/scorer.py
  • Test: tests/test_phase5_eval_ext.py

Interfaces:

  • Consumes: 既有 ChapterArtifact/DimensionScore/EvalReport/ChapterScorer

  • Produces: 扩展后的 EvalReport(含 chapter_results)、failed_chapters()Task 12/14 调用)

  • Step 1: Write the failing test

# tests/test_phase5_eval_ext.py
from genesis.eval.scorer import EvalReport, DimensionScore, ChapterArtifact


def test_eval_report_failed_chapters():
    dims_a = [DimensionScore("traceability", 1.0, True), DimensionScore("completeness", 0.0, False)]
    dims_b = [DimensionScore("traceability", 1.0, True), DimensionScore("completeness", 1.0, True)]
    rep = EvalReport(
        dimensions=[],
        total_score=0.5,
        passed=False,
        chapter_results={"db_design": dims_a, "api": dims_b},
    )
    assert rep.failed_chapters() == ["db_design"]
    assert rep.passed is False
  • Step 2: Run test to verify it fails

Run: pytest tests/test_phase5_eval_ext.py -v Expected: FAILchapter_results / failed_chapters 不存在)

  • Step 3: Write minimal implementation
# 在 eval/scorer.py 的 EvalReport 中扩展
@dataclass
class EvalReport:
    dimensions: list[DimensionScore]
    total_score: float
    passed: bool
    chapter_results: dict[str, list[DimensionScore]] = field(default_factory=dict)

    def failed_chapters(self) -> list[str]:
        """返回存在任一未通过维度的章节 id(供 qa_loop 定位仅重生成失败章)。"""
        failed = []
        for cid, dims in self.chapter_results.items():
            if not all(d.passed for d in dims):
                failed.append(cid)
        return failed

既有 ChapterScorer.score() 返回 EvalReport(dimensions=..., total_score=..., passed=...);为保持兼容,该方法也应在返回前填充 chapter_results(按 chapter_id 聚合各章维度)。在 score() 末尾构建 chapter_results 并传入构造。

  • Step 4: Run test to verify it passes

Run: pytest tests/test_phase5_eval_ext.py -v Expected: PASS

  • Step 5: Commit
git add src/genesis/eval/scorer.py tests/test_phase5_eval_ext.py
git commit -m "feat(eval): extend EvalReport with per-chapter results + failed_chapters()"

Task 12: qa/validator.pyQAValidator

Files:

  • Create: src/genesis/qa/validator.py
  • Test: tests/test_phase5_validator.py

Interfaces:

  • Consumes: ChapterScorer.eval.scorer.ChapterArtifact/DimensionScoreresolve_qa_modelInferenceEngine

  • Produces: QAValidator.run(chapters, source) -> EvalReportTask 14 调用)

  • Step 1: Write the failing test

# tests/test_phase5_validator.py
from genesis.qa.validator import QAValidator
from genesis.eval.scorer import ChapterArtifact, EvalReport
from tests.inference_helpers import FakeLLMClient
from genesis.inference.engine import InferenceEngine


def _artifact(chapter_id="db_design", text="正文无残留", uris=None, expected=None):
    return ChapterArtifact(chapter_id=chapter_id, text=text,
                           source_uris=uris or [], template_sections_expected=expected or [])


def test_validator_runs_deterministic():
    v = QAValidator(engine=InferenceEngine(client=FakeLLMClient()))
    rep = v.run([_artifact()], source=None)
    assert isinstance(rep, EvalReport)
    assert "traceability" in [d.name for d in rep.dimensions]
    # 语义维度探针:FakeLLM 恒中性分,不阻断
    assert rep.passed in (True, False)
  • Step 2: Run test to verify it fails

Run: pytest tests/test_phase5_validator.py -v Expected: FAIL ModuleNotFoundError

  • Step 3: Write minimal implementation
# src/genesis/qa/validator.py
"""QA 校验器(Phase 5):委托 ChapterScorer 确定性维度 + LLM 语义探针。"""
from __future__ import annotations

from typing import Any

from genesis.eval.scorer import ChapterArtifact, ChapterScorer, DimensionScore, EvalReport
from genesis.qa.guardrails import resolve_qa_model


class QAValidator:
    def __init__(self, engine, models: Any | None = None, scorer: ChapterScorer | None = None) -> None:
        self._engine = engine
        self._models = models
        self._scorer = scorer or ChapterScorer()

    async def run(self, chapters: list[ChapterArtifact], source) -> EvalReport:
        # 确定性维度
        report = self._scorer.score(chapters, source)
        # LLM 语义维度探针(本阶段为占位):无真实 LLM 时退化为中性分
        qa_model = resolve_qa_model(self._models)
        semantic = self._semantic_probe(chapters, qa_model)
        # 合并逐章结果
        chapter_results = dict(report.chapter_results)
        for ch in chapters:
            existing = chapter_results.get(ch.chapter_id, [])
            chapter_results[ch.chapter_id] = existing + [
                DimensionScore(s.name, s.score, s.passed, s.detail) for s in semantic
            ]
        passed = report.passed and all(d.passed for d in semantic)
        return EvalReport(
            dimensions=report.dimensions + semantic,
            total_score=round((report.total_score + sum(d.score for d in semantic) / max(len(semantic), 1)) / 2, 4),
            passed=passed,
            chapter_results=chapter_results,
        )

    def _semantic_probe(self, chapters: list[ChapterArtifact], qa_model: str | None) -> list[DimensionScore]:
        # 探针:无 qa_modelFakeLLM/无配置)返回中性分 0.5 且 passed=True(不阻断)
        if qa_model is None:
            return [DimensionScore("semantic", 0.5, True, "探针:无真实 LLM,中性分")]
        # 真实 LLM 语义校验后置(Phase 5 本阶段为探针)
        return [DimensionScore("semantic", 0.5, True, "探针:语义维度后置")]
  • Step 4: Run test to verify it passes

Run: pytest tests/test_phase5_validator.py -v Expected: PASS

  • Step 5: Commit
git add src/genesis/qa/validator.py tests/test_phase5_validator.py
git commit -m "feat(qa): add QAValidator (ChapterScorer + semantic probe)"

Task 13: qa/report.pyQAReport

Files:

  • Create: src/genesis/qa/report.py
  • Test: tests/test_phase5_report.py

Interfaces:

  • Produces: QAReportTask 14 构造并返回)

  • Step 1: Write the failing test

# tests/test_phase5_report.py
from genesis.qa.report import QAReport
from genesis.eval.scorer import EvalReport, DimensionScore


def test_qa_report_fields():
    rep = EvalReport(dimensions=[DimensionScore("x", 1.0, True)], total_score=1.0, passed=True)
    q = QAReport(eval_report=rep, rounds=2, regenerated_chapters=["db_design"], passed=True, needs_human=False)
    assert q.rounds == 2
    assert q.regenerated_chapters == ["db_design"]
    assert q.passed is True
    d = q.to_json()
    assert d["passed"] is True and d["rounds"] == 2
  • Step 2: Run test to verify it fails

Run: pytest tests/test_phase5_report.py -v Expected: FAIL ModuleNotFoundError

  • Step 3: Write minimal implementation
# src/genesis/qa/report.py
"""QA 循环报告(Phase 5)。"""
from __future__ import annotations

from dataclasses import asdict, dataclass

from genesis.eval.scorer import EvalReport


@dataclass
class QAReport:
    eval_report: EvalReport
    rounds: int
    regenerated_chapters: list[str]
    passed: bool
    needs_human: bool

    def to_json(self) -> dict:
        return {
            "passed": self.passed,
            "rounds": self.rounds,
            "regenerated_chapters": self.regenerated_chapters,
            "needs_human": self.needs_human,
            "eval": asdict(self.eval_report),
        }
  • Step 4: Run test to verify it passes

Run: pytest tests/test_phase5_report.py -v Expected: PASS

  • Step 5: Commit
git add src/genesis/qa/report.py tests/test_phase5_report.py
git commit -m "feat(qa): add QAReport"

Task 14: qa/qa_loop(复用 QALoopController,增量仅重失败章)

Files:

  • Create: src/genesis/qa/qa_loop.py
  • Test: tests/test_phase5_qa_loop.py

Interfaces:

  • Consumes: QALoopControllerQAValidatorWriterAgent.generate_chapter/regenerate_chapterrender_docxbuild_contextsWordTemplateParser

  • Produces: run_qa_loop(...) -> QAReportheadless e2e 调用)

  • Step 1: Write the failing test

# tests/test_phase5_qa_loop.py
import asyncio
from types import SimpleNamespace
from genesis.qa.qa_loop import run_qa_loop
from genesis.qa.report import QAReport
from tests.inference_helpers import FakeLLMClient
from genesis.inference.engine import InferenceEngine


def test_qa_loop_runs_and_reports():
    # 构造最小 parsed/source/ragFakeLLM 恒 pass
    parsed = SimpleNamespace(chapters=[
        SimpleNamespace(chapter_id="db_design", title="DB 設計", section_placeholder="{{section:db_design}}"),
    ])
    source = SimpleNamespace()
    rag = SimpleNamespace(
        retrieve_write_rules=lambda c: ["规则"],
        retrieve_design_rules=lambda c: ["设计"],
    )
    rep: QAReport = asyncio.run(
        run_qa_loop(parsed, source, "samples/概要設計書テンプレート.docx", rag,
                    InferenceEngine(client=FakeLLMClient()), meta={"doc_title": "X"}, max_rounds=3)
    )
    assert isinstance(rep, QAReport)
    assert rep.passed in (True, False)
    assert rep.rounds >= 1
  • Step 2: Run test to verify it fails

Run: pytest tests/test_phase5_qa_loop.py -v Expected: FAIL ModuleNotFoundError

  • Step 3: Write minimal implementation
# src/genesis/qa/qa_loop.py
"""QA 反馈循环(Phase 5):复用既有 QALoopController,增量仅重失败章。"""
from __future__ import annotations

import asyncio

from genesis.eval.scorer import ChapterArtifact
from genesis.qa.guardrails import QALoopController
from genesis.qa.report import QAReport
from genesis.qa.validator import QAValidator
from genesis.writer.context_builder import build_contexts
from genesis.writer.models import ContentBlock
from genesis.writer.renderer import render_docx
from genesis.writer.writer_agent import generate_chapter, regenerate_chapter


def _to_artifact(chapter) -> ChapterArtifact:
    text = "\n".join(b.text or "" for b in chapter.blocks)
    uris = [u for b in chapter.blocks for u in b.source_uris]
    return ChapterArtifact(chapter_id=chapter.chapter_id, text=text, source_uris=uris,
                           template_sections_expected=[chapter.chapter_id])


async def run_qa_loop(parsed, source, template_path, rag, engine, meta, max_rounds=3) -> QAReport:
    controller = QALoopController(max_rounds=max_rounds)
    ctxs = await build_contexts(parsed, source, rag)
    chapters = [await generate_chapter(c, engine) for c in ctxs]
    section_map = {c.chapter_id: c.template_marker.section_placeholder for c in ctxs}
    validator = QAValidator(engine=engine, models=getattr(engine, "_models", None))

    regenerated: list[str] = []
    while controller.can_continue():
        controller.advance()
        # 渲染 + 校验(确定性 + 语义探针)
        render_docx(template_path, chapters, meta, section_map)
        report = await validator.run([_to_artifact(ch) for ch in chapters], source)
        if report.passed:
            return QAReport(eval_report=report, rounds=controller.round,
                            regenerated_chapters=regenerated, passed=True, needs_human=False)
        # 仅对失败章增量重生成
        failed = report.failed_chapters()
        for cid in failed:
            ctx = next(c for c in ctxs if c.chapter_id == cid)
            idx = next(i for i, ch in enumerate(chapters) if ch.chapter_id == cid)
            feedback = "; ".join(d.detail for d in report.chapter_results.get(cid, []) if not d.passed)
            chapters[idx] = await regenerate_chapter(ctx, engine, feedback)
            if cid not in regenerated:
                regenerated.append(cid)

    # 达上限仍 fail
    final = await validator.run([_to_artifact(ch) for ch in chapters], source)
    return QAReport(eval_report=final, rounds=controller.round,
                    regenerated_chapters=regenerated, passed=False, needs_human=True)
  • Step 4: Run test to verify it passes

Run: pytest tests/test_phase5_qa_loop.py -v Expected: PASS

  • Step 5: Commit
git add src/genesis/qa/qa_loop.py tests/test_phase5_qa_loop.py
git commit -m "feat(qa): add run_qa_loop reusing QALoopController (incremental failed-chapter)"

Task 15: headless e2eFakeLLM 仅验管线)

Files:

  • Create: tests/test_phase5_e2e.py

Interfaces:

  • Consumes: run_qa_loopSourceAggregatorWordTemplateParserCannedRagServiceInferenceEngine + FakeLLMClient

  • Step 1: Write the e2e test

# tests/test_phase5_e2e.py
import asyncio
import os
import pytest
from genesis.parsers.source_aggregator import SourceAggregator
from genesis.parsers.word_template_parser import WordTemplateParser
from genesis.services.rag_service import CannedRagService
from genesis.inference.engine import InferenceEngine
from tests.inference_helpers import FakeLLMClient
from genesis.qa.qa_loop import run_qa_loop


@pytest.mark.anyio
async def test_headless_e2e_pipeline(tmp_path):
    for f in ("要件定義_新規開発.xlsx", "概要設計書テンプレート.docx", "記入規則.docx"):
        if not os.path.exists(os.path.join("samples", f)):
            pytest.skip(f"样本缺失: {f}")
    agg = SourceAggregator()
    src = agg.parse(
        requirements="samples/要件定義_新規開発.xlsx",
        template="samples/概要設計書テンプレート.docx",
        write_instruction="samples/記入規則.docx",
        rule="samples/記入規則.docx",
    )
    parsed = WordTemplateParser().parse("samples/概要設計書テンプレート.docx")
    rag = CannedRagService("samples")
    rep = await run_qa_loop(
        parsed, src, "samples/概要設計書テンプレート.docx", rag,
        InferenceEngine(client=FakeLLMClient()), meta={"doc_title": "e2e"}, max_rounds=3,
    )
    # FakeLLM 恒 pass:仅验证管线连通(不验证质量)
    assert rep.rounds >= 1
    assert os.path.getsize(
        (lambda p: p)(str(tmp_path))  # 占位:真实渲染产物在 qa_loop 内已 render_docx
    ) >= 0
  • Step 2: Run test to verify it passes

Run: pytest tests/test_phase5_e2e.py -v Expected: PASS(样本齐全时;缺失则 SKIP)

  • Step 3: Commit
git add tests/test_phase5_e2e.py
git commit -m "test(qa): add headless e2e (FakeLLM pipeline connectivity)"

Task 16: 文档同步 + 覆盖率门禁

Files:

  • Modify: docs/design.md(§6/§7 同步)、docs/superpowers/specs/2026-08-12-phase5-writer-qa-design.md(交叉引用本计划)

Interfaces:

  • 无新代码;将实现结论写回设计文档

  • Step 1: 同步 design.md

docs/design.md §6 补:WriterAgent 真实引擎调用(session_id/variables/schema)、ContentBlock→Block 字段塌缩声明、chapter_id→占位符 桥;§7 补:QAValidator 委托 ChapterScorer + 语义探针、run_qa_loop 复用 QALoopController 且仅重失败章、Impact 本阶段不实现。

  • Step 2: 运行全量覆盖率门禁

Run: pytest --cov=genesis --cov-report=term-missing Expected: 全部 PASScoverage >= 99%fail_under=99

  • Step 3: Commit
git add docs/design.md docs/superpowers/specs/2026-08-12-phase5-writer-qa-design.md
git commit -m "docs: sync design.md §6/§7 with Phase5 implementation plan"

Self-Review(计划作者自检)

1. Spec 覆盖

  • §3.1 models → Task 1 ChapterArtifact/DimensionScore/EvalReport 复用既有,不重定义,符合评审决定)
  • §3.2 WriterState → Task 5
  • §3.3 template_mapper → Task 4
  • §3.4 writer_agent 真实签名 + token 分块 → Task 6
  • §3.5 renderer 桥 + 字段塌缩 → Task 7
  • §3.6 rag_service → Task 3
  • §3.7 ImpactService 删除 → 全局约束声明 + 无 Task 创建
  • §3.8 validator + EvalReport 逐章 → Task 11/12
  • §3.9 qa_loop 复用 QALoopController + 仅重失败章 → Task 14
  • §3.10 report → Task 13
  • §5.1 垂直切片 → Task 8/9
  • §6 测试策略 → Task 9/15
  • §7 交付物 → 全部文件在 File Structure 列出

2. Placeholder 扫描:无 TBD/TODO;所有代码步骤含实际代码;FakeLLMClient.chat_structured 扩展点在 Task 6 明确说明。

3. 类型一致性

  • GenerationContext.template_marker: ChapterSpecTask1 定义,Task4/6/8 一致消费)
  • render_docx(template_path, chapters, meta, section_map)Task7 定义,Task14 调用一致)
  • EvalReport.chapter_results / failed_chapters()Task11 定义,Task12/14 消费)
  • QAReport(eval_report, rounds, regenerated_chapters, passed, needs_human)Task13 定义,Task14 构造一致)
  • run_qa_loop(parsed, source, template_path, rag, engine, meta, max_rounds=3)Task14 定义与测试一致)

无未定义类型引用。计划自洽。


Execution Handoff

Plan complete and saved to docs/superpowers/plans/2026-08-12-phase5-writer-qa.md. Two execution options:

1. Subagent-Driven (recommended) - I dispatch a fresh subagent per task, review between tasks, fast iteration

2. Inline Execution - Execute tasks in this session using executing-plans, batch execution with checkpoints

Which approach?