feat(impact): Impact Agent MVP —— 变更点定位 + 影响调查书（追加改修场景）

- 门控：用户提供 existing_system 路径 → 进入影响调查；未提供 → 原流程不变
- CodeParser 解析 Java（@RestController/@Service/@Entity/@TableName）+ ExistingSystemExplorer 组装
- ImpactAgent 变更点定位（变更区分×既存対応 确定性比对，无 LLM）→ ImpactReport（JSON 可下载）
- 影响调查结果作为 Writer 生成概要设计书的主上下文（{{impact}}，无专用影响章）
- source_aggregator 解除 existing_system=None 硬编码
- 既有系统样本 sunOnly/stock-trade-system（无 LICENSE，仅测试输入，保留来源标注）
- 新造股票交易域追加改修样本 要件定義_追加改修_股票.xlsx（对齐 sunOnly 真实类名）
- 全量 351 passed / 99.27% 覆盖；门禁 PASS（16 要素：新规5/変更8/削除3/未受影响50）

src/genesis/data_models.py

@@ -175,6 +175,53 @@ class ExistingSystemInfo:
     source_path: str
 
 
+class ChangeType(Enum):
+    """变更点定位的变更区分（对应要件定義 変更区分 列值）"""
+    NEW = "新規"
+    MODIFIED = "変更"
+    DELETED = "削除"
+    UNCHANGED = "不变"
+
+
+@dataclass
+class ChangeElement:
+    """变更点定位结果中的一个要素（Impact Agent MVP）"""
+    element_id: str
+    element_type: str          # 機能/画面/帳票/DB/IF/バッチ（取 ElementType.value 或表名）
+    name: str
+    change_type: ChangeType
+    existing_mapping: list[str] = field(default_factory=list)   # 既存対応 声明值（类名清单）
+    impacted_existing: list[str] = field(default_factory=list)  # 确认命中的既有类
+    evidence: str = ""         # 命中的既有类 source_uri / 空
+    status: str = "ok"         # "ok" | "conflict" | "warning"
+
+
+@dataclass
+class ImpactWarning:
+    """影响调查告警（不阻断，供用户/QA 关注）"""
+    element_id: str
+    issue: str
+
+
+@dataclass
+class ChangeAnalysis:
+    """变更点定位结果集合（Impact Agent MVP）"""
+    project_type: str                       # "enhancement"（追加改修）
+    new_elements: list[ChangeElement]
+    modified_elements: list[ChangeElement]
+    deleted_elements: list[ChangeElement]
+    unchanged_elements: list[ChangeElement]
+    warnings: list[ImpactWarning]
+
+
+@dataclass
+class ImpactReport:
+    """影响调查书（MVP 子集，供 Writer 生成 + 独立下载）"""
+    metadata: dict
+    change_analysis: ChangeAnalysis | None = None
+    summary: dict = field(default_factory=dict)
+
+
 @dataclass
 class UnifiedDocument:
     """FileReader 的统一输出（多格式归一化）"""
@@ -223,6 +270,7 @@ class StructuredSource:
     image_analyses: list[ImageAnalysis]
     existing_system: ExistingSystemInfo | None
     comments: list[CellComment]
+    impact_report: "ImpactReport | None" = None  # 影响调查书（生成后回填，门控未提供时为 None）
 
 
 @dataclass

src/genesis/impact/code_parser.py

@@ -0,0 +1,186 @@
+"""CodeParser：Java 项目源码解析 → CodeStructure（Impact Agent MVP）。
+
+识别 @RestController/@Controller、@Service、@Entity/@Table 及方法级路由映射，
+输出供 ExistingSystemExplorer 组装 ExistingSystemInfo 的结构化清单。
+"""
+from __future__ import annotations
+
+import re
+from pathlib import Path
+
+from genesis.data_models import (
+    CodeStructure,
+    ControllerInfo,
+    EndpointInfo,
+    EntityInfo,
+    ServiceInfo,
+)
+
+
+class CodeParseError(Exception):
+    """既有系统解析失败（路径无效或非 Java 源码项目）。"""
+
+
+_JAVA_EXT = ".java"
+
+_CLASS_RE = re.compile(
+    r"(?:public\s+|abstract\s+|final\s+)?(?:class|interface|enum|record)\s+(\w+)"
+)
+_TABLE_RE = re.compile(r"@(?:Table|TableName)\s*\(\s*(?:name\s*=\s*)?[\"']([^\"']+)[\"']")
+_CLASS_MAPPING_RE = re.compile(r"@RequestMapping\s*\(\s*[\"']([^\"']+)[\"']")
+_METHOD_MAPPING_RE = re.compile(
+    r"@(Get|Post|Put|Delete|Patch|Request)Mapping\s*(?:\(\s*[\"']([^\"']*)[\"'])?"
+)
+_METHOD_DECL_RE = re.compile(
+    r"(?:public|private|protected|)\s+(?:static\s+|final\s+|synchronized\s+)*"
+    r"[\w<>\[\],.?]+\s+(\w+)\s*\("
+)
+_FIELD_DECL_RE = re.compile(
+    r"(?:private|public|protected)\s+[\w<>\[\],]+\s+(\w+)\s*;"
+)
+_IMPORT_RE = re.compile(r"^import\s+([\w.]+);", re.MULTILINE)
+
+
+def _relative(path: Path, root: Path) -> str:
+    return path.relative_to(root).as_posix()
+
+
+def _read(path: Path) -> str:
+    return path.read_text(encoding="utf-8", errors="ignore")
+
+
+class CodeParser:
+    """解析 Java 项目目录，输出 CodeStructure（控制器/服务/实体/端点/模块）。"""
+
+    def parse(self, root_path: str | Path) -> CodeStructure:
+        root = Path(root_path)
+        if not root.is_dir():
+            raise CodeParseError(f"既有系统路径无效或不存在: {root_path}")
+
+        java_files = sorted(p for p in root.rglob(f"*{_JAVA_EXT}") if p.is_file())
+        if not java_files:
+            raise CodeParseError(f"未找到 Java 源码: {root_path}")
+
+        controllers: list[ControllerInfo] = []
+        services: list[ServiceInfo] = []
+        entities: list[EntityInfo] = []
+        endpoints: list[EndpointInfo] = []
+        classes: list[dict] = []
+        raw_imports: list[dict] = []
+
+        for path in java_files:
+            text = _read(path)
+            rel = _relative(path, root)
+            class_name = self._class_name(text)
+
+            imports = _IMPORT_RE.findall(text)
+            raw_imports.append({"path": rel, "imports": imports})
+
+            if not class_name:
+                # package-info.java 等无类声明文件：仅登记 imports，不参与要素提取
+                continue
+
+            is_controller = "@RestController" in text or "@Controller" in text
+            is_service = "@Service" in text
+            # 既有系统实体可能用 JPA @Entity 或 MyBatis-Plus @TableName 标注
+            is_entity = "@Entity" in text or "@TableName" in text
+
+            if is_controller:
+                controllers.append(self._parse_controller(text, rel, class_name, endpoints))
+            elif is_service:
+                services.append(self._parse_service(text, rel, class_name))
+            elif is_entity:
+                entities.append(self._parse_entity(text, rel, class_name))
+
+            classes.append({"class_name": class_name, "path": rel})
+
+        modules = self._modules(root, java_files)
+
+        return CodeStructure(
+            root_path=str(root),
+            language="java",
+            modules=modules,
+            classes=classes,
+            controllers=controllers,
+            services=services,
+            entities=entities,
+            endpoints=endpoints,
+            raw_imports=raw_imports,
+        )
+
+    @staticmethod
+    def _class_name(text: str) -> str | None:
+        m = _CLASS_RE.search(text)
+        return m.group(1) if m else None
+
+    @staticmethod
+    def _parse_controller(
+        text: str, rel: str, class_name: str, endpoints: list[EndpointInfo]
+    ) -> ControllerInfo:
+        base_path = ""
+        m = _CLASS_MAPPING_RE.search(text)
+        if m:
+            base_path = m.group(1)
+
+        ctrl_endpoints: list[str] = []
+        for m in _METHOD_MAPPING_RE.finditer(text):
+            verb, sub = m.group(1).upper(), m.group(2) or ""
+            if verb == "REQUEST":
+                verb = "ANY"
+            full = f"{base_path.rstrip('/')}/{sub.lstrip('/')}".rstrip("/") or base_path
+            ctrl_endpoints.append(full)
+            endpoints.append(
+                EndpointInfo(
+                    method=verb,
+                    path=full,
+                    controller=class_name,
+                    description="",
+                    source_uri=rel,
+                )
+            )
+
+        return ControllerInfo(
+            name=class_name,
+            class_name=class_name,
+            path=rel,
+            base_path=base_path,
+            endpoints=ctrl_endpoints,
+            source_uri=rel,
+        )
+
+    @staticmethod
+    def _parse_service(text: str, rel: str, class_name: str) -> ServiceInfo:
+        methods = list(dict.fromkeys(_METHOD_DECL_RE.findall(text)))
+        return ServiceInfo(
+            name=class_name,
+            class_name=class_name,
+            path=rel,
+            methods=methods,
+            source_uri=rel,
+        )
+
+    @staticmethod
+    def _parse_entity(text: str, rel: str, class_name: str) -> EntityInfo:
+        table_name = None
+        m = _TABLE_RE.search(text)
+        if m:
+            table_name = m.group(1)
+        fields = list(dict.fromkeys(_FIELD_DECL_RE.findall(text)))
+        return EntityInfo(
+            name=class_name,
+            class_name=class_name,
+            path=rel,
+            table_name=table_name,
+            fields=fields,
+            source_uri=rel,
+        )
+
+    @staticmethod
+    def _modules(root: Path, java_files: list[Path]) -> list[str]:
+        """顶层目录中凡包含 Java 源码者视为一个模块（按名排序，保证确定性）。"""
+        mods = {
+            p.relative_to(root).parts[0]
+            for p in java_files
+            if len(p.relative_to(root).parts) > 1
+        }
+        return sorted(mods)

src/genesis/impact/existing_system_explorer.py

@@ -0,0 +1,20 @@
+"""ExistingSystemExplorer：CodeStructure → ExistingSystemInfo（Impact Agent MVP）。
+
+纯组装：将 CodeParser 的控制器/服务/实体/端点分层映射到 ExistingSystemInfo。
+"""
+from __future__ import annotations
+
+from genesis.data_models import CodeStructure, ExistingSystemInfo
+
+
+class ExistingSystemExplorer:
+    """将代码结构组装为既有系统信息（供 ImpactAgent 比对）。"""
+
+    def explore(self, code: CodeStructure) -> ExistingSystemInfo:
+        return ExistingSystemInfo(
+            controller_layer=code.controllers,
+            service_layer=code.services,
+            entity_layer=code.entities,
+            api_endpoints=code.endpoints,
+            source_path=code.root_path,
+        )

src/genesis/impact/impact_agent.py

@@ -0,0 +1,271 @@
+"""ImpactAgent：变更点定位（Impact Agent MVP，确定性规则，无 LLM）。
+
+从要件定義各表（機能/画面/DB/IF/バッチ）取 変更区分 + 既存対応 列，
+与 ExistingSystemInfo 连接比对，输出 ImpactReport（影响调查书）。
+"""
+from __future__ import annotations
+
+import logging
+from datetime import date
+
+from genesis.data_models import (
+    ChangeAnalysis,
+    ChangeElement,
+    ChangeType,
+    ExcelTable,
+    ImpactReport,
+    ImpactWarning,
+    SheetType,
+    StructuredSource,
+)
+
+_LOGGER = logging.getLogger(__name__)
+
+_SHEET_ELEMENT_TYPE = {
+    SheetType.FUNCTION: "機能",
+    SheetType.SCREEN: "画面",
+    SheetType.REPORT: "帳票",
+    SheetType.DATABASE: "DB",
+    SheetType.INTERFACE: "IF",
+    SheetType.BATCH: "バッチ",
+}
+
+_CHANGE_TYPE_MAP = {
+    "新規": ChangeType.NEW,
+    "変更": ChangeType.MODIFIED,
+    "削除": ChangeType.DELETED,
+    "不变": ChangeType.UNCHANGED,
+}
+
+
+def _element_to_dict(el: ChangeElement) -> dict:
+    """ChangeElement → JSON 可序列化 dict（ChangeType 枚举转 value）。"""
+    return {
+        "element_id": el.element_id,
+        "element_type": el.element_type,
+        "name": el.name,
+        "change_type": el.change_type.value,
+        "existing_mapping": list(el.existing_mapping),
+        "impacted_existing": list(el.impacted_existing),
+        "evidence": el.evidence,
+        "status": el.status,
+    }
+
+
+def impact_report_to_dict(report: ImpactReport) -> dict:
+    """影响调查书 → JSON 可序列化 dict（供下载/日志，api-design §2.4 impact-result）。"""
+    ca = report.change_analysis
+    return {
+        "metadata": dict(report.metadata),
+        "change_analysis": {
+            "project_type": ca.project_type,
+            "new_elements": [_element_to_dict(e) for e in ca.new_elements],
+            "modified_elements": [_element_to_dict(e) for e in ca.modified_elements],
+            "deleted_elements": [_element_to_dict(e) for e in ca.deleted_elements],
+            "unchanged_elements": [_element_to_dict(e) for e in ca.unchanged_elements],
+            "warnings": [{"element_id": w.element_id, "issue": w.issue} for w in ca.warnings],
+        },
+        "summary": dict(report.summary),
+    }
+
+
+def _header_index(headers: list[str], *keywords: str) -> int | None:
+    """按关键词定位列索引（如 変更区分 / 既存対応）。"""
+    for i, h in enumerate(headers):
+        hl = str(h).strip()
+        if any(k in hl for k in keywords):
+            return i
+    return None
+
+
+class ImpactAgent:
+    """变更点定位 → 影响调查书（MVP）。"""
+
+    def run(
+        self,
+        structured_source: StructuredSource,
+        session_id: str = "impact",
+        scope: dict | None = None,
+    ) -> ImpactReport:
+        existing = structured_source.existing_system
+        if existing is None:
+            raise ValueError("未提供既有系统（existing_system），无法执行影响调查")
+
+        lookup = self._build_lookup(existing)
+        new_elements: list[ChangeElement] = []
+        modified_elements: list[ChangeElement] = []
+        deleted_elements: list[ChangeElement] = []
+        warnings: list[ImpactWarning] = []
+        matched: set[str] = set()
+
+        for table in structured_source.tables:
+            self._classify_table(
+                table, lookup, new_elements, modified_elements, deleted_elements, warnings, matched
+            )
+
+        if scope:
+            # scope 参数预留：MVP 默认全量调查；模块/深度收窄由调用方确认后传入，当前忽略
+            _LOGGER.warning("scope 参数预留（MVP 默认全量调查），当前忽略: %s", scope)
+
+        unchanged = self._unchanged_count(existing, matched)
+        change_analysis = ChangeAnalysis(
+            project_type="enhancement",
+            new_elements=new_elements,
+            modified_elements=modified_elements,
+            deleted_elements=deleted_elements,
+            unchanged_elements=[],
+            warnings=warnings,
+        )
+        summary = {
+            "total": len(new_elements) + len(modified_elements) + len(deleted_elements),
+            "new": len(new_elements),
+            "modified": len(modified_elements),
+            "deleted": len(deleted_elements),
+            "unchanged": unchanged,
+            "warnings": len(warnings),
+        }
+        return ImpactReport(
+            metadata={
+                "version": "v1",
+                "session_id": session_id,
+                "created_at": date.today().isoformat(),
+                "llm_model": "none",  # MVP 确定性规则，无 LLM 参与
+                "source": existing.source_path,
+            },
+            change_analysis=change_analysis,
+            summary=summary,
+        )
+
+    # ---------- 内部 ----------
+
+    def _classify_table(
+        self,
+        table: ExcelTable,
+        lookup: dict[str, list[dict]],
+        new_elements: list[ChangeElement],
+        modified_elements: list[ChangeElement],
+        deleted_elements: list[ChangeElement],
+        warnings: list[ImpactWarning],
+        matched: set[str],
+    ) -> None:
+        headers = [str(h) for h in table.headers]
+        change_idx = _header_index(headers, "変更区分", "区分")
+        if change_idx is None:
+            return  # 无变更区分列的表（如新規開発的帳票一覧）不参与变更点定位
+
+        mapping_idx = _header_index(headers, "既存対応")
+        element_type = _SHEET_ELEMENT_TYPE.get(table.detected_type, table.name)
+        name_idx = _header_index(headers, "名") or 1
+
+        for row in table.rows:
+            change_val = self._cell(row, headers, change_idx)
+            if not change_val:
+                continue
+            change_type = _CHANGE_TYPE_MAP.get(str(change_val).strip())
+            if change_type is None:
+                continue
+
+            element_id = str(self._cell(row, headers, 0) or "")
+            name = str(self._cell(row, headers, name_idx) or "")
+            mapping = (
+                str(self._cell(row, headers, mapping_idx) or "")
+                if mapping_idx is not None
+                else ""
+            )
+            tokens = [
+                t.strip()
+                for t in mapping.replace("，", ",").replace(" ", "").split(",")
+                if t.strip()
+            ]
+            hits = self._match_tokens(tokens, lookup)
+
+            impacted = [h["class_name"] for h in hits]
+            evidence = hits[0]["source_uri"] if hits else ""
+            for h in hits:
+                matched.add(h["class_name"])
+
+            status = "ok"
+            issue: str | None = None
+            if change_type is ChangeType.NEW and tokens:
+                status = "conflict"
+                issue = f"新規要素却声明了既存対応: {mapping}"
+            elif change_type in (ChangeType.MODIFIED, ChangeType.DELETED):
+                if not tokens:
+                    status = "warning"
+                    issue = "缺少既存対応，无法定位修改/删除对象"
+                elif not hits:
+                    status = "warning"
+                    issue = f"既存対応无法匹配既有类: {mapping}"
+
+            element = ChangeElement(
+                element_id=element_id,
+                element_type=element_type,
+                name=name,
+                change_type=change_type,
+                existing_mapping=tokens,
+                impacted_existing=impacted,
+                evidence=evidence,
+                status=status,
+            )
+            if change_type is ChangeType.NEW:
+                new_elements.append(element)
+            elif change_type is ChangeType.MODIFIED:
+                modified_elements.append(element)
+            elif change_type is ChangeType.DELETED:
+                deleted_elements.append(element)
+
+            if issue:
+                warnings.append(ImpactWarning(element_id=element_id, issue=issue))
+
+    @staticmethod
+    def _build_lookup(existing) -> dict[str, list[dict]]:
+        """类名/表名（小写）→ 既有要素索引，供 token 匹配。"""
+        lookup: dict[str, list[dict]] = {}
+
+        def add(key: str, item: dict) -> None:
+            lookup.setdefault(key.lower(), []).append(item)
+
+        for c in existing.controller_layer:
+            add(c.class_name, {"kind": "controller", "class_name": c.class_name, "source_uri": c.source_uri})
+        for s in existing.service_layer:
+            add(s.class_name, {"kind": "service", "class_name": s.class_name, "source_uri": s.source_uri})
+        for e in existing.entity_layer:
+            add(e.class_name, {"kind": "entity", "class_name": e.class_name, "source_uri": e.source_uri})
+            if e.table_name:
+                add(e.table_name, {"kind": "entity", "class_name": e.class_name, "source_uri": e.source_uri})
+        return lookup
+
+    def _match_tokens(self, tokens: list[str], lookup: dict[str, list[dict]]) -> list[dict]:
+        """token → 既有类命中列表（去重）。匹配规则：类名/表名完全相等 或 类名前缀匹配。"""
+        hits: list[dict] = []
+        seen: set[tuple[str, str]] = set()
+        for token in tokens:
+            key = token.lower()
+            for candidate in lookup.get(key, []):
+                if (candidate["kind"], candidate["class_name"]) not in seen:
+                    seen.add((candidate["kind"], candidate["class_name"]))
+                    hits.append(candidate)
+            # 前缀匹配：token 是类名前缀（如 OrderService → OrderServiceImpl）
+            for k, items in lookup.items():
+                if k.startswith(key) and k != key:
+                    for candidate in items:
+                        if (candidate["kind"], candidate["class_name"]) not in seen:
+                            seen.add((candidate["kind"], candidate["class_name"]))
+                            hits.append(candidate)
+        return hits
+
+    @staticmethod
+    def _cell(row: dict, headers: list[str], idx: int):
+        if idx is None or idx >= len(headers):
+            return ""
+        cell = row.get(headers[idx])
+        return cell.value if cell is not None else ""
+
+    @staticmethod
+    def _unchanged_count(existing, matched: set[str]) -> int:
+        all_classes = (
+            {c.class_name for c in existing.controller_layer}
+            | {s.class_name for s in existing.service_layer}
+            | {e.class_name for e in existing.entity_layer}
+        )
+        return len(all_classes - matched)

src/genesis/parsers/source_aggregator.py

@@ -3,6 +3,8 @@ from __future__ import annotations
 from pathlib import Path
 
 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.rule_doc_parser import RuleDocParser
 from genesis.parsers.word_template_parser import WordTemplateParser
@@ -42,8 +44,13 @@ class SourceParser:
         template_path: str | Path | None = None,
         write_instruction_paths: list[str | Path] | None = None,
         rule_paths: list[str | Path] | None = None,
+        existing_system_path: str | Path | None = None,
     ) -> StructuredSource:
-        """扩展名校验先于存在性校验（不存在的文件若扩展名未知将抛出 ValueError 而非 FileNotFoundError）。"""
+        """扩展名校验先于存在性校验（不存在的文件若扩展名未知将抛出 ValueError 而非 FileNotFoundError）。
+
+        existing_system_path：既有系统源码目录（追加/改修场景）。提供时解析为
+        ExistingSystemInfo（门控通过 → 进入影响调查）；未提供/解析失败 → 保持 None。
+        """
 
         requirement_paths = requirement_paths or []
         write_instruction_paths = write_instruction_paths or []
@@ -68,11 +75,16 @@ class SourceParser:
             # 做成说明书与记入规则均为 Type A 写入规则 → write（api-design §2.2）
             rule_docs.append(RuleDocParser().parse(path, category="write"))
 
+        existing_system = None
+        if existing_system_path is not None:
+            code = CodeParser().parse(existing_system_path)
+            existing_system = ExistingSystemExplorer().explore(code)
+
         return StructuredSource(
             tables=tables,
             template=template,
             rule_docs=rule_docs,
             image_analyses=[],
-            existing_system=None,
+            existing_system=existing_system,
             comments=comments,
         )

src/genesis/writer/context_builder.py

@@ -26,6 +26,7 @@ def build_contexts(structured_source: StructuredSource, samples_dir: str = "samp
                 design_rules=design_rules,
                 template_styles=set(used),
                 prior_state=None,
+                impact_report=getattr(structured_source, "impact_report", None),
             )
         )
     return out

src/genesis/writer/models.py

@@ -76,6 +76,7 @@ class GenerationContext:
     design_rules: list[str]
     template_styles: set[str]
     prior_state: object | None = None  # WriterState，避免循环 import 用 object
+    impact_report: object | None = None  # ImpactReport 影响调查书（生成主上下文）
 
     def to_vars(self) -> dict:
         """返回供 prompt 渲染的变量字典。"""
@@ -90,4 +91,35 @@ class GenerationContext:
             "template_styles": ", ".join(sorted(self.template_styles)),
             "prior_state": str(self.prior_state) if self.prior_state is not None else "",
             "source": str(self.structured_source) if self.structured_source is not None else "",
+            "impact": _format_impact(self.impact_report),
         }
+
+
+def _format_impact(report: object | None) -> str:
+    """将影响调查书格式化为 prompt 可读文本（无报告/无分析时为空串）。"""
+    if report is None:
+        return ""
+    ca = getattr(report, "change_analysis", None)
+    if ca is None:
+        return ""
+    summary = getattr(report, "summary", {}) or {}
+    lines = [f"project_type={getattr(ca, 'project_type', '')}"]
+    lines.append(
+        "summary: new={new} modified={modified} deleted={deleted} "
+        "unchanged={unchanged} warnings={warnings}".format(
+            new=summary.get("new", 0), modified=summary.get("modified", 0),
+            deleted=summary.get("deleted", 0), unchanged=summary.get("unchanged", 0),
+            warnings=summary.get("warnings", 0),
+        )
+    )
+    for el in getattr(ca, "new_elements", []) or []:
+        lines.append(f"[新規] {el.element_id} {el.element_type} {el.name}")
+    for el in getattr(ca, "modified_elements", []) or []:
+        impacted = ", ".join(el.impacted_existing) or "-"
+        lines.append(f"[変更] {el.element_id} {el.element_type} {el.name} → 受影响: {impacted}")
+    for el in getattr(ca, "deleted_elements", []) or []:
+        impacted = ", ".join(el.impacted_existing) or "-"
+        lines.append(f"[削除] {el.element_id} {el.element_type} {el.name} → 受影响: {impacted}")
+    for w in getattr(ca, "warnings", []) or []:
+        lines.append(f"[警告] {w.element_id}: {w.issue}")
+    return "\n".join(lines)

src/genesis/writer/orchestrator.py

@@ -1,10 +1,17 @@
-"""Writer 编排：上下文装配 → 逐章生成 → 渲染 → docx 注入（Phase 5 垂直切片）。"""
+"""Writer 编排：上下文装配 → 逐章生成 → 渲染 → docx 注入（Phase 5 垂直切片）。
+
+Impact Agent MVP（2026-08-23）：门控 = 用户是否提供既有系统（existing_system 非 None）。
+门控通过且未显式传入 impact_report 时，自动运行 ImpactAgent 生成影响调查书，
+并作为生成主上下文（GenerationContext.impact_report → prompt 的 {{impact}} 变量）。
+"""
 from __future__ import annotations
 
 import logging
+from datetime import date
 from pathlib import Path
 
 from genesis.data_models import StructuredSource
+from genesis.impact.impact_agent import ImpactAgent
 from genesis.inference.factory import build_inference_engine
 from genesis.inference.prompt_registry import PromptRegistry
 from genesis.writer.context_builder import build_contexts
@@ -47,9 +54,18 @@ class WriteOrchestrator:
         engine=None,
         prompt_registry=None,
         template_path: str | None = None,
+        impact_report=None,
+        meta: dict | None = None,
     ) -> list[ChapterContent]:
         engine = engine or build_inference_engine()
         prompt_registry = prompt_registry or PromptRegistry()
+        if impact_report is None and getattr(structured_source, "existing_system", None) is not None:
+            # 门控：用户提供了既有系统（existing_system 非 None）→ 自动执行影响调查
+            _LOGGER.info("检测到既有系统，自动执行影响调查（追加改修场景）")
+            impact_report = ImpactAgent().run(structured_source, session_id=session_id)
+        if impact_report is not None:
+            # 回填 structured_source，便于 QA/日志/后续下载
+            structured_source.impact_report = impact_report
         ctxs = build_contexts(structured_source, samples_dir)
         _warn_unanchored(ctxs)
         state = WriterState([c.chapter_id for c in ctxs])
@@ -68,7 +84,13 @@ class WriteOrchestrator:
         tpl = template_path or getattr(structured_source.template, "file_name", None)
         if not tpl:
             raise ValueError("template_path 必须提供（structured_source.template.file_name 为空）")
-        doc = DocxInjector(tpl).inject(sections, meta={})
+        if meta is None:
+            meta = {
+                "doc_title": Path(tpl).stem,
+                "version": "v1",
+                "created_at": date.today().isoformat(),
+            }
+        doc = DocxInjector(tpl).inject(sections, meta=meta)
         Path(output_path).parent.mkdir(parents=True, exist_ok=True)
         doc.save(output_path)
         return contents

src/genesis/writer/writer_agent.py

@@ -21,6 +21,7 @@ WRITER_PROMPT_TEMPLATE = (
     "写入规则:\n{{write_rules}}\n"
     "设计规则:\n{{design_rules}}\n"
     "模板样式:\n{{template_styles}}\n"
+    "影响调查上下文:\n{{impact}}\n"
     "参考资料:\n{{source}}\n"
     "请输出符合 schema 的章节内容 JSON。\n"
     "【语言约束】章节正文（所有 block 的 text 字段）所使用的自然语言，"