diff --git a/_AI_USAGE_LOG.md b/_AI_USAGE_LOG.md index df3293b..17fa506 100644 --- a/_AI_USAGE_LOG.md +++ b/_AI_USAGE_LOG.md @@ -171,3 +171,5 @@ | 2026-09-13 00:47 | 反馈迭代 | 要件定义书改为待发附件随消息发送：前端 pendingAttachment chip→send 先上传再发消息(body.attachment)；消息绑定 attachment(store 增列+旧库迁移, agent 透传)；新增 GET /api/sessions/{sid}/files/{file_type}；service.file_path；前端加 xlsx(SheetJS) 在线查看多 sheet 弹层 | src/genesis/server/store.py; src/genesis/server/service.py; src/genesis/server/app.py; src/genesis/chat/agent.py; frontend/src/main.ts; frontend/src/types.ts; frontend/src/chat.css; frontend/package.json; tests/test_server_store.py; tests/test_server_api.py | deepseek-chat | | 2026-09-13 01:00 | 反馈迭代 | 附件展示调整：用户消息改为 .msg-stack 列布局，文字在气泡内、附件 chip 在气泡下方独立一行右对齐；另重启后端使 /api/sessions/{sid}/files/{file_type} 新路由生效(原 404 系旧进程未含新路由) | frontend/src/main.ts; frontend/src/chat.css; src/genesis/server/static/chat.js; src/genesis/server/static/chat.css | deepseek-chat | | 2026-09-13 21:43 | 反馈迭代 | 推送代码到 Gitea（方案1：仅同步 main，29 提交 b14c221→afd8080，远程仅保留 main）；修复推送认证 403/pre-receive declined：根因为全局 ~/.gitconfig 中 url.@gittea.dev/.insteadOf 重写规则回填只读令牌，已删除该规则、清空失效 .git-credentials、清理 Windows 凭据缓存，改用 GCM 输入有 write 权限账号后推送成功 | .git/config; ~/.gitconfig; ~/.git-credentials | deepseek-flash | +| 2026-09-15 11:30 | Agent 实现 | P0 修复(W1，Writer 输出完整性)：①表格表头行/Table Grid 边框不再塌缩（renderer 透传 headers、docx_injector 渲染表头行+边框）②列表逐项套用 List Bullet/Number 样式 ③行内占位符替换保留首个 run 字符格式 ④章间引用打通（WriterState.summary_of/summaries_before → 后章 prompt {{prior_summaries}}）⑤删除 writer_agent._chunk_source 死代码，章节数据经 DataGate 控 token 预算（超预算按最大表优先省略+标记，不阻断生成）⑥章节结果落盘快照 JSON + 命中快照跳过 LLM（录制重拍可续跑，损坏快照自动忽略）⑦生成阶段 per-chapter 日志可观测。全量 664 通过、覆盖率 99.36%（红线 99%） | src/genesis/writer/renderer.py; src/genesis/writer/docx_injector.py; src/genesis/writer/models.py; src/genesis/writer/writer_state.py; src/genesis/writer/writer_agent.py; src/genesis/writer/orchestrator.py; src/genesis/writer/context_builder.py; tests/test_phase5_renderer.py; tests/test_docx_injector.py; tests/test_phase5_models.py; tests/test_phase5_writer_state.py; tests/test_phase5_writer_orchestrator.py; tests/test_phase5_writer_agent.py; tests/test_writer_data_budget.py; tests/test_writer_snapshot.py; tests/test_writer_agent_branches.py | deepseek-flash | +| 2026-09-15 21:03 | Agent 实现 | P2-1/P2-2 廉价防灾：①P2-1 Word 占位符漏表格单元格——新增 _word_common.iter_all_paragraphs（按 body 顺序遍历正文+单元格段落含嵌套表，按 w:tc 去重合并单元格），word_template_parser 与 docx_injector 的 _inject_sections/_inject_inline/_has_residue/_collect_residue 全部改用它，消除「表格内锚点漏检→静默丢章/漏检残留」②P2-2 .xls 扩展名不一致——excel_reader 定为唯一事实来源(LEGACY_EXCEL_EXTS/HINT 可操作提示)，source_aggregator XLSX_EXTS 收窄为 .xlsx，service 接入 EXPECTED_SUFFIX 上传即校验(rules 修正为 docx-only，对齐解析层)、config.allowed_extensions 移除 .xls。全量 680 通过、覆盖率 99.37%（红线 99%） | src/genesis/parsers/_word_common.py; src/genesis/parsers/word_template_parser.py; src/genesis/parsers/excel_reader.py; src/genesis/parsers/source_aggregator.py; src/genesis/writer/docx_injector.py; src/genesis/server/service.py; src/genesis/config.py; tests/test_word_template_parser.py; tests/test_docx_injector.py; tests/test_excel_reader.py; tests/test_source_aggregator.py; tests/test_server_service.py; tests/test_config.py; tests/test_phase5_writer_orchestrator.py | deepseek-flash | diff --git a/src/genesis/config.py b/src/genesis/config.py index ef79b2e..2f5d486 100644 --- a/src/genesis/config.py +++ b/src/genesis/config.py @@ -17,7 +17,7 @@ ENV_PREFIX = "GENESIS_" class ServerConfig(BaseModel): max_upload_mb: int = 100 allowed_extensions: list[str] = Field( - default_factory=lambda: [".xlsx", ".xls", ".docx", ".pptx", ".java", ".xml", ".yml", + default_factory=lambda: [".xlsx", ".docx", ".pptx", ".java", ".xml", ".yml", ".py", ".ts", ".go", ".cs"] ) diff --git a/src/genesis/parsers/_word_common.py b/src/genesis/parsers/_word_common.py index 8f16119..55da7b6 100644 --- a/src/genesis/parsers/_word_common.py +++ b/src/genesis/parsers/_word_common.py @@ -1,7 +1,11 @@ -"""Word 解析共享小工具（WordTemplateParser / RuleDocParser 复用）。""" +"""Word 解析共享小工具（WordTemplateParser / RuleDocParser / DocxInjector 复用）。""" from __future__ import annotations +from docx.oxml.ns import qn +from docx.table import Table +from docx.text.paragraph import Paragraph + def heading_level(style_name: str) -> int: """从 Heading N 样式名解析大纲级别；非数字/无后缀兜底 1。""" @@ -9,3 +13,29 @@ def heading_level(style_name: str) -> int: return int(style_name.split()[-1]) except (ValueError, IndexError): return 1 + + +def iter_all_paragraphs(doc): + """按文档顺序遍历「正文段落 + 表格单元格段落（含嵌套表）」。 + + - 保持 body 顺序（正文段落与表格交错），使模板解析的「章 → 锚点」绑定 + 不错位——`doc.paragraphs` 会把正文段落与表格分成两批，破坏顺序。 + - 合并单元格的 `row.cells` 会重复指向同一 `w:tc` → 按 tc 去重， + 避免重复收集占位符 / 重复注入。 + """ + seen_tc: set = set() + + def walk(element): + for child in element.iterchildren(): + if child.tag == qn("w:p"): + yield Paragraph(child, doc) + elif child.tag == qn("w:tbl"): + for row in Table(child, doc).rows: + for cell in row.cells: + tc = cell._tc + if id(tc) in seen_tc: + continue + seen_tc.add(id(tc)) + yield from walk(tc) + + yield from walk(doc.element.body) diff --git a/src/genesis/parsers/excel_reader.py b/src/genesis/parsers/excel_reader.py index dc41b52..a88e22f 100644 --- a/src/genesis/parsers/excel_reader.py +++ b/src/genesis/parsers/excel_reader.py @@ -6,11 +6,19 @@ from typing import Any from openpyxl import load_workbook from openpyxl.worksheet.worksheet import Worksheet +# P2-2：旧版 .xls（BIFF）不受支持。此处为扩展名校验的单一事实来源， +# 其他层（source_aggregator / server service）均从此处引用，避免多处声明互相矛盾。 +LEGACY_EXCEL_EXTS = (".xls",) +LEGACY_EXCEL_HINT = "不支持旧版 Excel 格式 {ext}：请用 Excel 另存为 .xlsx 后重试" + def open_workbook(path: str | Path): - """普通模式打开 .xlsx（保留公式/样式/批注），.xls 报错。""" + """普通模式打开 .xlsx（保留公式/样式/批注）；.xls 给出可操作提示。""" path = Path(path) - if path.suffix.lower() != ".xlsx": + suffix = path.suffix.lower() + if suffix in LEGACY_EXCEL_EXTS: + raise ValueError(LEGACY_EXCEL_HINT.format(ext=path.suffix)) + if suffix != ".xlsx": raise ValueError(f"不支持的 Excel 格式: {path.suffix}") return load_workbook(path) diff --git a/src/genesis/parsers/source_aggregator.py b/src/genesis/parsers/source_aggregator.py index f161462..2c57c5a 100644 --- a/src/genesis/parsers/source_aggregator.py +++ b/src/genesis/parsers/source_aggregator.py @@ -6,22 +6,29 @@ from genesis.data_models import StructuredSource from genesis.impact.code_parser import CodeParser from genesis.impact.existing_system_explorer import ExistingSystemExplorer from genesis.parsers.excel_parser import ExcelParser +from genesis.parsers.excel_reader import LEGACY_EXCEL_EXTS, LEGACY_EXCEL_HINT from genesis.parsers.rule_doc_parser import RuleDocParser from genesis.parsers.word_template_parser import WordTemplateParser -XLSX_EXTS = (".xlsx", ".xls") +# .xlsx 为唯一受支持的 Excel 格式（旧版 .xls 显式拒绝，见 excel_reader）。 +XLSX_EXTS = (".xlsx",) DOCX_EXT = ".docx" +__all__ = ["SourceParser", "XLSX_EXTS", "DOCX_EXT", "LEGACY_EXCEL_EXTS", "LEGACY_EXCEL_HINT"] + def _validate_path(path: str | Path, allowed_exts: tuple[str, ...]) -> Path: """校验文件扩展名合法且文件存在（T7 DRY：消除三处重复校验）。 Raises: - ValueError: 扩展名不在 allowed_exts（含无扩展名） + ValueError: 旧版 .xls / 扩展名不在 allowed_exts（含无扩展名） FileNotFoundError: 文件不存在 """ p = Path(path) - if p.suffix.lower() not in allowed_exts: + suffix = p.suffix.lower() + if suffix in LEGACY_EXCEL_EXTS: + raise ValueError(LEGACY_EXCEL_HINT.format(ext=p.suffix)) + if suffix not in allowed_exts: raise ValueError(f"不支持的文件类型: {p.suffix or '(无扩展名)'}") if not p.exists(): raise FileNotFoundError(str(path)) diff --git a/src/genesis/parsers/word_template_parser.py b/src/genesis/parsers/word_template_parser.py index 5ac2742..746c6e8 100644 --- a/src/genesis/parsers/word_template_parser.py +++ b/src/genesis/parsers/word_template_parser.py @@ -7,7 +7,7 @@ from docx import Document from docx.oxml.ns import qn from genesis.data_models import ChapterMarker, ParsedTemplate -from genesis.parsers._word_common import heading_level +from genesis.parsers._word_common import heading_level, iter_all_paragraphs # 统一占位符正则：{{键名}} 或 {{键名:章节名}}（spec §3.2）。 # 宽容：键名大小写不敏感、分隔符支持半角(:)/全角(：)冒号；解析时归一为小写键名 + 半角冒号。 @@ -29,7 +29,8 @@ class WordTemplateParser: # 文档命名样式（定义集合） defined = {s.name for s in doc.styles if s.name} - for para in doc.paragraphs: + # P2-1：含表格单元格（按 body 顺序），避免表格内锚点漏检 / 错绑 + for para in iter_all_paragraphs(doc): style_name = para.style.name if para.style else "Normal" used_styles.add(style_name) text = para.text diff --git a/src/genesis/server/service.py b/src/genesis/server/service.py index de64619..4de73d2 100644 --- a/src/genesis/server/service.py +++ b/src/genesis/server/service.py @@ -21,7 +21,11 @@ from pathlib import Path from docx import Document -from genesis.parsers.source_aggregator import SourceParser +from genesis.parsers.source_aggregator import ( + LEGACY_EXCEL_EXTS, + LEGACY_EXCEL_HINT, + SourceParser, +) from genesis.server.store import SessionStore, SessionRecord, ProjectsStore, ProjectConfigError _LOGGER = logging.getLogger(__name__) @@ -30,12 +34,12 @@ ALLOWED_FILE_TYPES = { "requirements", "template", "write_instruction", "rules", "existing_system", } -# 各类型建议扩展名（宽松校验：仅拒绝明显非法的空文件） +# 各类型允许的扩展名（上传即校验，与解析层能力对齐；.xls 见 LEGACY_EXCEL_HINT） EXPECTED_SUFFIX = { "requirements": (".xlsx",), "template": (".docx",), "write_instruction": (".docx",), - "rules": (".docx", ".xlsx"), + "rules": (".docx",), # 解析层（source_aggregator）仅支持 docx 规则 "existing_system": (".zip",), } @@ -150,9 +154,27 @@ class GenesisService: "chunks": chunks, } + @staticmethod + def _validate_upload_suffix(file_type: str, filename: str) -> None: + """上传即校验扩展名（失败快于解析期）。 + + Raises: + FileTypeError: 旧版 .xls，或扩展名不在该 file_type 的允许集合内 + """ + suffix = Path(filename).suffix.lower() + if suffix in LEGACY_EXCEL_EXTS: + raise FileTypeError(LEGACY_EXCEL_HINT.format(ext=Path(filename).suffix)) + expected = EXPECTED_SUFFIX.get(file_type) + if expected and suffix not in expected: + raise FileTypeError( + f"不支持的扩展名 {suffix or '(无扩展名)'}" + f"（{file_type} 允许: {', '.join(expected)}）" + ) + def upload_file(self, session_id: str, file_type: str, filename: str, content: bytes) -> dict: if file_type not in ALLOWED_FILE_TYPES: raise FileTypeError(f"不支持的 file_type: {file_type}") + self._validate_upload_suffix(file_type, filename) sdir = self.data_root / session_id / "uploads" sdir.mkdir(parents=True, exist_ok=True) if file_type == "existing_system": diff --git a/src/genesis/writer/context_builder.py b/src/genesis/writer/context_builder.py index aa45d14..566fa42 100644 --- a/src/genesis/writer/context_builder.py +++ b/src/genesis/writer/context_builder.py @@ -29,7 +29,6 @@ def build_contexts( write_rules=write_rules, design_rules=design_rules, template_styles=set(used), - prior_state=None, impact_report=getattr(structured_source, "impact_report", None), output_language=output_language, ) diff --git a/src/genesis/writer/docx_injector.py b/src/genesis/writer/docx_injector.py index fae6ea9..dc963e8 100644 --- a/src/genesis/writer/docx_injector.py +++ b/src/genesis/writer/docx_injector.py @@ -21,6 +21,8 @@ from docx.document import Document as DocxDocument from docx.oxml.ns import qn from docx.text.paragraph import Paragraph +from genesis.parsers._word_common import iter_all_paragraphs + # 宽容：docx 正文锚点可能写为 {{Section:id}} / {{section：id}}（大小写/全角冒号） _SECTION_RE = re.compile(r"\{\{\s*section\s*[:：]\s*([^}]+?)\s*\}\}", re.IGNORECASE) _INLINE_RE = re.compile(r"\{\{([^}]+)\}\}") @@ -32,12 +34,19 @@ class DocxInjectError(Exception): @dataclass class Block: - """简化的内容块（ContentBlock 原型的子集）。""" + """简化的内容块（ContentBlock 原型的子集）。 - kind: str # "paragraph" | "heading" | "table" + - table: text=标题(caption)、headers=表头行、rows=数据行 + - list: items=列表项、style="bullet"|"numbered" + """ + + kind: str # "paragraph" | "heading" | "table" | "list" | "note" text: str = "" level: int = 1 # heading 层级 - rows: list[list[str]] = field(default_factory=list) # table 行 + headers: list[str] = field(default_factory=list) # table 表头行 + rows: list[list[str]] = field(default_factory=list) # table 数据行 + items: list[str] = field(default_factory=list) # list 列表项 + style: str = "" # list 样式: "bullet" | "numbered" class DocxInjector: @@ -130,7 +139,8 @@ class DocxInjector: # ---------- 内部 ---------- def _inject_sections(self, doc: DocxDocument, sections: dict[str, list[Block]]) -> None: - for para in list(doc.paragraphs): + # P2-1：含表格单元格（按 body 顺序）；先物化再注入，避免边注入边遍历 + for para in list(iter_all_paragraphs(doc)): m = _SECTION_RE.search(para.text) if not m: continue @@ -142,7 +152,7 @@ class DocxInjector: self._replace_paragraph_with_blocks(doc, para, blocks) def _inject_inline(self, doc: DocxDocument, meta: dict[str, str]) -> None: - for para in doc.paragraphs: + for para in iter_all_paragraphs(doc): if _INLINE_RE.search(para.text): # 仅替换行内占位符，保留模板其余文本 new_text = _INLINE_RE.sub(lambda mm: meta.get(mm.group(1), mm.group(0)), para.text) @@ -163,42 +173,79 @@ class DocxInjector: for el in reversed(self._block_element(doc, block)): parent.insert(para_idx, el) + _NUMBERED_STYLES = {"numbered", "number", "ordered"} + + @staticmethod + def _apply_style_if_present(doc: DocxDocument, obj, name: str) -> None: + """模板中定义了该样式才套用（自定义模板可能缺标准样式名）。""" + if name in doc.styles: # pragma: no cover - 取决于模板是否定义标准样式 + obj.style = name + def _block_element(self, doc: DocxDocument, block: Block) -> list: if block.kind == "heading": p = doc.add_paragraph(block.text, style=f"Heading {block.level}") return [p._p] if block.kind == "table": - elements: list = [] - if block.text: - style = "Caption" if "Caption" in doc.styles else None - cap = doc.add_paragraph(block.text, style=style) - elements.append(cap._p) - tbl = doc.add_table(rows=0, cols=len(block.rows[0]) if block.rows else 1) - for r in block.rows: - cells = tbl.add_row().cells - for i, val in enumerate(r): - cells[i].text = str(val) - elements.append(tbl._tbl) # type: ignore[attr-defined] - return elements + return self._table_element(doc, block) + if block.kind == "list": + return self._list_elements(doc, block) # 默认 paragraph p = doc.add_paragraph(block.text) return [p._p] + def _table_element(self, doc: DocxDocument, block: Block) -> list: + """表格：标题(Caption) + 表头行 + 数据行；套用 Table Grid 保证边框。""" + elements: list = [] + if block.text: + style = "Caption" if "Caption" in doc.styles else None + cap = doc.add_paragraph(block.text, style=style) + elements.append(cap._p) + widths = [len(block.headers)] + [len(r) for r in block.rows] + ncols = max(max(widths, default=0), 1) + tbl = doc.add_table(rows=0, cols=ncols) + self._apply_style_if_present(doc, tbl, "Table Grid") + if block.headers: + header_cells = tbl.add_row().cells + for i, val in enumerate(block.headers[:ncols]): + header_cells[i].text = str(val) + for r in block.rows: + cells = tbl.add_row().cells + for i, val in enumerate(r[:ncols]): + cells[i].text = str(val) + elements.append(tbl._tbl) # type: ignore[attr-defined] + return elements + + def _list_elements(self, doc: DocxDocument, block: Block) -> list: + """列表：逐项成段，套用 List Bullet / List Number 样式。""" + items = block.items or ([block.text] if block.text else []) + numbered = (block.style or "").lower() in self._NUMBERED_STYLES + style = "List Number" if numbered else "List Bullet" + paras = [doc.add_paragraph(str(it)) for it in items] + for p in paras: + self._apply_style_if_present(doc, p, style) + return [p._p for p in paras] + def _set_paragraph_text(self, para: Paragraph, text: str) -> None: - # 清空 run，写入单 run（原型简化；保留段落样式） - for run in list(para.runs): - run._r.getparent().remove(run._r) - para.add_run(text) + # 保留首个 run 的字符格式（字体/字号/加粗），替换其文本，其余 run 清空。 + # 旧实现清空全部 run 后新建，会丢失模板定义的字符格式（外视 #6 反转）。 + runs = list(para.runs) + if not runs: + para.add_run(text) + return + runs[0].text = text + for run in runs[1:]: + run.text = "" def _has_residue(self, doc: DocxDocument) -> bool: - for para in doc.paragraphs: + # P2-1：单元格内的未替换占位符也必须检出（否则静默丢内容） + for para in iter_all_paragraphs(doc): if _INLINE_RE.search(para.text): return True return False def _collect_residue(self, doc: DocxDocument) -> list[str]: found: list[str] = [] - for para in doc.paragraphs: + for para in iter_all_paragraphs(doc): for m in _INLINE_RE.finditer(para.text): found.append(m.group(0)) return found diff --git a/src/genesis/writer/models.py b/src/genesis/writer/models.py index c2128ed..22a2961 100644 --- a/src/genesis/writer/models.py +++ b/src/genesis/writer/models.py @@ -5,10 +5,14 @@ from dataclasses import dataclass, field from typing import Literal from genesis.data_models import ElementType, SheetType +from genesis.orchestrator.datagate import DataGate, DataGateError, DataSelector # 单表渲染行数上限（防 token 爆炸；样本量小，通常不触发） MAX_ROWS_PER_TABLE = 200 +# 章节数据 token 预算（OV5 DataGate 默认上限；可由 GenerationContext 注入覆盖） +MAX_CHAPTER_DATA_TOKENS = 8_000 + # design.md §6.8 ①「DataGate.load(structured_source, selector=该章数据)」的章节级选择器： # 章节占位符 id → 本章对应的 Excel Sheet 类型 # - introduction 概览章注入全部类型表 @@ -38,8 +42,8 @@ CHAPTER_IMPACT_ELEMENT: dict[str, ElementType | None] = { @dataclass class ContentBlock: - """LLM 生成的内容块。注意：table.headers/caption、list.items/style 在渲染至 - DocxInjector.Block 时显式丢弃（renderer 中声明并测试）。""" + """LLM 生成的内容块。table.headers/caption、list.items/style 会完整透传至 + DocxInjector.Block（渲染器保证表头/边框/列表样式，见 renderer）。""" block_id: str type: Literal["paragraph", "heading", "table", "list", "note"] @@ -107,9 +111,10 @@ class GenerationContext: write_rules: list[str] design_rules: list[str] template_styles: set[str] - prior_state: object | None = None # WriterState，避免循环 import 用 object + prior_summaries: str = "" # 前章摘要（design.md §6.9 章间引用，由编排层注入） impact_report: object | None = None # ImpactReport 影响调查书（生成主上下文） output_language: str = "auto" # "auto" | "zh" | "ja"（步骤 1：用户可选输出语言） + data_token_budget: int = MAX_CHAPTER_DATA_TOKENS # 本章数据 token 预算（OV5 DataGate） def _language_instruction(self) -> str: """根据 output_language 生成【语言约束】段的具体指令（步骤 1）。""" @@ -137,11 +142,12 @@ class GenerationContext: "sub_headings": "\n".join( f"- {h}" for h in (getattr(tm, "sub_headings", None) or []) ), - "prior_state": str(self.prior_state) if self.prior_state is not None else "", + "prior_summaries": self.prior_summaries, "language_instruction": self._language_instruction(), "data": _format_chapter_data( self.structured_source, CHAPTER_SHEET_TYPES.get(self.chapter_id, []), + token_budget=self.data_token_budget, ), "impact": _format_impact( self.impact_report, @@ -167,12 +173,47 @@ def _render_table(tb) -> list[str]: return lines -def _format_chapter_data(structured_source: object | None, sheet_types: list[SheetType]) -> str: +_TRUNCATION_NOTE = "（注：以下表因超出上下文 token 预算被省略：{names}）" + + +def _table_tokens(gate: DataGate, source: object, table) -> float: + """单表 token 估算；单表即超预算时视为无穷大（优先被省略）。""" + try: + return gate.load(source, DataSelector(table_ids=[table.name])).token_estimate + except DataGateError: + return float("inf") + + +def _fits_budget(gate: DataGate, source: object, tables: list) -> bool: + try: + gate.load(source, DataSelector(table_ids=[t.name for t in tables])) + return True + except DataGateError: + return False + + +def _shrink_to_budget(tables: list, source: object, gate: DataGate) -> tuple[list, list[str]]: + """按「最大表优先」省略，直到整体落入预算；返回 (保留表, 被省略表名)。""" + kept = list(tables) + omitted: list[str] = [] + while kept and not _fits_budget(gate, source, kept): + biggest = max(kept, key=lambda t: _table_tokens(gate, source, t)) + kept.remove(biggest) + omitted.append(biggest.name) + return kept, omitted + + +def _format_chapter_data( + structured_source: object | None, + sheet_types: list[SheetType], + token_budget: int = MAX_CHAPTER_DATA_TOKENS, +) -> str: """按章节定向格式化要件定义数据（design.md §6.8 ① selector=该章数据）。 - 命中 sheet_types 的表全部注入（章节主题数据） - GENERIC（自由記述）作为通用背景始终注入 - structured_source 为 None 或无任何可注入表时返回空串 + - 经 DataGate 控 token 预算（OV5）：超预算时按最大表优先省略并留省略标记，不抛错阻断生成 """ if structured_source is None: return "" @@ -185,11 +226,18 @@ def _format_chapter_data(structured_source: object | None, sheet_types: list[She selected = matched + generic if not selected: return "" + gate = DataGate(max_total_tokens=token_budget) + kept, omitted = _shrink_to_budget(selected, structured_source, gate) + if not kept: + return _TRUNCATION_NOTE.format(names=", ".join(omitted)) lines: list[str] = [] - for tb in selected: + for tb in kept: lines.extend(_render_table(tb)) lines.append("") - return "\n".join(lines).rstrip() + text = "\n".join(lines).rstrip() + if omitted: + text += "\n\n" + _TRUNCATION_NOTE.format(names=", ".join(omitted)) + return text # 影响调查标签本地化（步骤 B）：auto/ja 默认日文，zh 中文 diff --git a/src/genesis/writer/orchestrator.py b/src/genesis/writer/orchestrator.py index 3f89c21..73a52a7 100644 --- a/src/genesis/writer/orchestrator.py +++ b/src/genesis/writer/orchestrator.py @@ -6,6 +6,7 @@ Impact Agent MVP（2026-08-23）：门控 = 用户是否提供既有系统（exi """ from __future__ import annotations +import json import logging from datetime import date from pathlib import Path @@ -31,6 +32,34 @@ def _section_id_of(placeholder: str | None) -> str | None: return placeholder[len("section:"):] +def _load_snapshot_contents(snapshot_path: str | None) -> dict: + """读取章节快照的已生成内容；缺失/损坏 → 空 dict（从头生成，不阻断）。""" + if not snapshot_path: + return {} + path = Path(snapshot_path) + if not path.exists(): + return {} + try: + loaded = WriterState.from_dict(json.loads(path.read_text(encoding="utf-8"))) + except Exception: # 快照损坏不应阻断录制/生成 + _LOGGER.warning("章节快照损坏，忽略并从头生成：%s", snapshot_path) + return {} + return {cid: c for cid, c in loaded.contents.items() if c is not None} + + +def _write_snapshot(state: WriterState, snapshot_path: str | None) -> None: + """原子写章节快照（临时文件 + 替换），供重拍/续跑复用。""" + if not snapshot_path: + return + path = Path(snapshot_path) + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_suffix(path.suffix + ".tmp") + tmp.write_text( + json.dumps(state.to_dict(), ensure_ascii=False, indent=2), encoding="utf-8" + ) + tmp.replace(path) + + def _warn_unanchored(ctxs) -> None: """防静默丢章：对缺少 {{section:}} 锚点的章节打显式告警。 @@ -59,6 +88,7 @@ class WriteOrchestrator: meta: dict | None = None, output_language: str = "auto", chapter_attempts: int = 3, + snapshot_path: str | None = None, ) -> list[ChapterContent]: engine = engine or build_inference_engine() prompt_registry = prompt_registry or PromptRegistry() @@ -72,26 +102,25 @@ class WriteOrchestrator: ctxs = build_contexts(structured_source, samples_dir, output_language=output_language) _warn_unanchored(ctxs) state = WriterState([c.chapter_id for c in ctxs]) + # 快照命中：复用已生成章节，跳过 LLM（录制重拍不从头重跑） + for cid, cached in _load_snapshot_contents(snapshot_path).items(): + if cid in state.contents: + state.contents[cid] = cached + state.versions[cid] = 1 agent = WriterAgent(session_id=session_id, engine=engine, prompt_registry=prompt_registry, state=state) contents: list[ChapterContent] = [] sections: dict[str, list[Block]] = {} for ctx in ctxs: - # 章级管道重试（#1/#2）：真实 LLM 输出有随机方差，单章硬失败不连坐整次运行。 - # 每轮管道尝试内部已含 WriterAgent.max_retries 次 LLM 调用；chapter_attempts 为 - # 管道层兜底轮数（默认 3）。耗尽后仍抛错（不吞错）。 - content: ChapterContent | None = None - last_err: Exception | None = None - for attempt in range(max(1, chapter_attempts)): - try: - content = agent.generate_chapter(ctx) - break - except WriterGenerationError as e: - last_err = e - _LOGGER.warning("章节 %s 生成失败（第 %d/%d 轮管道重试）: %s", - ctx.chapter_id, attempt + 1, chapter_attempts, e) - if content is None: - raise WriterGenerationError(f"章节 {ctx.chapter_id} 管道重试耗尽: {last_err}") + # 章间引用（design.md §6.9）：注入此前已生成章节的摘要（串行生成保证前章已就绪） + ctx.prior_summaries = state.summaries_before(ctx.chapter_id) + content = state.contents.get(ctx.chapter_id) + if content is not None: + _LOGGER.info("章节 %s 命中快照，跳过 LLM 生成", ctx.chapter_id) + else: + content = self._generate_with_retries(agent, ctx, chapter_attempts) + _write_snapshot(state, snapshot_path) + _LOGGER.info("章节 %s 生成完成（%d 块）", ctx.chapter_id, len(content.blocks)) contents.append(content) blocks = render_chapter_blocks(content) sec_id = _section_id_of(ctx.template_marker.section_placeholder) @@ -111,3 +140,20 @@ class WriteOrchestrator: Path(output_path).parent.mkdir(parents=True, exist_ok=True) doc.save(output_path) return contents + + @staticmethod + def _generate_with_retries(agent: WriterAgent, ctx, chapter_attempts: int) -> ChapterContent: + """章级管道重试（#1/#2）：真实 LLM 输出有随机方差，单章硬失败不连坐整次运行。 + + 每轮管道尝试内部已含 WriterAgent.max_retries 次 LLM 调用；chapter_attempts 为 + 管道层兜底轮数（默认 3）。耗尽后仍抛错（不吞错）。 + """ + last_err: Exception | None = None + for attempt in range(max(1, chapter_attempts)): + try: + return agent.generate_chapter(ctx) + except WriterGenerationError as e: + last_err = e + _LOGGER.warning("章节 %s 生成失败（第 %d/%d 轮管道重试）: %s", + ctx.chapter_id, attempt + 1, chapter_attempts, e) + raise WriterGenerationError(f"章节 {ctx.chapter_id} 管道重试耗尽: {last_err}") diff --git a/src/genesis/writer/renderer.py b/src/genesis/writer/renderer.py index 27bb670..213b73b 100644 --- a/src/genesis/writer/renderer.py +++ b/src/genesis/writer/renderer.py @@ -1,7 +1,7 @@ """渲染：ChapterContent 的 ContentBlock 序列 → DocxInjector.Block 序列（Phase 5）。 -字段塌缩（外视 #6）：table.headers/caption、list.items/style 在渲染时显式丢弃， -仅保留 DocxInjector.Block 支持的 (kind, text, level, rows)。 +完整透传（外视 #6 反转）：table 的 headers/caption、list 的 items/style 全部保留， +使生成的 Word 表格带表头与边框、列表带项目符号，不再塌缩为纯文本。 """ from __future__ import annotations @@ -15,9 +15,15 @@ def render_chapter_blocks(content: ChapterContent) -> list[Block]: if b.type == "heading": out.append(Block(kind="heading", text=b.text or b.caption or "", level=b.level or 1)) elif b.type == "table": - out.append(Block(kind="table", text=b.caption or "", rows=b.rows or [])) + out.append(Block( + kind="table", text=b.caption or "", + headers=b.headers or [], rows=b.rows or [], + )) elif b.type == "list": - out.append(Block(kind="list", text="\n".join(b.items or []))) + out.append(Block( + kind="list", text="\n".join(b.items or []), + items=b.items or [], style=b.style or "bullet", + )) elif b.type == "note": out.append(Block(kind="note", text=b.text or b.caption or "")) else: # paragraph 及未知类型 diff --git a/src/genesis/writer/writer_agent.py b/src/genesis/writer/writer_agent.py index 1c4a795..1e9bf78 100644 --- a/src/genesis/writer/writer_agent.py +++ b/src/genesis/writer/writer_agent.py @@ -27,6 +27,7 @@ WRITER_PROMPT_TEMPLATE = ( "设计规则:\n{{design_rules}}\n" "模板样式:\n{{template_styles}}\n" "影响调查上下文:\n{{impact}}\n" + "前章摘要（已生成章节的要点，供本章跨章引用；勿重复展开）:\n{{prior_summaries}}\n" "参考资料（本章对应数据）:\n{{data}}\n" "请输出符合 schema 的章节内容 JSON。\n" "【小节约束】若上方「本章小节结构」非空，必须按该小节顺序组织内容，" @@ -81,13 +82,6 @@ class WriterAgent: self.state = state self.max_retries = max_retries - def _chunk_source(self, source) -> list[dict]: - if source is None: - return [{"index": 0, "text": ""}] - src = source if isinstance(source, str) else str(source) - n = max(1, len(src) // 1800 + 1) - return [{"index": i, "text": src[i * 1800:(i + 1) * 1800]} for i in range(n)] - def _resolve_prompt(self) -> Prompt: """取用/注册 writer.chapter 模板，并保证返回 Prompt 对象。 @@ -134,8 +128,6 @@ class WriterAgent: return result.data def generate_chapter(self, context: GenerationContext) -> ChapterContent: - self._chunk_source(context.structured_source) # 分块可用性验证（真实拼回留待后续） - # 步骤 A：推导本章期望输出语言（仅依赖 context，循环前置，稳定） # fallback 用「规则文档」（write_rules/design_rules，RAG 自日文作成说明书/记入规则检索， # 含假名可判日文）。不能用 impact/data（源数据）：样本含中文元素名（止损风控等）、 diff --git a/src/genesis/writer/writer_state.py b/src/genesis/writer/writer_state.py index 3a6630f..095ad4f 100644 --- a/src/genesis/writer/writer_state.py +++ b/src/genesis/writer/writer_state.py @@ -1,12 +1,16 @@ """Writer 跨章状态（Phase 5）。追踪各章版本/内容/最近评估结果。""" from __future__ import annotations -from genesis.writer.models import ChapterContent +import dataclasses + +from genesis.writer.models import ChapterContent, ContentBlock from genesis.eval.scorer import EvalReport class WriterState: def __init__(self, chapter_order: list[str]) -> None: + # 保留模板章节顺序，供章间引用（design.md §6.9）判断「前章」 + self.order: list[str] = list(chapter_order) self.versions: dict[str, int] = {cid: 0 for cid in chapter_order} self.contents: dict[str, ChapterContent | None] = {cid: None for cid in chapter_order} self.last_eval: dict[str, EvalReport | None] = {cid: None for cid in chapter_order} @@ -18,6 +22,59 @@ class WriterState: def record_eval(self, cid: str, report: EvalReport) -> None: self.last_eval[cid] = report + # ---------- 章间引用摘要（design.md §6.9） ---------- + + def summary_of(self, chapter_id: str) -> str | None: + """单章摘要：章 id + 标题 + 各表结构（表头 + 行数）。未生成 → None。""" + content = self.contents.get(chapter_id) + if content is None: + return None + parts = [f"[{chapter_id}] {content.title}"] + for b in content.blocks: + if b.type == "table" and b.headers: + parts.append(f" 表: {' | '.join(b.headers)}（{len(b.rows or [])} 行）") + return "\n".join(parts) + + def summaries_before(self, chapter_id: str) -> str: + """本章之前已生成章节的摘要集合（token 友好，供后章 prompt 引用）。""" + if chapter_id not in self.order: + return "" + idx = self.order.index(chapter_id) + return "\n".join( + s for cid in self.order[:idx] if (s := self.summary_of(cid)) + ) + + # ---------- 落盘快照（P0-b：录制重拍不从头重跑） ---------- + + def to_dict(self) -> dict: + """序列化为 JSON 友好 dict（仅含已生成章节）。""" + return { + "order": list(self.order), + "chapters": { + cid: { + "chapter_id": c.chapter_id, + "version": c.version, + "title": c.title, + "blocks": [dataclasses.asdict(b) for b in c.blocks], + } + for cid, c in self.contents.items() if c is not None + }, + } + + @classmethod + def from_dict(cls, data: dict) -> "WriterState": + state = cls(list(data.get("order", []))) + for cid, payload in (data.get("chapters") or {}).items(): + content = ChapterContent( + chapter_id=payload.get("chapter_id", cid), + version=payload.get("version", 1), + title=payload.get("title", ""), + blocks=[ContentBlock(**b) for b in payload.get("blocks", [])], + ) + state.contents[cid] = content + state.versions[cid] = state.versions.get(cid, 0) + 1 + return state + def needs_regeneration(self) -> list[str]: out: list[str] = [] for cid, content in self.contents.items(): diff --git a/tests/test_config.py b/tests/test_config.py index af7ba57..1ae7162 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -105,4 +105,12 @@ def test_rerank_env_override(monkeypatch): monkeypatch.setenv("GENESIS_RAG__RERANK__ENABLED", "false") s = Settings.from_dir(FIXTURES) assert s.rag.rerank.enabled is False - assert s.rag.rerank.model == "BAAI/bge-reranker-v2-m3" \ No newline at end of file + assert s.rag.rerank.model == "BAAI/bge-reranker-v2-m3" + + +# ---------- P2-2：allowed_extensions 不残留旧版 .xls（与解析层对齐） ---------- + +def test_default_allowed_extensions_excludes_legacy_xls(): + s = Settings() + assert ".xls" not in s.app.server.allowed_extensions + assert ".xlsx" in s.app.server.allowed_extensions \ No newline at end of file diff --git a/tests/test_docx_injector.py b/tests/test_docx_injector.py index 4303701..5d6e4f6 100644 --- a/tests/test_docx_injector.py +++ b/tests/test_docx_injector.py @@ -9,6 +9,7 @@ from __future__ import annotations import pytest from docx import Document +from docx.shared import Pt from genesis.writer.docx_injector import ( Block, @@ -31,7 +32,8 @@ def _section_blocks() -> list[Block]: return [ Block(kind="heading", text="3.1 テーブル一覧", level=2), Block(kind="paragraph", text="以下がDB表定义です。"), - Block(kind="table", rows=[["テーブル", "説明"], ["TB001", "社員"]]), + Block(kind="table", text="テーブル一覧", headers=["テーブル", "説明"], + rows=[["TB001", "社員"]]), ] @@ -183,3 +185,170 @@ def test_duplicate_subheading_with_content_not_removed(tmp_path): texts = [p.text for p in out.paragraphs] # 模板子节有内容 → 不删（保守策略：仅删除完全空的重复标题） assert "テーブル定義は別紙参照。" in texts + + +# ---------- 表格：表头行 + 边框 + 标题（外视 #6 反转） ---------- + +def test_table_header_row_rendered(tmp_path): + tpl = _make_template(tmp_path, "{{section:db_design}}") + out = DocxInjector(tpl).inject({"db_design": _section_blocks()}, {"doc_title": "T"}) + tbl = out.tables[0] + assert [c.text for c in tbl.rows[0].cells] == ["テーブル", "説明"] + assert [c.text for c in tbl.rows[1].cells] == ["TB001", "社員"] + + +def test_table_has_grid_borders(tmp_path): + tpl = _make_template(tmp_path, "{{section:db_design}}") + out = DocxInjector(tpl).inject({"db_design": _section_blocks()}, {"doc_title": "T"}) + assert out.tables[0].style is not None + assert out.tables[0].style.name == "Table Grid" + + +def test_table_caption_rendered(tmp_path): + tpl = _make_template(tmp_path, "{{section:db_design}}") + out = DocxInjector(tpl).inject({"db_design": _section_blocks()}, {"doc_title": "T"}) + assert any("テーブル一覧" in p.text for p in out.paragraphs) + + +# ---------- 列表：逐项 + 样式（外视 #6 反转） ---------- + +def test_list_block_uses_bullet_style(tmp_path): + tpl = _make_template(tmp_path, "{{section:x}}") + blocks = [Block(kind="list", items=["項目一", "項目二"], style="bullet")] + out = DocxInjector(tpl).inject({"x": blocks}, {"doc_title": "T"}) + bullets = [p for p in out.paragraphs if p.style.name == "List Bullet"] + assert [p.text for p in bullets] == ["項目一", "項目二"] + + +def test_list_block_uses_numbered_style(tmp_path): + tpl = _make_template(tmp_path, "{{section:x}}") + blocks = [Block(kind="list", items=["A", "B"], style="numbered")] + out = DocxInjector(tpl).inject({"x": blocks}, {"doc_title": "T"}) + numbered = [p for p in out.paragraphs if p.style.name == "List Number"] + assert [p.text for p in numbered] == ["A", "B"] + + +# ---------- 行内替换：保留字符格式 ---------- + +def test_inline_replacement_preserves_run_format(tmp_path): + doc = Document() + para = doc.add_paragraph() + run = para.add_run("標題：{{doc_title}}") + run.font.size = Pt(20) + run.bold = True + path = tmp_path / "tpl_fmt.docx" + doc.save(str(path)) + + out = DocxInjector(str(path)).inject({}, {"doc_title": "概要設計書"}) + out_para = out.paragraphs[0] + assert out_para.text == "標題：概要設計書" + assert out_para.runs[0].font.size == Pt(20) + assert out_para.runs[0].bold is True + + +# ---------- 防御分支：非数字 Heading / 无 run / 多 run 跨段 ---------- + +def test_heading_level_non_numeric_tail_returns_none(): + assert DocxInjector._heading_level("Heading X") is None + assert DocxInjector._heading_level(None) is None + + +def test_bare_subheading_followed_by_table_is_kept(tmp_path): + """裸子节后紧跟表格 → 非「纯空」（表格为其内容），保留不删。""" + doc = Document() + doc.add_paragraph("{{section:x}}") + doc.add_paragraph("補足", style="Heading 2") + doc.add_paragraph("") # 空段落 → 扫描继续 + doc.add_table(rows=1, cols=1) # 表格 → 终止「裸」判定 + path = tmp_path / "tpl_tbl.docx" + doc.save(str(path)) + out = DocxInjector(str(path)).inject({"x": []}, {"doc_title": "T"}) + assert any(p.text.strip() == "補足" for p in out.paragraphs) + + +def test_set_paragraph_text_adds_run_when_none(tmp_path): + """无 run 的段落：走 add_run 分支且不报错。""" + target = Document().add_paragraph() + DocxInjector(str(tmp_path / "unused.docx"))._set_paragraph_text(target, "新文本") + assert target.text == "新文本" + + +def test_inline_replacement_across_multiple_runs(tmp_path): + """行内占位符跨多个 run → 保留首个 run 格式并清空其余 run。""" + doc = Document() + para = doc.add_paragraph() + r1 = para.add_run("{{") + r1.bold = True + para.add_run("doc_title}}") + path = tmp_path / "tpl_multi.docx" + doc.save(str(path)) + out = DocxInjector(str(path)).inject({}, {"doc_title": "表題"}) + p = out.paragraphs[0] + assert p.text == "表題" + assert p.runs[0].bold is True + + +# ---------- P2-1：表格单元格内的占位符注入（此前只扫正文段落 → 漏注入/漏检残留） ---------- + +def _all_text(doc) -> str: + """正文段落 + 表格单元格段落的全部文本（用于单元格注入断言）。 + + 合并单元格的 row.cells 会重复指向同一 tc → 按 tc 去重，避免重复计数。 + """ + parts = [p.text for p in doc.paragraphs] + seen_tc: set = set() + for tbl in doc.tables: + for row in tbl.rows: + for cell in row.cells: + if id(cell._tc) in seen_tc: + continue + seen_tc.add(id(cell._tc)) + parts.extend(p.text for p in cell.paragraphs) + return "\n".join(parts) + + +def _cell_anchor_template(tmp_path, inner="{{section:db_design}}", name="tpl_cell.docx") -> str: + doc = Document() + doc.add_paragraph("{{doc_title}}") + tbl = doc.add_table(rows=1, cols=1) + tbl.rows[0].cells[0].paragraphs[0].text = inner + doc.add_paragraph("尾部固定内容") + path = tmp_path / name + doc.save(str(path)) + return str(path) + + +def test_section_placeholder_inside_table_cell_replaced(tmp_path): + tpl = _cell_anchor_template(tmp_path) + out = DocxInjector(tpl).inject({"db_design": _section_blocks()}, {"doc_title": "T"}) + text = _all_text(out) + assert "3.1 テーブル一覧" in text # 标题块注入进单元格 + assert "以下がDB表定义です。" in text + assert "{{section:db_design}}" not in text # 占位符已被替换 + + +def test_inline_meta_inside_table_cell_replaced(tmp_path): + tpl = _cell_anchor_template(tmp_path, inner="{{doc_title}}", name="tpl_cell_meta.docx") + out = DocxInjector(tpl).inject({}, {"doc_title": "表題"}) + text = _all_text(out) + assert "表題" in text and "{{doc_title}}" not in text + + +def test_residue_inside_table_cell_detected(tmp_path): + """单元格内未替换的占位符必须被残留检查捕获（否则静默丢内容）。""" + tpl = _cell_anchor_template(tmp_path, inner="{{section:unknown}}", name="tpl_cell_res.docx") + with pytest.raises(DocxInjectError, match="残留"): + DocxInjector(tpl).inject({}, {"doc_title": "T"}) + + +def test_merged_cell_placeholder_injected_once(tmp_path): + doc = Document() + tbl = doc.add_table(rows=1, cols=2) + tbl.rows[0].cells[0].merge(tbl.rows[0].cells[1]) + tbl.rows[0].cells[0].paragraphs[0].text = "{{section:x}}" + path = tmp_path / "tpl_merge.docx" + doc.save(str(path)) + + blocks = [Block(kind="paragraph", text="注入内容")] + out = DocxInjector(str(path)).inject({"x": blocks}, {}) + assert _all_text(out).count("注入内容") == 1 diff --git a/tests/test_excel_reader.py b/tests/test_excel_reader.py index e3b7694..8f0f850 100644 --- a/tests/test_excel_reader.py +++ b/tests/test_excel_reader.py @@ -17,8 +17,15 @@ def test_open_workbook_and_sheet_matrix(tmp_path): assert sheet_matrix(ws) == [["機能ID", "機能名"], ["F001", "社員登録"]] -def test_open_workbook_rejects_xls(tmp_path): +def test_open_workbook_rejects_legacy_xls_actionably(tmp_path): bad = tmp_path / "old.xls" bad.write_bytes(b"not really xls") - with pytest.raises(ValueError): + with pytest.raises(ValueError, match="另存为 .xlsx"): + open_workbook(bad) + + +def test_open_workbook_rejects_other_extension(tmp_path): + bad = tmp_path / "note.csv" + bad.write_text("a,b", encoding="utf-8") + with pytest.raises(ValueError, match="不支持的 Excel 格式"): open_workbook(bad) diff --git a/tests/test_phase5_models.py b/tests/test_phase5_models.py index 978e175..f45a62e 100644 --- a/tests/test_phase5_models.py +++ b/tests/test_phase5_models.py @@ -242,3 +242,15 @@ def test_to_vars_exposes_sub_headings(): def test_to_vars_sub_headings_empty_when_none(): vars_ = _ctx("db_design", "DB設計").to_vars() assert vars_["sub_headings"] == "" + + +# ---------- 章间引用摘要注入（design.md §6.9） ---------- + +def test_to_vars_exposes_prior_summaries(): + ctx = _ctx("db_design", "DB設計", source=None) + ctx.prior_summaries = "前章摘要" + assert ctx.to_vars()["prior_summaries"] == "前章摘要" + + +def test_to_vars_prior_summaries_defaults_empty(): + assert _ctx("db_design", "DB設計", source=None).to_vars()["prior_summaries"] == "" diff --git a/tests/test_phase5_renderer.py b/tests/test_phase5_renderer.py index c6545d9..93c8c11 100644 --- a/tests/test_phase5_renderer.py +++ b/tests/test_phase5_renderer.py @@ -6,8 +6,9 @@ def _content(): blocks = [ ContentBlock(block_id="1", type="heading", level=2, text="小節"), ContentBlock(block_id="2", type="paragraph", text="正文"), - ContentBlock(block_id="3", type="table", caption="表1", rows=[["a", "b"], ["1", "2"]]), - ContentBlock(block_id="4", type="list", items=["項目一", "項目二"]), + ContentBlock(block_id="3", type="table", caption="表1", + headers=["見出しA", "見出しB"], rows=[["a", "b"], ["1", "2"]]), + ContentBlock(block_id="4", type="list", items=["項目一", "項目二"], style="bullet"), ContentBlock(block_id="5", type="note", text="注意"), ] return ChapterContent(chapter_id="db_design", version=1, title="DB 設計", blocks=blocks) @@ -17,9 +18,21 @@ def test_render_maps_all_block_types(): out = render_chapter_blocks(_content()) kinds = [b.kind for b in out] assert kinds == ["heading", "paragraph", "table", "list", "note"] - # 表：rows 透传 + + +def test_render_preserves_table_headers_and_caption(): + # 外视 #6 反转：表头/标题不再塌缩，须透传至注入器 + out = render_chapter_blocks(_content()) table = out[2] assert table.kind == "table" + assert table.headers == ["見出しA", "見出しB"] + assert table.text == "表1" # caption assert table.rows == [["a", "b"], ["1", "2"]] - # 列表：items 拼接进 text - assert "項目一" in out[3].text and "項目二" in out[3].text + + +def test_render_preserves_list_items_and_style(): + out = render_chapter_blocks(_content()) + lst = out[3] + assert lst.kind == "list" + assert lst.items == ["項目一", "項目二"] + assert lst.style == "bullet" diff --git a/tests/test_phase5_writer_agent.py b/tests/test_phase5_writer_agent.py index 615075e..b4fe6d3 100644 --- a/tests/test_phase5_writer_agent.py +++ b/tests/test_phase5_writer_agent.py @@ -38,7 +38,7 @@ def _ctx(cid, title): write_rules=["W1"], design_rules=["D1"], template_styles={"Heading1"}, - prior_state=None, + prior_summaries="", ) diff --git a/tests/test_phase5_writer_orchestrator.py b/tests/test_phase5_writer_orchestrator.py index 4aaae33..81fc7bc 100644 --- a/tests/test_phase5_writer_orchestrator.py +++ b/tests/test_phase5_writer_orchestrator.py @@ -202,3 +202,89 @@ def test_generate_missing_template_path_raises(tmp_path): SimpleNamespace(template=parsed), str(tmp_path / "out.docx"), samples_dir="nonexistent_dir_xyz", engine=FakeEngine(), ) + + +# ---------- 章间引用（design.md §6.9：后章 prompt 注入前章摘要） ---------- + +class RecordingEngine: + """记录每次调用的 variables，并返回含表格的章节内容（供摘要提取）。""" + + def __init__(self): + self.calls = [] + + def chat_structured(self, *, session_id, prompt, variables, schema, retry_count=2): + self.calls.append(variables) + return SimpleNamespace( + data={"title": variables["title"], "blocks": [ + {"type": "paragraph", "text": f"{variables['title']}の内容"}, + {"type": "table", "headers": ["ID", "名称"], "rows": [["1", "a"]]}, + ]}, + status="ok", + ) + + +def _two_chapter_template(path): + doc = Document() + doc.add_paragraph("第1章 機能一覧", style="Heading 1") + doc.add_paragraph("{{section:function_list}}") + doc.add_paragraph("第2章 画面一覧", style="Heading 1") + doc.add_paragraph("{{section:screen_list}}") + doc.save(path) + + +def test_prior_chapter_summaries_injected_for_later_chapters(tmp_path): + tpl = tmp_path / "tpl2.docx" + out = tmp_path / "out2.docx" + _two_chapter_template(str(tpl)) + parsed = ParsedTemplate( + file_name=str(tpl), + sections=[ + ChapterMarker(type="heading", name="第1章 機能一覧", level=1), + ChapterMarker(type="placeholder", name="section:function_list", level=0), + ChapterMarker(type="heading", name="第2章 画面一覧", level=1), + ChapterMarker(type="placeholder", name="section:screen_list", level=0), + ], + placeholders={}, styles={"defined": ["Heading 1"], "used": ["Heading 1"]}, + ) + engine = RecordingEngine() + WriteOrchestrator().generate( + SimpleNamespace(template=parsed), str(out), + samples_dir="nonexistent_dir_xyz", engine=engine, template_path=str(tpl), + ) + assert len(engine.calls) == 2 + # 第1章无前章摘要 + assert engine.calls[0]["prior_summaries"] == "" + # 第2章含第1章摘要（含前章标题与表结构） + second = engine.calls[1]["prior_summaries"] + assert "function_list" in second + assert "機能一覧" in second + assert "ID" in second and "名称" in second + + +def test_writer_prompt_template_exposes_prior_summaries(): + from genesis.writer.writer_agent import WRITER_PROMPT_TEMPLATE + assert "{{prior_summaries}}" in WRITER_PROMPT_TEMPLATE + + +def test_template_with_cell_anchor_injects_chapter(tmp_path): + """P2-1 端到端：锚点位于表格单元格时，章节内容注入该单元格而非被丢弃。""" + from genesis.parsers.word_template_parser import WordTemplateParser + + doc = Document() + doc.add_paragraph("2. 機能一覧", style="Heading 1") + tbl = doc.add_table(rows=1, cols=1) + tbl.rows[0].cells[0].paragraphs[0].text = "{{section:function_list}}" + tpl = tmp_path / "tpl_cell.docx" + doc.save(str(tpl)) + + parsed = WordTemplateParser().parse(str(tpl)) + out = tmp_path / "out_cell.docx" + contents = WriteOrchestrator().generate( + SimpleNamespace(template=parsed), str(out), + samples_dir="nonexistent_dir_xyz", engine=RecordingEngine(), + template_path=str(tpl), + ) + assert len(contents) == 1 + loaded = Document(str(out)) + cell_text = "\n".join(c.text for t in loaded.tables for r in t.rows for c in r.cells) + assert "機能一覧の内容" in cell_text diff --git a/tests/test_phase5_writer_state.py b/tests/test_phase5_writer_state.py index dbbb0fb..ebae3e9 100644 --- a/tests/test_phase5_writer_state.py +++ b/tests/test_phase5_writer_state.py @@ -1,6 +1,6 @@ """WriterState 测试（P5-T5）。""" from genesis.writer.writer_state import WriterState -from genesis.writer.models import ChapterContent +from genesis.writer.models import ChapterContent, ContentBlock from genesis.eval.scorer import EvalReport @@ -30,3 +30,40 @@ def test_failed_eval_marks_regeneration(): st.record_success(_content("a")) st.record_eval("a", _report(["a"])) assert st.needs_regeneration() == ["a"] + + +# ---------- 章间引用摘要（design.md §6.9） ---------- + +def _content_with_table(cid, title, headers, nrows): + rows = [[str(i) for _ in headers] for i in range(nrows)] + return ChapterContent(chapter_id=cid, version=1, title=title, blocks=[ + ContentBlock(block_id="b1", type="table", headers=headers, rows=rows), + ]) + + +def test_summary_of_includes_title_and_table_structure(): + st = WriterState(["a"]) + st.record_success(_content_with_table("a", "機能一覧", ["機能ID", "機能名"], 3)) + s = st.summary_of("a") + assert "a" in s and "機能一覧" in s + assert "機能ID" in s and "機能名" in s + assert "3" in s + + +def test_summary_of_none_when_not_generated(): + st = WriterState(["a"]) + assert st.summary_of("a") is None + + +def test_summaries_before_respects_order(): + st = WriterState(["a", "b", "c"]) + st.record_success(_content_with_table("a", "第一章", ["ID"], 1)) + st.record_success(_content_with_table("b", "第二章", ["ID"], 2)) + before_c = st.summaries_before("c") + assert "第一章" in before_c and "第二章" in before_c + assert st.summaries_before("a") == "" + + +def test_summaries_before_unknown_chapter_empty(): + st = WriterState(["a"]) + assert st.summaries_before("zzz") == "" diff --git a/tests/test_server_service.py b/tests/test_server_service.py index 3c9c642..06393a9 100644 --- a/tests/test_server_service.py +++ b/tests/test_server_service.py @@ -75,6 +75,33 @@ def test_create_session_name_and_project(tmp_path): assert got.project == "projA" +# ---------- P2-2：上传即校验扩展名（EXPECTED_SUFFIX 生效；.xls 显式拒绝） ---------- + +def test_upload_rejects_legacy_xls_requirements(svc): + s = svc.create_session("u1") + with pytest.raises(FileTypeError, match="另存为 .xlsx"): + svc.upload_file(s.session_id, "requirements", "old.xls", b"legacy-biff") + + +def test_upload_rejects_xlsx_rules_aligning_with_parser(svc): + """rules 实际仅支持 .docx（source_aggregator 侧）→ 上传即拒绝 .xlsx。""" + s = svc.create_session("u1") + with pytest.raises(FileTypeError, match="不支持的扩展名"): + svc.upload_file(s.session_id, "rules", "rules.xlsx", b"x") + + +def test_upload_rejects_wrong_suffix_for_template(svc): + s = svc.create_session("u1") + with pytest.raises(FileTypeError, match="不支持的扩展名"): + svc.upload_file(s.session_id, "template", "tpl.xlsx", b"x") + + +def test_upload_accepts_docx_rules(svc): + s = svc.create_session("u1") + svc.upload_file(s.session_id, "rules", "rules_entry_ja.docx", b"PK\x03\x04") + assert svc.get_session(s.session_id).files["rules"]["name"] == "rules_entry_ja.docx" + + def test_upload_requirements_auto_names_session(tmp_path): svc = GenesisService( store=SessionStore(db_path=str(tmp_path / "s.db")), diff --git a/tests/test_source_aggregator.py b/tests/test_source_aggregator.py index 48c281d..b6ea89a 100644 --- a/tests/test_source_aggregator.py +++ b/tests/test_source_aggregator.py @@ -108,6 +108,21 @@ def test_parse_extensionless_rule(tmp_path): SourceParser().parse(rule_paths=[bad]) +# ---------- P2-2：旧版 .xls 显式拒绝（可操作消息，非含糊格式错误） ---------- + +def test_parse_legacy_xls_requirement_rejected_actionably(tmp_path): + bad = tmp_path / "old.xls" + bad.write_bytes(b"legacy-biff") + with pytest.raises(ValueError, match="另存为 .xlsx"): + SourceParser().parse(requirement_paths=[bad]) + + +def test_xlsx_exts_excludes_legacy_xls(): + from genesis.parsers.source_aggregator import LEGACY_EXCEL_EXTS, XLSX_EXTS + assert XLSX_EXTS == (".xlsx",) + assert ".xls" in LEGACY_EXCEL_EXTS + + # ---------- T7: DRY _validate_path helper（I8） ---------- def test_validate_path_rejects_bad_extension(tmp_path): diff --git a/tests/test_word_template_parser.py b/tests/test_word_template_parser.py index 3fbc730..cbb346d 100644 --- a/tests/test_word_template_parser.py +++ b/tests/test_word_template_parser.py @@ -130,3 +130,50 @@ def test_parse_extracts_placeholder_fullwidth_colon(tmp_path): assert result.placeholders == {"section:2": "{{section：2}}"} ph = [s for s in result.sections if s.type == "placeholder"] assert [s.name for s in ph] == ["section:2"] + + +# ---------- P2-1：表格单元格内的占位符（此前漏检 → 静默丢章） ---------- + +def _doc_with_cell_placeholder(): + """H1 → 表格(单元格内含锚点) → H1。锚点必须归属其前的 H1。""" + doc = new_document() + doc.add_heading("1. はじめに", level=1) + tbl = doc.add_table(rows=1, cols=1) + tbl.rows[0].cells[0].paragraphs[0].text = "{{section:introduction}}" + doc.add_heading("2. 機能一覧", level=1) + doc.add_paragraph("{{section:function_list}}") + return doc + + +def test_parse_placeholder_inside_table_cell(tmp_path): + path = save_document(tmp_path, _doc_with_cell_placeholder()) + result = WordTemplateParser().parse(path) + assert "section:introduction" in result.placeholders + ph = [s.name for s in result.sections if s.type == "placeholder"] + assert "section:introduction" in ph + + +def test_parse_table_placeholder_keeps_document_order(tmp_path): + """单元格锚点须按文档顺序落位（在其前的 H1 与在其后的 H1 之间），不得错绑。""" + path = save_document(tmp_path, _doc_with_cell_placeholder()) + result = WordTemplateParser().parse(path) + names = [(s.type, s.name) for s in result.sections] + i_intro = names.index(("heading", "1. はじめに")) + i_ph = names.index(("placeholder", "section:introduction")) + i_func = names.index(("heading", "2. 機能一覧")) + assert i_intro < i_ph < i_func + + +def test_parse_merged_cell_placeholder_not_duplicated(tmp_path): + """合并单元格的 row.cells 会重复指向同一 tc → 不得重复收集。""" + doc = new_document() + doc.add_heading("1. 章", level=1) + tbl = doc.add_table(rows=1, cols=2) + tbl.rows[0].cells[0].merge(tbl.rows[0].cells[1]) + tbl.rows[0].cells[0].paragraphs[0].text = "{{section:db_design}}" + path = save_document(tmp_path, doc) + + result = WordTemplateParser().parse(path) + + ph = [s.name for s in result.sections if s.type == "placeholder"] + assert ph.count("section:db_design") == 1 diff --git a/tests/test_writer_agent_branches.py b/tests/test_writer_agent_branches.py new file mode 100644 index 0000000..a880f9a --- /dev/null +++ b/tests/test_writer_agent_branches.py @@ -0,0 +1,57 @@ +"""WriterAgent 防御分支覆盖（P0 补测）。 + +覆盖：_resolve_prompt 直返回 Prompt、引擎异常状态无详情、LLM 返回非法 data 触发重试。 +""" +import pytest +from types import SimpleNamespace + +from genesis.inference.types import Prompt +from genesis.writer.writer_agent import WriterAgent, WriterGenerationError +from genesis.writer.writer_state import WriterState +from genesis.writer.models import ChapterSpec, GenerationContext + + +class _Registry: + def get_or_create(self, name, template): + return Prompt(name=name, version="1", template=template) + + +class _FailedEngine: + def chat_structured(self, **kw): + return SimpleNamespace(status="failed", data=None) # 无 error / error_code + + +class _BadDataEngine: + def chat_structured(self, **kw): + return SimpleNamespace(status="ok", data={"title": "x", "blocks": None}) + + +def _ctx(): + return GenerationContext( + chapter_id="db_design", title="DB 設計", + template_marker=ChapterSpec(chapter_id="db_design", title="DB 設計", + section_placeholder=None), + structured_source="src", write_rules=["W"], design_rules=["D"], + template_styles={"Heading 1"}, output_language="ja", + ) + + +def _agent(engine, max_retries=1): + return WriterAgent( + session_id="s", engine=engine, prompt_registry=_Registry(), + state=WriterState(["db_design"]), max_retries=max_retries, + ) + + +def test_resolve_prompt_passthrough_when_registry_returns_prompt(): + assert isinstance(_agent(_FailedEngine())._resolve_prompt(), Prompt) + + +def test_engine_failed_status_without_details_raises(): + with pytest.raises(WriterGenerationError, match="failed"): + _agent(_FailedEngine()).generate_chapter(_ctx()) + + +def test_invalid_llm_data_triggers_retry_and_raises(): + with pytest.raises(WriterGenerationError, match="重试耗尽"): + _agent(_BadDataEngine()).generate_chapter(_ctx()) diff --git a/tests/test_writer_data_budget.py b/tests/test_writer_data_budget.py new file mode 100644 index 0000000..250b97f --- /dev/null +++ b/tests/test_writer_data_budget.py @@ -0,0 +1,119 @@ +"""Writer 章节数据 token 预算（OV5 DataGate 接入）。 + +背景：design.md §6.8 ① 要求 `DataGate.load(structured_source, selector=该章数据)`， +但此前 `_format_chapter_data` 直接渲染、未过 DataGate，且 writer_agent 的 +`_chunk_source` 只做「可用性验证」不生效（死代码）。本文件验证： + - 章节数据超出 token 预算时按「最大表优先」省略，并留下省略标记（不抛错、不阻断生成） + - 预算可由 GenerationContext.data_token_budget 注入 + - 死代码 _chunk_source 已移除 +""" +from genesis.data_models import ( + CellValue, + ExcelTable, + ParsedTemplate, + ChapterMarker, + Provenance, + SheetType, + StructuredSource, +) +from genesis.orchestrator.datagate import DataGate, DataSelector +from genesis.writer.models import ChapterSpec, GenerationContext +from genesis.writer.writer_agent import WriterAgent + + +def _cell(v): + return CellValue(value=v, provenance=Provenance("要件.xlsx", "s1", 1, "A", "列")) + + +def _table(name, sheet_type, headers, rows): + return ExcelTable( + name=name, + detected_type=sheet_type, + extraction_method="openpyxl", + headers=headers, + rows=[{h: _cell(v) for h, v in zip(headers, row)} for row in rows], + ) + + +def _source(tables): + template = ParsedTemplate( + file_name="t.docx", + sections=[ChapterMarker(type="heading", name="x", level=1)], + placeholders={}, + styles={"used": []}, + ) + return StructuredSource( + tables=tables, template=template, rule_docs=[], image_analyses=[], + existing_system=None, comments=[], + ) + + +def _ctx(chapter_id, source, budget=None): + kwargs = {} if budget is None else {"data_token_budget": budget} + return GenerationContext( + chapter_id=chapter_id, title="機能一覧", + template_marker=ChapterSpec( + chapter_id=chapter_id, title="機能一覧", + section_placeholder=f"section:{chapter_id}", + ), + structured_source=source, + write_rules=[], design_rules=[], template_styles=set(), + **kwargs, + ) + + +def _small_table(): + return _table("機能一覧", SheetType.FUNCTION, ["機能ID", "機能名"], [["F001", "止損"]]) + + +def _big_table(): + rows = [[f"F{i:03d}", "非常に長い説明文" * 20] for i in range(200)] + return _table("大量機能", SheetType.FUNCTION, ["機能ID", "説明"], rows) + + +# ---------- 预算常量 ---------- + +def test_default_data_token_budget_exposed(): + from genesis.writer.models import MAX_CHAPTER_DATA_TOKENS + assert MAX_CHAPTER_DATA_TOKENS == 8_000 + + +# ---------- 超预算：省略 + 标记（不抛错） ---------- + +def test_tiny_budget_omits_all_tables_with_marker(): + src = _source([_small_table()]) + data = _ctx("function_list", src, budget=1).to_vars()["data"] + # 预算=1 → 无表可留 → 仅省略标记，且不抛 DataGateError + assert data == "" or "省略" in data + assert "F001" not in data + + +def test_budget_drops_largest_table_and_keeps_smaller(): + small, big = _small_table(), _big_table() + src = _source([small, big]) + # 测量用 gate 需足够大（大表单独即超默认 8000 预算，否则测量本身会抛错） + gate = DataGate(max_total_tokens=10 ** 9) + small_tokens = gate.load(src, DataSelector(table_ids=["機能一覧"])).token_estimate + big_tokens = gate.load(src, DataSelector(table_ids=["大量機能"])).token_estimate + assert big_tokens > small_tokens + + # 预算仅够小表 → 大表被省略，小表内容保留，标记点名被省略的表 + data = _ctx("function_list", src, budget=small_tokens + 1).to_vars()["data"] + assert "F001" in data and "止損" in data + assert "大量機能" not in data or "省略" in data + assert "省略" in data and "大量機能" in data + + +# ---------- 预算内：原样渲染 ---------- + +def test_within_budget_renders_without_marker(): + src = _source([_small_table()]) + data = _ctx("function_list", src).to_vars()["data"] + assert "F001" in data and "止損" in data + assert "省略" not in data + + +# ---------- 死代码已移除 ---------- + +def test_writer_agent_has_no_dead_chunk_source(): + assert not hasattr(WriterAgent, "_chunk_source") diff --git a/tests/test_writer_snapshot.py b/tests/test_writer_snapshot.py new file mode 100644 index 0000000..df29ca0 --- /dev/null +++ b/tests/test_writer_snapshot.py @@ -0,0 +1,141 @@ +"""录制稳健性：章节结果落盘快照 + 可恢复（P0-b）。 + +真实 LLM 长耗时串行生成，录制重拍时不应从头重跑。本文件验证： + - 每章生成后写快照（JSON） + - 重跑时命中快照的章节跳过 LLM（即使引擎不可用也能完成文档） + - 快照损坏时忽略并从头生成 + - WriterState 可 JSON 往返 +""" +import json +from types import SimpleNamespace + +from docx import Document + +from genesis.data_models import ChapterMarker, ParsedTemplate +from genesis.writer.orchestrator import WriteOrchestrator +from genesis.writer.writer_state import WriterState +from genesis.writer.models import ChapterContent, ContentBlock + + +class RecordingEngine: + def __init__(self): + self.calls = [] + + def chat_structured(self, *, session_id, prompt, variables, schema, retry_count=2): + self.calls.append(variables["title"]) + return SimpleNamespace( + data={"title": variables["title"], "blocks": [ + {"type": "paragraph", "text": f"{variables['title']}の内容"}, + {"type": "table", "headers": ["ID", "名称"], "rows": [["1", "a"]]}, + ]}, + status="ok", + ) + + +class ExplodingEngine: + """若被调用即失败：用于证明快照命中时不再走 LLM。""" + + def chat_structured(self, **kwargs): + raise AssertionError("不应调用 LLM（章节快照应命中）") + + +def _two_chapter_template(path): + doc = Document() + doc.add_paragraph("第1章 機能一覧", style="Heading 1") + doc.add_paragraph("{{section:function_list}}") + doc.add_paragraph("第2章 画面一覧", style="Heading 1") + doc.add_paragraph("{{section:screen_list}}") + doc.save(path) + + +def _parsed(tpl): + return ParsedTemplate( + file_name=tpl, + sections=[ + ChapterMarker(type="heading", name="第1章 機能一覧", level=1), + ChapterMarker(type="placeholder", name="section:function_list", level=0), + ChapterMarker(type="heading", name="第2章 画面一覧", level=1), + ChapterMarker(type="placeholder", name="section:screen_list", level=0), + ], + placeholders={}, styles={"defined": ["Heading 1"], "used": ["Heading 1"]}, + ) + + +def _run(tpl, out, engine, snap): + return WriteOrchestrator().generate( + SimpleNamespace(template=_parsed(tpl)), str(out), + samples_dir="nonexistent_dir_xyz", engine=engine, + template_path=str(tpl), snapshot_path=str(snap), + ) + + +def test_snapshot_written_with_all_chapters(tmp_path): + tpl, out, snap = tmp_path / "t.docx", tmp_path / "o.docx", tmp_path / "snap.json" + _two_chapter_template(str(tpl)) + _run(tpl, out, RecordingEngine(), snap) + + payload = json.loads(snap.read_text(encoding="utf-8")) + assert set(payload["chapters"]) == {"function_list", "screen_list"} + assert payload["chapters"]["function_list"]["blocks"][0]["type"] == "paragraph" + + +def test_resume_from_snapshot_skips_llm(tmp_path): + tpl, out1, snap = tmp_path / "t.docx", tmp_path / "o1.docx", tmp_path / "snap.json" + _two_chapter_template(str(tpl)) + _run(tpl, out1, RecordingEngine(), snap) + + # 第二次：引擎会爆炸，但快照命中 → 不调用 LLM，仍产出完整文档 + out2 = tmp_path / "o2.docx" + _run(tpl, out2, ExplodingEngine(), snap) + joined = "\n".join(p.text for p in Document(str(out2)).paragraphs) + assert "第1章 機能一覧の内容" in joined + assert "第2章 画面一覧の内容" in joined + + +def test_corrupt_snapshot_ignored_and_regenerated(tmp_path): + tpl, out, snap = tmp_path / "t.docx", tmp_path / "o.docx", tmp_path / "snap.json" + _two_chapter_template(str(tpl)) + snap.write_text("{ this is not json", encoding="utf-8") + + engine = RecordingEngine() + _run(tpl, out, engine, snap) + assert len(engine.calls) == 2 # 损坏快照 → 从头生成 + assert json.loads(snap.read_text(encoding="utf-8"))["chapters"] # 并重写有效快照 + + +def test_snapshot_with_extra_chapter_ignored(tmp_path): + """快照含当前模板没有的章节 id → 忽略该章，其余照常生成。""" + tpl, out, snap = tmp_path / "t.docx", tmp_path / "o.docx", tmp_path / "snap.json" + _two_chapter_template(str(tpl)) + snap.write_text(json.dumps({ + "order": ["function_list", "screen_list", "ghost"], + "chapters": {"ghost": {"chapter_id": "ghost", "version": 1, + "title": "幽灵章", "blocks": []}}, + }, ensure_ascii=False), encoding="utf-8") + + engine = RecordingEngine() + _run(tpl, out, engine, snap) + assert len(engine.calls) == 2 # 两章均重新生成，幽灵章被忽略 + + +def test_generate_with_explicit_meta_skips_defaults(tmp_path): + """显式传入 meta → 不再构造默认 meta（覆盖分支）。""" + tpl, out = tmp_path / "t.docx", tmp_path / "o.docx" + _two_chapter_template(str(tpl)) + contents = WriteOrchestrator().generate( + SimpleNamespace(template=_parsed(tpl)), str(out), + samples_dir="nonexistent_dir_xyz", engine=RecordingEngine(), + template_path=str(tpl), + meta={"doc_title": "T", "version": "v9", "created_at": "2026-01-01"}, + ) + assert len(contents) == 2 + + +def test_writer_state_dict_roundtrip(): + st = WriterState(["a"]) + st.record_success(ChapterContent(chapter_id="a", version=1, title="T", blocks=[ + ContentBlock(block_id="1", type="table", headers=["H1", "H2"], rows=[["x", "y"]]), + ])) + st2 = WriterState.from_dict(st.to_dict()) + assert st2.summary_of("a") == st.summary_of("a") + assert st2.to_dict()["chapters"]["a"]["blocks"][0]["headers"] == ["H1", "H2"]