feat(flowguard): 阶段状态迁移矩阵 + 项目级文档校验 + 模块栈覆盖对照

README.md

@@ -42,10 +42,14 @@ The ten stages are a verifiable process skeleton, not a Hook-driven script. Read
 ```bash
 /flowguard-discover    # read-only; never installs or initializes tools
 /flowguard-context     # classify and bind this session/worktree/task
+/flowguard-init        # create project-level stage docs 02/07/10
+/flowguard-stage       # read and advance docs/ stage documents
 /flowguard-evidence    # record real verification evidence
 /flowguard-governance  # check code-write, commit, or release readiness
 ```
 
+The CLI additionally provides `validate` (artifact content, traceability matrix and module-stack coverage checks).
+
 Kimi registers the same command prompts as namespaced Markdown commands, for example `/flowguard:flowguard-discover`. The Markdown files in `kimi-commands/` are generated from `commands/*.json`; regenerate with `python3 scripts/generate_kimi_commands.py --write` after changing a source command. Kimi Shell may not expose `KIMI_PLUGIN_ROOT`; in that case, use `/plugins info flowguard` to locate the enabled plugin before running its bundled CLI. The plugin remains centrally cataloged in `full-stack-plugins`; this source repository is maintained separately.
 
 CLI example:

README.zh-CN.md

@@ -46,13 +46,21 @@ flowchart LR
 # 2. 智能体分类后绑定本次任务
 /flowguard-context
 
-# 3. 实施过程中记录真实证据
+# 3. 创建项目级阶段文档 02/07/10
+/flowguard-init
+
+# 4. 读取与推进 docs/ 阶段文档
+/flowguard-stage
+
+# 5. 实施过程中记录真实证据
 /flowguard-evidence
 
-# 4. 写码、提交或发布前复核
+# 6. 写码、提交或发布前复核
 /flowguard-governance
 ```
 
+CLI 另提供 `validate`（产物内容、追溯矩阵与模块栈覆盖检查）。
+
 Kimi 将同源命令注册为带命名空间的 Markdown 命令，例如 `/flowguard:flowguard-discover`。`kimi-commands/` 由 `commands/*.json` 机械生成；修改 JSON 后运行 `python3 scripts/generate_kimi_commands.py --write`。若 Kimi Shell 未提供 `KIMI_PLUGIN_ROOT`，先通过 `/plugins info flowguard` 确认已启用插件的安装目录，再运行其自带 CLI。插件仍由 `full-stack-plugins` 统一登记与发布管理，源码仓库独立维护。
 
 对应 CLI：

hooks/flowguard_artifact_check.py

@@ -11,15 +11,10 @@
 ROOT = Path(__file__).resolve().parents[1]
 sys.path.insert(0, str(ROOT / "scripts"))
 
-from flowguard_lib import codereview_evidence, context, evidence, stage_docs, tool_scope  # noqa: E402
+from flowguard_lib import codereview_evidence, context, evidence, registry, stage_docs, tool_scope  # noqa: E402
 
-# artifact 文件名 → 阶段（用于降级与校验路由）
-ARTIFACT_STAGE = {
-    "01-requirements": "requirements", "02-architecture": "architecture",
-    "03-solution": "solution", "04-testcases": "testcases",
-    "05-hld": "hld", "06-lld": "lld", "07-standards": "standards",
-    "08-review": "review", "09-docs": "docs", "10-release": "release",
-}
+# artifact 文件名 → 阶段（失效提示路由；映射单源 registry.ARTIFACTS）
+ARTIFACT_STAGE = {aid: spec["stage"] for aid, spec in registry.ARTIFACTS.items()}
 CODEGUARD_MCP_TOOLS = ("mcp__codeguard__check_code_style", "mcp__codeguard__auto_fix")
 
 

scripts/flowguard_lib/stage_docs.py

@@ -4,13 +4,27 @@
 import re
 from pathlib import Path
 
-from . import ids, registry, state, validation
+from . import detect, ids, registry, state, validation
 
 VALID_STATUSES = (
     "pending", "in_progress", "pending_acceptance", "accepted",
     "inherited", "skipped", "invalidated",
 )
 SATISFIED = ("accepted", "inherited", "skipped")
+
+# 合法状态迁移表（单一事实源；测试做全矩阵断言）。
+# invalidated 是指纹派生态：正文/前置变化后由 read() 推导，不可直接写入。
+# 满足态之间禁止互跳与重复写：回改须显式退回 in_progress，重新走验收；
+# 仅 invalidated（正文已变）可直接重新验收。
+LEGAL = {
+    "pending": ("in_progress", "accepted", "inherited", "skipped"),
+    "in_progress": ("pending_acceptance", "accepted", "inherited", "skipped"),
+    "pending_acceptance": ("in_progress", "accepted", "inherited", "skipped"),
+    "invalidated": ("in_progress", "accepted", "inherited", "skipped"),
+    "accepted": ("in_progress",),
+    "inherited": ("in_progress",),
+    "skipped": ("in_progress",),
+}
 FIELDS = ("任务", "父任务", "阶段", "阶段状态", "规格事实源", "原生产物", "批准依据")
 _ROW = re.compile(r"^\|\s*([^|]+?)\s*\|\s*([^|]*?)\s*\|\s*$", re.MULTILINE)
 
@@ -275,6 +289,11 @@ def _advance_unlocked(root, task_id, stage, target, *, approval_ref=None, reason
     current = read(root, task_id, stage)
     if current["missing"]:
         raise StageDocError(f"阶段文档不存在: {path}")
+    allowed = LEGAL.get(current["status"], ())
+    if target not in allowed:
+        raise StageDocError(
+            f"非法状态迁移 {current['status']} → {target}"
+            f"（合法目标: {', '.join(allowed) or '无'}）")
     if target in SATISFIED and not approval_ref:
         raise StageDocError("验收、继承或跳过必须提供批准依据；理由文本不能代替批准")
     if target in ("inherited", "skipped") and not reason:
@@ -295,6 +314,12 @@ def _advance_unlocked(root, task_id, stage, target, *, approval_ref=None, reason
             issues = validation.validate_testcases(body, req_ids, root)
         elif stage == "08-review":
             issues = validation.validate_review(body)
+        elif stage == "02-architecture":
+            issues = validation.validate_architecture(body)
+        elif stage == "07-standards":
+            issues = validation.validate_standards(body, detect.detect(root)["modules"])
+        elif stage == "10-release":
+            issues = validation.validate_release(body, validation.known_task_ids(root))
         else:
             issues = []
         blocking = [item["message"] for item in issues if item["level"] in ("ERROR", "WARNING")]

scripts/flowguard_lib/state.py

@@ -1,11 +1,16 @@
 """状态锁与原子写工具；项目流程事实保存在 docs/（十阶段文档为唯一事实源）。"""
 import contextlib
-import fcntl
 import os
 import pathlib
 import stat
 import tempfile
 
+try:
+    import fcntl
+except ImportError:  # Windows：改用 msvcrt 字节锁，语义同为进程退出自动释放
+    fcntl = None
+    import msvcrt
+
 
 class StateError(Exception):
     pass
@@ -17,9 +22,12 @@ def state_lock(root):
     from .runtime import repository_state_dir
     lock = repository_state_dir(root, create=True) / ".lock"
     lock.parent.mkdir(parents=True, exist_ok=True)
-    fh = lock.open("w")
+    fh = lock.open("w+")
     try:
-        fcntl.flock(fh, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        if fcntl is not None:
+            fcntl.flock(fh, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        else:  # pragma: no cover - Windows
+            msvcrt.locking(fh.fileno(), msvcrt.LK_NBLCK, 1)
     except OSError:
         fh.close()
         raise StateError("状态文件被其它进程锁定，请稍后重试")

scripts/flowguard_lib/validation.py

@@ -234,6 +234,98 @@ def flush():
     return issues
 
 
+def known_task_ids(root):
+    """docs/features/ 下的 kebab 任务目录即已知功能清单。"""
+    base = Path(root) / "docs" / "features"
+    if not base.is_dir():
+        return []
+    return sorted(p.name for p in base.iterdir()
+                  if p.is_dir() and ids.is_kebab(p.name))
+
+
+def _section(text, start_re, end_re):
+    """截取 start_re 标题到 end_re 标题（不含）之间的正文。"""
+    out, active = [], False
+    for ln in text.splitlines():
+        if not active and re.match(start_re, ln):
+            active = True
+            continue
+        if active and re.match(end_re, ln):
+            break
+        if active:
+            out.append(ln)
+    return "\n".join(out)
+
+
+def validate_architecture(text):
+    """每条非占位 ADR 条目须带 `feature: <来源>` 与 `状态: proposed|accepted`（追加式格式契约）。
+
+    含 `<占位符>`/`{{...}}` 的条目视为模板示例，不参与机械校验。
+    """
+    lines = text.splitlines()
+    masked = _mask_code_fences(lines)
+    issues = []
+    for ln, hide in zip(lines, masked):
+        if hide or not re.match(r"^\s*-\s*ADR-", ln):
+            continue
+        if "<" in ln or "{{" in ln:
+            continue
+        if not re.search(r"feature:\s*\S+", ln):
+            issues.append(_issue("ERROR", "02-architecture.md",
+                                 f"ADR 条目缺 feature 来源标注: {ln.strip()!r}",
+                                 "追加式条目须带 `feature: <来源功能>`（禁止改写既有条目）"))
+        if not re.search(r"状态:\s*(proposed|accepted)", ln):
+            issues.append(_issue("ERROR", "02-architecture.md",
+                                 f"ADR 条目缺合法状态: {ln.strip()!r}",
+                                 "状态须为 proposed 或 accepted"))
+    return issues
+
+
+def validate_standards(text, modules):
+    """已填规范集（2. 规范集节出现非占位条目）须覆盖全部已注册模块的栈。
+
+    纯模板（规范集节全为占位示例）不参与机械校验；无栈模块（无法判定技术栈）跳过。
+    """
+    issues = []
+    body = _section(text, r"^##\s*2\.", r"^##\s*3\.")
+    headings = [ln.lower() for ln in body.splitlines()
+                if re.match(r"^###\s+", ln) and "<" not in ln and "{{" not in ln]
+    if not headings:
+        return issues
+    for name, mod in sorted((modules or {}).items()):
+        stack = (mod or {}).get("stack")
+        if not stack:
+            continue
+        tokens = {name.lower(), str(stack).lower()}
+        tokens.update(t for t in str(stack).lower().replace("_", "-").split("-") if len(t) >= 3)
+        if not any(tok in h for h in headings for tok in tokens):
+            issues.append(_issue("ERROR", "07-standards.md",
+                                 f"规范集未覆盖模块 {name}（技术栈 {stack}）",
+                                 "按模块栈补充 2.1 规范条目（覆盖所有已注册模块的栈）"))
+    return issues
+
+
+def validate_release(text, task_ids):
+    """发布内容表引用的功能须是已知 task-id；占位符行视为模板示例跳过。"""
+    issues = []
+    scope = _section(text, r"^##\s*3\.", r"^##\s*4\.")
+    known = set(task_ids)
+    for ln in scope.splitlines():
+        if "<" in ln or "{{" in ln:
+            continue
+        m = re.match(r"^\|\s*([^|]+?)\s*\|", ln)
+        if not m:
+            continue
+        cell = m.group(1).strip()
+        if not cell or set(cell) <= set(":- ") or cell.startswith("功能"):
+            continue
+        if cell not in known:
+            issues.append(_issue("ERROR", "10-release.md",
+                                 f"发布内容引用未知功能: {cell}",
+                                 "功能须是已存在的 docs/features/<task-id>"))
+    return issues
+
+
 def missing_tier2(refs, root=None):
     """refs: (skill, pkg, install_cmd)；已安装判定见 _installed。缺失产出 WARNING。"""
     out = []

scripts/flowguard_state.py

@@ -11,7 +11,7 @@
 sys.path.insert(0, str(Path(__file__).resolve().parent))
 
 from flowguard_lib import (  # noqa: E402
-    context, discovery, evidence, governance, registry,
+    context, detect, discovery, evidence, governance, registry,
     stage_docs, state, validation,
 )
 from flowguard_lib.diag import emit as emit_diag, envelope as mk_env  # noqa: E402
@@ -147,6 +147,14 @@ def cmd_validate(args):
             _die(mk_env("ERROR", "unknown_task", f"功能文档不存在: {args.task_id}",
                         "先用 context bind 创建 docs/features/<task-id>/"), as_json=args.json)
     issues = []
+    issues += validation.validate_architecture(
+        _read(stage_docs.path_for(root, "project", "02-architecture")))
+    issues += validation.validate_standards(
+        _read(stage_docs.path_for(root, "project", "07-standards")),
+        detect.detect(root)["modules"])
+    issues += validation.validate_release(
+        _read(stage_docs.path_for(root, "project", "10-release")),
+        validation.known_task_ids(root))
     for task in tasks:
         task_id = task["task_id"]
         if task.get("error"):

tests/test_docs_pipeline.py

@@ -184,7 +184,7 @@ def test_code_and_commit_wait_for_stage_documents_and_evidence(self):
         (self.root / "tests").mkdir()
         (self.root / "tests/test_refund.py").write_text("def test_refund():\n    assert True\n", encoding="utf-8")
         for path in (self.root / "docs").rglob("*.md"):
-            text = re.sub(r"\{\{[^}\n]+\}\}", "已确认", path.read_text(encoding="utf-8"))
+            text = re.sub(r"\{\{[^}\n]+\}\}", "已确认", path.read_text(encoding="utf-8")).replace("状态: 已确认", "状态: accepted")
             text = text.replace("测试文件: 已确认", "测试文件: tests/test_refund.py")
             text = text.replace("- 结论: fix|wontfix|deferred", "- 结论: fix")
             path.write_text(text, encoding="utf-8")
@@ -219,7 +219,7 @@ def test_release_acceptance_invalidates_when_listed_feature_docs_change(self):
             encoding="utf-8",
         )
         for path in (self.root / "docs").rglob("*.md"):
-            text = re.sub(r"\{\{[^}\n]+\}\}", "已确认", path.read_text(encoding="utf-8"))
+            text = re.sub(r"\{\{[^}\n]+\}\}", "已确认", path.read_text(encoding="utf-8")).replace("状态: 已确认", "状态: accepted")
             text = text.replace("测试文件: 已确认", "测试文件: tests/test_refund.py")
             text = text.replace("- 结论: fix|wontfix|deferred", "- 结论: fix")
             path.write_text(text, encoding="utf-8")
@@ -248,7 +248,7 @@ def test_release_scope_tracks_every_feature_with_escaped_table_text(self):
             encoding="utf-8",
         )
         for path in (self.root / "docs").rglob("*.md"):
-            text = re.sub(r"\{\{[^}\n]+\}\}", "已确认", path.read_text(encoding="utf-8"))
+            text = re.sub(r"\{\{[^}\n]+\}\}", "已确认", path.read_text(encoding="utf-8")).replace("状态: 已确认", "状态: accepted")
             text = text.replace("测试文件: 已确认", "测试文件: tests/test_refund.py")
             text = text.replace("- 结论: fix|wontfix|deferred", "- 结论: fix")
             path.write_text(text, encoding="utf-8")
@@ -454,7 +454,7 @@ def test_editing_accepted_stage_invalidates_its_document_fingerprint(self):
         self.bind()
         from flowguard_lib import stage_docs
         path = self.root / "docs/features/refund/01-requirements.md"
-        text = re.sub(r"\{\{[^}\n]+\}\}", "已确认", path.read_text(encoding="utf-8"))
+        text = re.sub(r"\{\{[^}\n]+\}\}", "已确认", path.read_text(encoding="utf-8")).replace("状态: 已确认", "状态: accepted")
         path.write_text(text, encoding="utf-8")
         stage_docs.advance(self.root, "refund", "01-requirements", "accepted", approval_ref="user-receipt:1")
 
@@ -466,7 +466,7 @@ def test_reaccepting_changed_requirement_does_not_silently_reuse_downstream_acce
         from flowguard_lib import stage_docs
         for stage in ("01-requirements", "02-architecture", "03-solution"):
             path = stage_docs.path_for(self.root, "refund", stage)
-            path.write_text(re.sub(r"\{\{[^}\n]+\}\}", "已确认", path.read_text(encoding="utf-8")),
+            path.write_text(re.sub(r"\{\{[^}\n]+\}\}", "已确认", path.read_text(encoding="utf-8")).replace("状态: 已确认", "状态: accepted"),
                             encoding="utf-8")
             stage_docs.advance(self.root, "refund", stage, "accepted", approval_ref=f"user:{stage}")
         requirement = stage_docs.path_for(self.root, "refund", "01-requirements")

tests/test_stage_transitions.py

@@ -0,0 +1,81 @@
+"""阶段状态迁移全矩阵：满足态互跳/重复写必须被拒，回改走 in_progress。"""
+import re
+import tempfile
+import unittest
+from pathlib import Path
+
+from flowguard_lib import context, stage_docs
+
+
+class TransitionMatrixTest(unittest.TestCase):
+    def setUp(self):
+        self.root = Path(tempfile.mkdtemp())
+        context.bind(self.root, session_id="s", task_id="f1", task_type="simple_change",
+                     spec_system="none", spec_ref=None)
+
+    def fill(self, stage="01-requirements"):
+        path = stage_docs.path_for(self.root, "f1", stage)
+        path.write_text(re.sub(r"\{\{[^}\n]+\}\}", "已确认", path.read_text(encoding="utf-8")),
+                        encoding="utf-8")
+
+    def advance(self, target, stage="01-requirements", **kw):
+        return stage_docs.advance(self.root, "f1", stage, target, **kw)
+
+    def test_legal_table_covers_all_statuses(self):
+        self.assertEqual(set(stage_docs.LEGAL), set(stage_docs.VALID_STATUSES))
+
+    def test_pending_to_accepted_is_legal(self):
+        self.fill()
+        self.assertEqual(self.advance("accepted", approval_ref="user:1")["status"], "accepted")
+
+    def test_accepted_cannot_be_rewritten_without_content_change(self):
+        self.fill()
+        self.advance("accepted", approval_ref="user:1")
+        with self.assertRaisesRegex(stage_docs.StageDocError, "非法状态迁移 accepted → accepted"):
+            self.advance("accepted", approval_ref="user:2")
+
+    def test_satisfied_states_cannot_flip_sideways(self):
+        self.fill()
+        self.advance("accepted", approval_ref="user:1")
+        for target in ("skipped", "inherited"):
+            with self.subTest(target=target):
+                with self.assertRaisesRegex(stage_docs.StageDocError, "非法状态迁移"):
+                    self.advance(target, approval_ref="user:x", reason="r")
+        self.advance("in_progress")
+        self.advance("skipped", approval_ref="user:skip", reason="范围不适用")
+        for target in ("accepted", "inherited", "skipped"):
+            with self.subTest(target=target):
+                with self.assertRaisesRegex(stage_docs.StageDocError, "非法状态迁移"):
+                    self.advance(target, approval_ref="user:x", reason="r")
+
+    def test_rework_must_go_through_in_progress(self):
+        self.fill()
+        self.advance("accepted", approval_ref="user:1")
+        self.assertEqual(self.advance("in_progress")["status"], "in_progress")
+        self.assertEqual(self.advance("accepted", approval_ref="user:2")["status"], "accepted")
+
+    def test_invalidated_allows_direct_reaccept(self):
+        self.fill()
+        self.advance("accepted", approval_ref="user:1")
+        path = stage_docs.path_for(self.root, "f1", "01-requirements")
+        path.write_text(path.read_text(encoding="utf-8") + "\n新增业务条件。\n", encoding="utf-8")
+        self.assertEqual(stage_docs.read(self.root, "f1", "01-requirements")["status"], "invalidated")
+        self.assertEqual(self.advance("accepted", approval_ref="user:recheck")["status"], "accepted")
+
+    def test_pending_and_invalidated_are_not_writable_targets(self):
+        self.fill()
+        self.advance("in_progress")
+        for target in ("pending", "invalidated"):
+            with self.subTest(target=target):
+                with self.assertRaises(stage_docs.StageDocError):
+                    self.advance(target)
+
+    def test_pending_acceptance_round_trip(self):
+        self.fill()
+        self.advance("in_progress")
+        self.assertEqual(self.advance("pending_acceptance")["status"], "pending_acceptance")
+        self.assertEqual(self.advance("accepted", approval_ref="user:1")["status"], "accepted")
+
+
+if __name__ == "__main__":
+    unittest.main()