#!/usr/bin/env python3 """SDD MVCP 生成器（模板版本）用途： - 从 PRD 目录快速生成可复用的 SDD 骼架 - 输出 01~06 章节骨架 + sdd-mvcp-contract.json - 自动根据工作量规模生成执行图（sdd-graph.mmd / sdd-graph.runbook.json） - 仅用于“先有可追溯框架，再逐项填充真实规则” 默认输入： - PRD 目录（含功能清单/流程/页面说明）输出： - 00-SDD-生成记录.md - 01-M-模型设计.md - 02-V-界面与交互.md - 03-C-控制与接口.md - 04-P-流程与状态流转.md - 05-实施约束与依赖.md - 06-验收与测试锚点.md - sdd-mvcp-contract.json - sdd_exec_graph_manifest.json（若生成图开启） - sdd-graph.mmd（若生成图开启） - sdd-graph.runbook.json（若生成图开启） """ from __future__ import annotations import argparse import hashlib import json import re import subprocess import sys from collections import OrderedDict from datetime import datetime from pathlib import Path from typing import Any, Dict, Iterable, List, Optional REQ_ID_RE = re.compile(r"(?P(REQ|PRD|PBI|TASK|ISS)[-_]?\d+(?:\.\d+)?|[A-Za-z]+-REQ-\d+)", re.I) DEFAULT_AGENT_POOL = ["agent-1", "agent-2", "agent-3", "agent-4", "agent-5"] # PRD 文本语义规则（默认） # 仅用于 manifest 推导；显式传入 --manifest 时不生效。 PRD_PARALLEL_BAN_KEYWORDS = [ "支付", "财务", "对账", "扣费", "结算", "退款", "税", "风控", "权限", "审计", "合规", "资金", "库存", "并发", ] PRD_FLOW_BOUNDARY_KEYWORDS = [ "审批", "审核", "异步", "回调", "超时", "重试", "补偿", "状态机", "队列", "任务", "通知", "消息", "人工干预", ] PRD_SCOPE_HINT_KEYWORDS = [ "页面", "视图", "表单", "界面", "功能", "场景", "接口", "流程", ] def parse_requirements_from_lines(lines: List[str]) -> List[str]: req_ids = [] for line in lines: m = REQ_ID_RE.search(line) if m: rid = m.group("id") if rid not in req_ids: req_ids.append(rid) return req_ids def infer_requirements(prd_root: Path) -> List[str]: # 从 PRD 文件中抽取需求 ID；无 ID 时按占位需求编号。 ids: List[str] = [] for p in _iter_prd_text_files(prd_root): try: lines = p.read_text(encoding="utf-8", errors="ignore").splitlines() except OSError: continue found = parse_requirements_from_lines(lines) for rid in found: if rid not in ids: ids.append(rid) if ids: return ids # 无显式编号时，按功能章节生成占位需求 ID return [f"REQ-{i:02d}" for i in range(1, 6)] def _iter_prd_text_files(prd_root: Path) -> list[Path]: """按可预期后缀读取 PRD 文本文件（递归）。""" suffix_set = {".md", ".markdown", ".txt"} if not prd_root.exists() or not prd_root.is_dir(): return [] return sorted([p for p in prd_root.rglob("*") if p.is_file() and p.suffix.lower() in suffix_set]) def _file_sha256(path: Path) -> str: h = hashlib.sha256() with path.open("rb") as fp: for chunk in iter(lambda: fp.read(8192), b""): h.update(chunk) return h.hexdigest() def _path_sha256(paths: Iterable[Path], base: Path) -> str: h = hashlib.sha256() for p in sorted(paths): rel = str(p.resolve().relative_to(base.resolve())) if p.is_absolute() else str(p) h.update((rel + "\n").encode("utf-8")) try: h.update((p.read_bytes()[:16384] + b"\n") ) except Exception: continue return h.hexdigest()[:16] def _git_head(base: Path) -> str: try: proc = subprocess.run( ["git", "-C", str(base), "rev-parse", "HEAD"], check=False, capture_output=True, text=True, timeout=10, ) if proc.returncode == 0: return proc.stdout.strip() except Exception: pass return "" def _read_prd_text(prd_root: Path, max_files: int = 12, max_chars: int = 24000) -> str: """读取 PRD 文本用于默认 manifest 规则推导（非权威规则）。""" chunks: List[str] = [] for idx, p in enumerate(_iter_prd_text_files(prd_root)): if idx >= max_files: break try: text = p.read_text(encoding="utf-8", errors="ignore") except OSError: continue chunks.append(text) if sum(len(x) for x in chunks) >= max_chars: break return "\n".join(chunks)[:max_chars] def infer_manifest_rules(prd_root: Path, requirements_count: int, threshold: int) -> dict: """基于 PRD 文本语义推导 manifest 默认规则。返回值为可解释字段，供默认 manifest meta 和节点可并行策略引用。 """ text = _read_prd_text(prd_root) if not text: return { "matched_keywords": [], "heading_level2_count": 0, "heading_level3_count": 0, "parallel_blockers": [], "flow_boundary_markers": [], "scope_markers": [], "enable_mv_parallel": requirements_count >= threshold, "p_depends_on_view": False, "c_depends_on_process": False, "strict_contract_mode": False, "reason": "PRD 文本为空，仅按规模规则回退", } text_lower = text.lower() heading_level2_count = len(re.findall(r"^##\s+", text, flags=re.M)) heading_level3_count = len(re.findall(r"^###\s+", text, flags=re.M)) parallel_blockers = [kw for kw in PRD_PARALLEL_BAN_KEYWORDS if kw in text_lower] flow_markers = [kw for kw in PRD_FLOW_BOUNDARY_KEYWORDS if kw in text] scope_markers = [kw for kw in PRD_SCOPE_HINT_KEYWORDS if kw in text] strict_contract_mode = bool(parallel_blockers) flow_boundary_mode = bool(flow_markers) or heading_level3_count >= 8 p_depends_on_view = flow_boundary_mode and "页面" in text c_depends_on_process = flow_boundary_mode and ("审批" in text or "异步" in text or "补偿" in text) if strict_contract_mode: # 风险域：默认更保守，减少误并行，避免错误的跨阶段重叠。 enable_mv_parallel = False else: enable_mv_parallel = requirements_count >= threshold reason = ( f"parallel={'开' if enable_mv_parallel else '关'}（requirements={requirements_count} " f"threshold={threshold}；blockers={len(parallel_blockers)}）" ) return { "matched_keywords": sorted(set(parallel_blockers + flow_markers + scope_markers)), "heading_level2_count": heading_level2_count, "heading_level3_count": heading_level3_count, "parallel_blockers": sorted(set(parallel_blockers)), "flow_boundary_markers": sorted(set(flow_markers)), "scope_markers": sorted(set(scope_markers)), "enable_mv_parallel": enable_mv_parallel, "p_depends_on_view": p_depends_on_view, "c_depends_on_process": c_depends_on_process, "strict_contract_mode": strict_contract_mode, "reason": reason, } def render_head(*parts: str) -> str: return "\n\n".join(p.strip("\n") for p in parts if p) def write_markdown(path: Path, title: str, body: str) -> None: path.write_text(f"# {title}\n\n{body}\n", encoding="utf-8") def write_contract(path: Path, contract: dict) -> None: path.write_text(json.dumps(contract, ensure_ascii=False, indent=2), encoding="utf-8") def write_text_file(path: Path, content: str) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(content, encoding="utf-8") def _make_contract_stubs(args: argparse.Namespace, requirements: List[str]) -> Dict[str, Any]: model_stub = [ { "entity": "placeholder_entity", "owner": args.owner, "purpose": "待替换：请在 PRD 对齐后补齐真实实体定义", "fields": [ { "name": "id", "type": "bigint", "nullable": False, "constraints": ["pk"], "index": "idx", "default": None, } ], "indexes": ["id"], "unique": ["id"], "state_machine": ["DRAFT", "ACTIVE", "DELETED"], "rules": ["请补齐关键约束"], "migration_notes": "请补齐上线/回滚约束", } ] view_stub = [ { "view_id": "v_placeholder", "route": "/todo", "role_permissions": ["owner:all"], "fields": [ { "name": "todo_field", "visible_when": "all", "validation": ["required"], "required": True, } ], "events": [ { "event": "submit", "target_controller_id": "c_placeholder", "payload": {"todo_field": "string"}, } ], } ] controller_stub = [ { "operation_id": "c_placeholder", "method": "POST", "path": "/api/placeholder", "auth": "rbac:placeholder", "input": {"todo_field": "string"}, "output": {"success": True, "id": "int"}, "idempotent": False, "transaction": "required", "errors": [ { "code": "INVALID_PARAM", "retryable": False, "description": "参数校验失败", "side_effects": "none", } ], "audit_trace": { "trace_id": "trace_id", "actor": "actor_id", "business_id": "business_id", }, } ] process_stub = [ { "flow_id": "p_placeholder", "start_state": "DRAFT", "end_state": "DONE", "transitions": [ { "from": "DRAFT", "to": "DONE", "trigger": "submit", "action": "c_placeholder", "guard": "todo_guard", "error_state": "FAILED", "retry_max": 1, } ], "timeout_ms": 3000, "concurrency": "serial", "compensation": "请补齐回滚动作", } ] acceptance = [] for rid in requirements: acceptance.append( { "requirement_id": rid, "coverage": { "model": ["placeholder_entity"], "view": ["v_placeholder"], "controller": ["c_placeholder"], "process": ["p_placeholder"], }, "acceptance_cases": [ {"type": "happy", "description": "待补充", "expected": "待补充"}, {"type": "boundary", "description": "待补充", "expected": "待补充"}, {"type": "failure", "description": "待补充", "expected": "待补充"}, ], } ) return { "model_contract": model_stub, "view_contract": view_stub, "controller_contract": controller_stub, "process_contract": process_stub, "acceptance": acceptance, } def build_contract(args: argparse.Namespace, requirements: List[str], prd_root: Path, manifest: Optional[dict] = None, manifest_source: str = "") -> Dict[str, Any]: now = datetime.utcnow().replace(microsecond=0).isoformat() + "Z" stubs = _make_contract_stubs(args, requirements) source_files = _iter_prd_text_files(prd_root) contract = OrderedDict( [ ( "meta", { "project": args.project_name, "version": args.version, "effective_date": now, "owner": args.owner, "source_prd_root": str(prd_root), "requirements_count": len(requirements), "generator": "sdd_mvcp_generator.py", "generator_version": "1.2.0+seeded", "requirements_source": "requirements-json" if args.requirements_json else "prd_scan", "source_fingerprint": _path_sha256(source_files, prd_root), "source_file_count": len(source_files), "git_head": _git_head(prd_root), "build_manifest_source": manifest_source, "manifest_derived_rules": manifest.get("meta", {}).get("derived_rules", {}) if manifest else {}, }, ), ("assumptions", [ { "id": "A-1", "statement": "待补充", "owner": args.owner, "verification_plan": "待补充", "status": "pending", } ]), ("facts", []), ("model_contract", stubs["model_contract"]), ("view_contract", stubs["view_contract"]), ("controller_contract", stubs["controller_contract"]), ("process_contract", stubs["process_contract"]), ("non_goals", ["待补充：请在 PRD 中显式列出不做事项"]), ("risks", [{"id": "R-1", "description": "待补充", "impact": "medium", "mitigation": "待补充", "status": "pending"}]), ("acceptance_matrix", stubs["acceptance"]), ] ) return contract def build_m_section() -> str: return render_head( "## 1. 实体清单\n- entity: <实体名>\n- purpose: <为什么要建这个表>\n- owner: <模块 owner>", "### 字段（每行要有：name/type/nullable/constraints/index/default）\n- name: <字段>\n - type: <类型>\n - nullable: \n - constraints: [<约束>]\n - index: \n - default: <默认值>", "### 生命周期\n- state_machine: [, , ]\n- migration_notes: <上线/回滚约束>", ) def build_v_section() -> str: return render_head( "## 1. 页面清单\n- view_id: v_...\n- route: /...\n- role_permissions: [<角色-可见字段映射>]", "### 字段与校验\n- name: <字段>\n - type: <输入类型>\n - required: \n - validation: [<规则>]\n - error_message: <提示文案>", "### 交互事件映射\n- event: <点击/提交/...>\n - target_controller_id: \n - payload: ", ) def build_c_section() -> str: return render_head( "## 1. 接口清单（按资源分组）\n- operation_id: c_...\n- method: GET|POST|PUT|PATCH|DELETE\n- path: /api/...\n- auth: <鉴权规则>", "### 入参与出参\n- input_contract: \n- output_contract: <成功+失败字段定义>", "### 错误码\n- code: \n - retryable: \n - description: <错误说明>\n - side_effects: <副作用>", ) def build_p_section() -> str: return render_head( "## 1. 流程清单\n- flow_id: p_...\n- start_state: \n- end_state: ", "### 转移定义\n- from: \n - to: \n - trigger: <事件>\n - action: \n - guard: <前置条件>\n - error_state: \n - retry_max: <次数>", "### 并发与补偿\n- concurrency: \n- timeout_ms: <整数>\n- compensation: <回滚动作>", ) def build_impl_section() -> str: return render_head( "## 1. 依赖图\n- upstream: [<系统/服务>]\n- downstream: [<调用方>]", "## 2. 上下文边界\n- in_scope: [<本次实现范围>]\n- out_scope: []", "## 3. 迁移与回滚\n- migration_order: [<顺序>]\n- rollback_plan: <回滚策略>", "## 4. 并发与幂等\n- conflict_hotspots: [<热点项>]\n- idempotent_plan: <唯一键/锁/乐观锁>", ) def build_test_section() -> str: return render_head( "## 1. 覆盖矩阵\n- requirement_id: PRD-REQ-01 -> model/view/controller/process", "## 2. 观察点\n- 接口合约测试：<接口>\n- 数据约束测试：<字段/索引>\n- 流程测试：<状态机>\n- 视图端到端：<场景>", "## 3. 门禁项（未填不得发起 coding）\n- [ ] 每条 C 对应 M/V\n- [ ] 每条 P 有 timeout + error_state\n- [ ] non-goals 已确认", ) def build_workflow_manifest(args: argparse.Namespace, requirements: List[str], prd_root: Path) -> dict: # 先做 PRD 文本语义规则推导，再结合规模阈值决定并行边界 parallel_threshold = max(1, args.graph_parallel_threshold) rules = infer_manifest_rules(prd_root, len(requirements), parallel_threshold) m_v_parallel = bool(rules["enable_mv_parallel"]) p_depends_on_view = bool(rules["p_depends_on_view"]) c_depends_on_process = bool(rules["c_depends_on_process"]) c_depends_on = ["p0_input", "m_model", "v_view"] if c_depends_on_process: c_depends_on.append("p_process") p_depends_on = ["p0_input", "m_model"] if p_depends_on_view: p_depends_on.append("v_view") nodes = [ { "id": "p0_input", "name": "P0 输入与边界", "owner": "agent-1", "depends_on": [], "outputs": ["facts", "constraints", "non_goals"], "parallelizable": False, "assumption": "PRD/BRD 与非目标先清晰归一", }, { "id": "m_model", "name": "M 模型", "owner": "agent-2", "depends_on": ["p0_input"], "outputs": ["01-M-模型设计.md", "model_contract"], "parallelizable": m_v_parallel, "assumption": "M 与 V 在边界清晰时并行", }, { "id": "v_view", "name": "V 视图", "owner": "agent-3", "depends_on": ["p0_input"], "outputs": ["02-V-界面与交互.md", "view_contract"], "parallelizable": m_v_parallel, "assumption": "页面与状态映射需基于需求边界", }, { "id": "c_controller", "name": "C 控制", "owner": "agent-2", "depends_on": c_depends_on, "outputs": ["03-C-控制与接口.md", "controller_contract"], "parallelizable": False, "assumption": "M/V 确认后汇总到接口，若出现审批/异步/回调则待流程澄清后再进入", }, { "id": "p_process", "name": "P 流程", "owner": "agent-4", "depends_on": p_depends_on, "outputs": ["04-P-流程与状态流转.md", "process_contract"], "parallelizable": True, "assumption": "在 M 稳定后推进；若文本含强边界描述则补充等待 V 反馈", }, { "id": "s_synthesis", "name": "S 约束聚合", "owner": "agent-5", "depends_on": ["m_model", "v_view", "c_controller", "p_process"], "outputs": ["05-实施约束与依赖.md", "sdd-mvcp-contract.json", "06-验收与测试锚点.md"], "parallelizable": False, "assumption": "等待关键链路完成后聚合", }, { "id": "q_gate", "name": "Q 质量门禁", "owner": "agent-1", "depends_on": ["s_synthesis"], "outputs": ["gate_report.md"], "parallelizable": False, "assumption": "反模式与字段约束必须通过", }, { "id": "u_units", "name": "U Unit 拆分与发布", "owner": "agent-1", "depends_on": ["q_gate"], "outputs": ["unit_matrix.json", "handoff_bundle.md"], "parallelizable": False, "assumption": "仅在约束包完成且通过门禁后拆分", }, ] pool = [x.strip() for x in args.agent_pool.split(",") if x.strip()] if not pool: pool = [f"agent-{i}" for i in range(1, 6)] # 若要求的 agent pool 有自定义则映射 if args.agent_pool: for idx, node in enumerate(nodes): if node["owner"] == f"agent-{min(idx + 1, len(pool))}": node["owner"] = pool[idx % len(pool)] return { "meta": { "project": args.project_name, "version": args.version, "out_dir": str(Path(args.out).resolve()), "requirements_count": len(requirements), "prd_root": str(prd_root.resolve()), "parallel_mode": "auto" if m_v_parallel else "serial_first", "derived_rules": rules, "generated_at": datetime.utcnow().replace(microsecond=0).isoformat() + "Z", }, "nodes": nodes, "loop_rules": [ { "from": "q_gate", "on_failure": { "m_model": ["缺字段", "模型约束冲突"], "v_view": ["事件未映射", "权限遗漏"], "c_controller": ["错误码缺失", "入参输出不齐"], "p_process": ["错误态缺失", "无补偿策略"], }, } ], "agent_pool": pool, "constraints": { "no_merge_without_contract": True, "gate_required": True, "max_loop_round": args.graph_loop_round, "strict_contract_mode": rules["strict_contract_mode"], }, } def _validate_contract(contract: dict, schema_path: Path) -> None: errors: List[str] = [] required_top = ["meta", "model_contract", "view_contract", "controller_contract", "process_contract", "acceptance_matrix"] for key in required_top: if key not in contract: errors.append(f"缺少必填字段: {key}") if "meta" in contract: for field in ["project", "version", "effective_date", "owner"]: if not contract["meta"].get(field): errors.append(f"meta 缺少 {field}") for key in ["model_contract", "view_contract", "controller_contract", "process_contract", "acceptance_matrix"]: if key in contract and (not isinstance(contract[key], list) or len(contract[key]) == 0): errors.append(f"{key} 不能为空") acceptance = contract.get("acceptance_matrix", []) for acc in acceptance: for k in ["requirement_id", "coverage", "acceptance_cases"]: if k not in acc: errors.append(f"acceptance 项缺少 {k}") if isinstance(acc.get("coverage"), dict): for ck in ["model", "view", "controller", "process"]: if ck not in acc["coverage"]: errors.append(f"acceptance {acc.get('requirement_id')} coverage 缺少 {ck}") if not acc.get("acceptance_cases"): errors.append(f"acceptance {acc.get('requirement_id')} 缺少 acceptance_cases") if errors: raise SystemExit("contract 结构校验失败:\n" + "\n".join(f"- {e}" for e in errors)) if not schema_path.exists(): print(f"warn: schema 文件缺失，跳过 jsonschema 校验: {schema_path}") return try: import jsonschema # type: ignore except Exception: # 无强依赖，保底做本地结构校验 return schema = json.loads(schema_path.read_text(encoding="utf-8")) try: jsonschema.validate(contract, schema) except Exception as exc: raise SystemExit(f"contract jsonschema 校验失败: {exc}") def emit_graph_artifacts( out_dir: Path, manifest: dict, manifest_path: Path, script_path: Path, sdd_dir: Path, ) -> dict: manifest_path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8") cmd = [ sys.executable, str(script_path), "--manifest", str(manifest_path), "--sdd-dir", str(sdd_dir), "--out-dir", str(out_dir), ] proc = subprocess.run(cmd, capture_output=True, text=True) if proc.returncode != 0: raise SystemExit(f"执行图生成失败: {proc.stderr.strip()}") if proc.stdout: print(proc.stdout.strip()) runbook_path = out_dir / "sdd-graph.runbook.json" if not runbook_path.exists(): raise SystemExit(f"执行图失败：未生成 runbook {runbook_path}") runbook = json.loads(runbook_path.read_text(encoding="utf-8")) return runbook def build_unit_matrix(contract: dict, manifest: dict, source_manifest: str = "") -> Dict[str, Any]: units = [] acceptance_matrix = contract.get("acceptance_matrix", []) for idx, item in enumerate(acceptance_matrix, start=1): rid = item.get("requirement_id", f"REQ-{idx:02d}") units.append( { "unit_id": f"U-{rid}", "name": f"需求 {rid} 落地", "depends_on": ["q_gate"], "scope": item.get("coverage", {}), "coverage_requirements": item.get("coverage", {}), "acceptance_cases": item.get("acceptance_cases", []), "contract_file": "sdd-mvcp-contract.json", } ) if not units: units.append( { "unit_id": "U-BASE", "name": "SDD 依赖与接口契约", "depends_on": ["q_gate"], "scope": {}, "acceptance_cases": [], "contract_file": "sdd-mvcp-contract.json", } ) return { "meta": { "project": manifest.get("meta", {}).get("project", ""), "version": manifest.get("meta", {}).get("version", ""), "created_at": manifest.get("meta", {}).get("generated_at", datetime.utcnow().replace(microsecond=0).isoformat() + "Z"), "source_manifest": source_manifest, }, "units": units, } def write_handoff_bundle(path: Path, runbook: dict, unit_matrix: dict) -> None: lines = ["# SDD 分工执行包\n", f"- generated_at: {datetime.utcnow().replace(microsecond=0).isoformat()}Z\n"] lines.append(f"- out_dir: {runbook.get('sdd_dir', '')}\n") lines.append("- waves:\n") for wave in runbook.get("waves", []): lines.append(f" - {wave.get('wave')}: {wave.get('nodes')}\n") lines.append("\n- units:\n") for u in unit_matrix.get("units", []): lines.append(f" - {u['unit_id']}: {u['name']}\n") path.write_text("".join(lines), encoding="utf-8") def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Generate SDD MVCP skeleton files") parser.add_argument("--project-name", required=True) parser.add_argument("--version", required=True) parser.add_argument("--owner", required=True) parser.add_argument("--prd-root", default=".", help="PRD 源目录，默认当前目录") parser.add_argument("--out", default="./SDD", help="SDD 输出目录") parser.add_argument("--requirements-json", default="", help="可选：包含 req_id 数组的 JSON 文件") parser.add_argument("--disable-graph", action="store_true", help="关闭执行图自动生成") parser.add_argument("--manifest", default="", help="可选：指定自定义 workflow manifest") parser.add_argument("--agent-pool", default=",".join(DEFAULT_AGENT_POOL), help="可选：可用 agent 列表，逗号分隔") parser.add_argument("--graph-parallel-threshold", type=int, default=3, help="需求数>=阈值时允许 M/V 并行") parser.add_argument("--graph-loop-round", type=int, default=2, help="执行图 loop 规则允许回灌轮次") return parser.parse_args() def build_requirements(prd_root: Path, requirements_json: str) -> List[str]: requirements: List[str] = [] if requirements_json: p = Path(requirements_json) try: data = json.loads(p.read_text(encoding="utf-8")) if isinstance(data, list): requirements = [str(x) for x in data] elif isinstance(data, dict): requirements = [str(x) for x in data.get("requirements", [])] else: raise ValueError("requirements-json 须为数组或 {requirements: [...]} 对象") except Exception as exc: raise SystemExit(f"requirements-json 无法解析: {p} ({exc})") if requirements: return list(dict.fromkeys(requirements)) if not prd_root.is_dir(): raise SystemExit(f"PRD 根目录不存在: {prd_root}") return infer_requirements(prd_root) def main() -> int: args = parse_args() prd_root = Path(args.prd_root).expanduser().resolve() out_dir = Path(args.out).expanduser().resolve() out_dir.mkdir(parents=True, exist_ok=True) requirements = build_requirements(prd_root, args.requirements_json) write_markdown(out_dir / "01-M-模型设计.md", "MVCP-M 模型设计", build_m_section()) write_markdown(out_dir / "02-V-界面与交互.md", "MVCP-V 视图", build_v_section()) write_markdown(out_dir / "03-C-控制与接口.md", "MVCP-C 控制", build_c_section()) write_markdown(out_dir / "04-P-流程与状态流转.md", "MVCP-P 流程", build_p_section()) write_markdown(out_dir / "05-实施约束与依赖.md", "实施约束与依赖", build_impl_section()) write_markdown(out_dir / "06-验收与测试锚点.md", "验收与测试锚点", build_test_section()) contract = None manifest = None manifest_source = str(out_dir / "sdd_exec_graph_manifest.json") if args.disable_graph: manifest = None else: manifest_path = out_dir / "sdd_exec_graph_manifest.json" manifest_source = str(manifest_path) if args.manifest: user_manifest = Path(args.manifest).expanduser() if user_manifest.exists(): manifest_source = str(user_manifest) manifest = json.loads(user_manifest.read_text(encoding="utf-8")) else: raise SystemExit(f"指定的 manifest 不存在: {user_manifest}") else: manifest = build_workflow_manifest(args, requirements, prd_root) contract = build_contract(args, requirements, prd_root, manifest, manifest_source) schema_path = Path(__file__).resolve().parent.parent / "references" / "sdd_mvcp_contract_schema.json" _validate_contract(contract, schema_path) write_contract(out_dir / "sdd-mvcp-contract.json", contract) if not args.disable_graph: graph_script = Path(__file__).resolve().parent / "sdd_exec_graph.py" runbook = emit_graph_artifacts( out_dir=out_dir, manifest=manifest, manifest_path=manifest_path, script_path=graph_script, sdd_dir=out_dir, ) unit_matrix = build_unit_matrix(contract, manifest, str(manifest_path)) write_contract(out_dir / "unit_matrix.json", unit_matrix) write_handoff_bundle( out_dir / "handoff_bundle.md", runbook, unit_matrix, ) # 生成器记录（覆盖写入，保证审计可追踪） graph_record_lines = [ "# SDD 生成记录\n", f"- generated_at: {contract['meta']['effective_date']}\n", f"- project: {args.project_name}\n", f"- source_prd_root: {prd_root}\n", f"- source_prd_root_git_head: {contract['meta'].get('git_head', '')}\n", f"- source_file_fingerprint: {contract['meta'].get('source_fingerprint', '')}\n", f"- requirements_count: {len(requirements)}\n", f"- graph_enabled: {not args.disable_graph}\n", f"- generator: {contract['meta'].get('generator', '')} {contract['meta'].get('generator_version', '')}\n", f"- command: {' '.join(sys.argv)}\n", f"- invocation: interactive\n", ] if not args.disable_graph: graph_record_lines.append("- graph_inference: 基于 PRD 文本语义 + 规模阈值（显式 manifest 时不参与）\n") if manifest and isinstance(manifest, dict): manifest_constraints = manifest.get("constraints", {}) if not isinstance(manifest_constraints, dict): manifest_constraints = {} graph_record_lines.append(f"- manifest_meta_keys: {sorted(manifest.get('meta', {}).keys())}\n") graph_record_lines.append(f"- manifest_constraint_keys: {sorted(manifest_constraints.keys())}\n") write_text_file( out_dir / "00-SDD-生成记录.md", "".join(graph_record_lines), ) print( "sdd_skeleton_written=ok " f"out_dir={out_dir} requirements={len(requirements)} graph_enabled={not args.disable_graph}" ) return 0 if __name__ == "__main__": raise SystemExit(main())