diff --git a/session-mechanism/SKILL.md b/session-mechanism/SKILL.md
index d228c87..31415af 100644
--- a/session-mechanism/SKILL.md
+++ b/session-mechanism/SKILL.md
@@ -1028,6 +1028,8 @@ SKILL.md  install.py  install.log(装后生成)
 env/             roots.env(装后生成·不入库)  roots.env.example(入库样例)
                  🔴 「每台机器各不同的配置」只放这里（脚本是脚本 配置是配置）
 references/  architecture collab collab-detail rules deploy pitfalls taskgraph forensics manifest
+graph/           graphctl.py test_graphctl.py README.md
+                 ⇒ 图模式：与项目机制**并列的独立功能**（零常驻·零钩子·不读写现有三份文件）
 scripts/hooks/   6 份宿主钩子（10 条接线）
 scripts/lock/    handoff-guard.sh preflight-lock.sh handoff-status.py op-lock.sh
 scripts/         collabd.py board.py board_ext.py goalctl.py guard.py wake-session.py
@@ -1039,6 +1041,9 @@ assets/          board.html design-tokens.css
 
 ⚠️ 逐文件清单 ＋ 来源（provenance）⇒ `references/manifest.md`（**唯一权威**，⛔ 别按本树猜）。
 
+🔴 **图模式（与项目机制并列的独立功能）**：把目标拆成「节点 ＋ 依赖 ＋ 域」，按 `ready` 逐批放行；**零常驻、零钩子、不读写 `goal.json`／`tasks.json`／`collabd-state.json`**，停用就只是不再调用它。
+用法（建图 → `check` → `ready` → `start`／`report` → `list`／`show`）见 `graph/README.md`，自测 `python graph/test_graphctl.py`（应 `Ran 49 tests … OK`）。
+
 **⛔ 已无必需外部依赖（2026-10-04）**：原先声明「依赖 `agent-operating-rules`」——现已把**排版核心块收进本包并定为权威** ⇒ `references/03-回复排版-核心块.md`
 （2026-10-06 校正：原措辞是「内联副本」，与 03 号自称「权威源」打架 ⇒ 权威声明分裂）、把 `_env.py` 的技能库识别特征改为**候选数组（首个＝本包自己）** ⇒ **本包自包含**。📌 **2026-10-07 订正**：上面末句原写「**那个技能**若同时装着，属可选增强（多了跨项目排版/去 AI 味的完整版），⛔ 不装也能跑」——
 那个技能（`agent-operating-rules`）**已于 2026-10-06 整包并入本包并删除**，⛔ **现在没有它可装**（照旧句去找必然扑空）。
diff --git a/session-mechanism/graph/README.md b/session-mechanism/graph/README.md
index 0c895eb..631a156 100644
--- a/session-mechanism/graph/README.md
+++ b/session-mechanism/graph/README.md
@@ -81,7 +81,32 @@
 
 **两套可以同时存在**：图模式不读也不改现有那三份文件，现有机制也永远不会看到 `graph/` 目录。
 
-## 五、还没做的（下一步）
+## 五、怎么跑测试
+
+    python graph/test_graphctl.py -v
+
+全绿应看到 `Ran 49 tests ... OK`，用时 20 秒上下（每条用例起一次子进程）。
+
+测试只往临时目录写：`<GRAPH_TEST_TMP>/graph-test-<pid>/<用例名>`，默认 `GRAPH_TEST_TMP` = `AIProject/ai1net-dsh-server/tmp`，由每条用例的 `tearDown` ＋ 进程退出的 `atexit` 兜底删除。想换落点就设环境变量 `GRAPH_TEST_TMP`。
+
+清残留在本机是常态：跑完后 `tmp/` 下偶尔还能看到 `graph-test-<pid>/`（进程内已确认删空，是本机沙箱在进程退出后把删除路径又放回来了，与测试代码无关）。一条命令清掉：
+
+    rm -rf "E:/ProgramData/AIProject/ai1net-dsh-server/tmp"/graph-test-*
+
+八个用例类各管一摊：
+
+1、`T1_Init` —— 建图：`--kpi` 按条数生成节点（全角分号也认）、`--nodes-json` 带依赖与域、重复 id 报错、自动编号避让、`max_parallel`。
+2、`T2_CheckErrors` —— 硬错三条：环（含自环）、依赖指向不存在的节点、状态非法。
+3、`T3_CheckWarnings` —— 警告三类：无域、无判据、域重叠（`docs/` 与 `docs/ui/` 双向都算）。
+4、`T4_Ready` —— 就绪集：依赖未满足、域与在跑冲突、超 `max_parallel`（running 占名额）、两个域重叠的节点不进同一批。
+5、`T5_Flow` —— 流转：`pending→running→done`、`events.jsonl` 只追加、`blocked` 累加 `attempts`、回报后回带 `next_ready`。
+6、`T6_Illegal` —— 非法操作：`running`／`done` 再 `start` 拒绝，`blocked` 可重开；`--state` 只认 `done` / `blocked`；不存在的节点／图报错。
+7、`T7_Edge` —— 边界：单节点图、空节点图、gid 含中文（显式与自动生成各一条）、`list` 统计与空工作区。
+8、`T8_Isolation` —— 独立性：工作区里预埋的 `goal.json` / `tasks.json` / `collabd-state.json` 跑完字节不变，且不新建这类文件，写入范围只有 `<ws>/.workbuddy/graph/`。
+
+测试对 `graphctl.py` 的唯一要求是「能当命令行跑」，不 import 它的内部函数 —— 所以改内部实现不会连带改测试。
+
+## 六、还没做的（下一步）
 
 - 自动派活：读 `ready` 后自动建排期（要复用现有排期硬规则：cwds 正斜杠、workspace_scope、next_run_at 毫秒、scheduled_at ISO）。
 - 判定节点完成：现在靠回报 `--state done`，还没有「看证据够不够」的自动判定。
diff --git a/session-mechanism/graph/graphctl.py b/session-mechanism/graph/graphctl.py
index 7fe03d8..94c463c 100644
--- a/session-mechanism/graph/graphctl.py
+++ b/session-mechanism/graph/graphctl.py
@@ -42,7 +42,15 @@
 ## 就绪的判据（四条全中）
 
     ① 依赖全部 done   ② 自己还是 pending
-    ③ 域不与任何 running 节点重叠   ④ 不超 max_parallel
+    ③ 域不与任何 running 节点重叠，也不与本批已选节点重叠
+    ④ 不超 max_parallel（running 占名额）
+
+## 状态的走法
+
+    pending --start--> running --report done--> done
+                          |--report blocked--> blocked --start--> running（重试，attempts 已累加）
+
+`start` 只拒 `running`（已在跑）与 `done`（已完成）；`blocked` 可重开。
 
 ## 域的用法
 
@@ -174,7 +182,8 @@ def cmd_check(a) -> int:
             if d not in nodes:
                 errs.append("%s 的依赖 %s 不存在" % (nid, d))
         if not _norm_dom(n.get("domain")):
-            warns.append("%s 没有声明域 ⇒ 只能串行，不能与其他节点并行" % nid)
+            warns.append("%s 没有声明域 ⇒ 域比较视为不冲突，可与任意节点同批；"
+                         "保守做法是让它单独跑" % nid)
         if str(n.get("status")) not in STATUSES:
             errs.append("%s 的状态非法：%r" % (nid, n.get("status")))
         if not (n.get("accepts") or []):
@@ -220,12 +229,24 @@ def _ready(g: dict) -> dict:
         ready.append(nid)
 
     slots = max(0, cap - len(running))
-    picked = ready[:slots]
-    for nid in ready[slots:]:
-        waiting.append({"node": nid, "why": "并发上限 %d 已占满" % cap})
+    picked, backlog = [], []
+    for nid in ready:
+        if len(picked) >= slots:
+            backlog.append(nid)
+            waiting.append({"node": nid, "why": "并发上限 %d 已占满" % cap})
+            continue
+        dom = _norm_dom((nodes.get(nid) or {}).get("domain"))
+        clash = [p for p in picked
+                 if dom and _norm_dom((nodes.get(p) or {}).get("domain"))
+                 and _dom_overlap(dom, _norm_dom((nodes.get(p) or {}).get("domain")))]
+        if clash:
+            # 同批也不能撞域：check 已警告「永不同批」，ready 必须兑现
+            waiting.append({"node": nid, "why": "域与本批已选节点重叠：%s" % ", ".join(clash)})
+            continue
+        picked.append(nid)
 
     return {"gid": g.get("gid"), "running": running, "ready": picked,
-            "backlog": [x for x in ready[slots:]], "waiting": waiting,
+            "backlog": backlog, "waiting": waiting,
             "cap": cap, "slots": slots}
 
 
@@ -241,8 +262,11 @@ def cmd_start(a) -> int:
     n = (g.get("nodes") or {}).get(a.node)
     if not n:
         raise SystemExit("没有这个节点：%s" % a.node)
-    if str(n.get("status")) != "pending":
-        raise SystemExit("节点 %s 当前是 %s，不能开工" % (a.node, n.get("status")))
+    if str(n.get("status")) == "running":
+        raise SystemExit("节点 %s 已在跑，不能重复开工" % a.node)
+    if str(n.get("status")) == "done":
+        raise SystemExit("节点 %s 已完成，不能开工" % a.node)
+    # pending 与 blocked 都可开工：blocked 是「这轮没过」，重试时 attempts 保留
     n["status"] = "running"
     _save(a.ws, a.gid, g)
     _event(a.ws, a.gid, {"node": a.node, "to": "running", "by": a.by or ""})
@@ -293,7 +317,15 @@ def cmd_init(a) -> int:
 
     nodes = {}
     for i, n in enumerate(spec.get("nodes") or []):
-        nid = str(n.get("id") or "N%d" % (i + 1))
+        raw = n.get("id")
+        nid = str(raw).strip() if raw not in (None, "") else ""
+        if not nid:
+            k = i + 1
+            while ("N%d" % k) in nodes:      # 自动编号要避开已占用的 id
+                k += 1
+            nid = "N%d" % k
+        if nid in nodes:                     # 显式重复 ⇒ 硬错，⛔ 不静默覆盖前一个
+            raise SystemExit("节点 id 重复：%s" % nid)
         nodes[nid] = {
             "title": str(n.get("title") or nid),
             "deps": [str(x) for x in (n.get("deps") or [])],
diff --git a/session-mechanism/graph/test_graphctl.py b/session-mechanism/graph/test_graphctl.py
new file mode 100644
index 0000000..4b55f5b
--- /dev/null
+++ b/session-mechanism/graph/test_graphctl.py
@@ -0,0 +1,529 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""test_graphctl.py —— 「图模式」自动化测试（2026-10-18）。
+
+## 覆盖八类
+1 建图（--kpi / --nodes-json）            5 状态流转（pending→running→done、blocked 累计 attempts）
+2 校验硬错（环 / 悬空依赖 / 状态非法）      6 非法操作（重复开工 / 非法 state / 不存在的节点）
+3 校验警告（无域 / 无判据 / 域重叠）        7 边界（单节点 / 空节点 / gid 含中文 / 图不存在）
+4 就绪集（依赖 / 域冲突 / 并发上限）        8 独立性（只写临时目录，不碰现有机制文件）
+
+## 跑法
+    python graph/test_graphctl.py -v
+
+## 落点
+临时工作区 = `$GRAPH_TEST_TMP`（默认 `AIProject/ai1net-dsh-server/tmp`）/`graph-test-<pid>`/<用例名>，
+整体删除；每个用例一个干净工作区，互不影响。
+"""
+from __future__ import annotations
+
+import atexit
+import atexit
+import json
+import os
+import shutil
+import subprocess
+import sys
+import time
+import unittest
+from pathlib import Path
+
+HERE = Path(__file__).resolve().parent
+GRAPHCTL = HERE / "graphctl.py"
+PY = sys.executable
+
+_DEFAULT_TMP = "E:/ProgramData/AIProject/ai1net-dsh-server/tmp"
+TMP_ROOT = Path(os.environ.get("GRAPH_TEST_TMP") or _DEFAULT_TMP) / ("graph-test-%d" % os.getpid())
+
+_ENV = dict(os.environ)
+_ENV["PYTHONIOENCODING"] = "utf-8"
+
+
+# ── 基础设施 ────────────────────────────────────────────────────────────
+def _run(*args, stdin: str | None = None):
+    """跑一次 CLI，返回 (rc, stdout 文本, stderr 文本)。"""
+    p = subprocess.run([PY, str(GRAPHCTL), *[str(x) for x in args]],
+                       input=(stdin.encode("utf-8") if stdin is not None else b""),
+                       stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=_ENV)
+    return (p.returncode,
+            p.stdout.decode("utf-8", "replace"),
+            p.stderr.decode("utf-8", "replace"))
+
+
+def ok(*args, stdin: str | None = None) -> dict:
+    """要求 rc=0 并把 stdout 解析成 dict。"""
+    rc, out, err = _run(*args, stdin=stdin)
+    if rc != 0:
+        raise AssertionError("期望 rc=0，实际 rc=%d\nstderr=%s\nstdout=%s" % (rc, err, out))
+    return json.loads(out)
+
+
+def bad(*args, stdin: str | None = None):
+    """要求 rc!=0，返回 stderr（供断错误原因）。"""
+    rc, out, err = _run(*args, stdin=stdin)
+    if rc == 0:
+        raise AssertionError("期望失败，实际成功：stdout=%s" % out)
+    return err
+
+
+def _drop(p: Path, tries: int = 5) -> None:
+    """删临时目录：Windows 上刚写完的文件可能被杀软/索引器短暂占用，重试几次。"""
+    last = None
+    for _ in range(tries):
+        try:
+            shutil.rmtree(p)
+            return
+        except FileNotFoundError:
+            return
+        except Exception as e:      # noqa: BLE001 —— 重试后仍失败才报，带上原因
+            last = e
+            time.sleep(0.1)
+    sys.stderr.write("[cleanup] 删不掉（已重试 %d 次）：%s ⇒ %r\n" % (tries, p, last))
+
+
+class Base(unittest.TestCase):
+    def setUp(self):
+        self.ws = TMP_ROOT / self._testMethodName
+        shutil.rmtree(self.ws, ignore_errors=True)
+        self.ws.mkdir(parents=True, exist_ok=True)
+        self.gid = "g1"
+
+    def tearDown(self):
+        global _TD_COUNT
+        _TD_COUNT += 1
+        _drop(self.ws)
+
+    # -- 便捷动作 --
+    def gpath(self, gid=None) -> Path:
+        return self.ws / ".workbuddy" / "graph" / (gid or self.gid) / "graph.json"
+
+    def load(self, gid=None) -> dict:
+        return json.loads(self.gpath(gid).read_text(encoding="utf-8"))
+
+    def dump(self, g: dict, gid=None) -> None:
+        self.gpath(gid).write_text(json.dumps(g, ensure_ascii=False, indent=2),
+                                   encoding="utf-8")
+
+    def init(self, spec: dict | None = None, gid=None, title="测试目标"):
+        gid = gid or self.gid
+        args = ["init", "--ws", self.ws, "--title", title, "--gid", gid]
+        if spec is None:
+            args += ["--kpi", "判据一"]
+            return ok(*args)
+        return ok(*args, "--nodes-json", "-",
+                  stdin=json.dumps(spec, ensure_ascii=False))
+
+    def state(self, node, status):
+        g = self.load()
+        g["nodes"][node]["status"] = status
+        self.dump(g)
+
+    def events(self, gid=None) -> list:
+        p = self.ws / ".workbuddy" / "graph" / (gid or self.gid) / "events.jsonl"
+        return [json.loads(x) for x in p.read_text(encoding="utf-8").splitlines() if x.strip()]
+
+
+# ── 1 建图 ──────────────────────────────────────────────────────────────
+class T1_Init(Base):
+    def test_kpi_one_node_per_criterion(self):
+        ok("init", "--ws", self.ws, "--title", "三件事", "--gid", "k1",
+           "--kpi", "判据A;判据B；判据C")
+        g = self.load("k1")
+        self.assertEqual(sorted(g["nodes"]), ["N1", "N2", "N3"])
+        self.assertEqual(g["nodes"]["N1"]["accepts"], ["判据A"])
+        self.assertEqual(g["nodes"]["N3"]["accepts"], ["判据C"])   # 全角分号也认
+
+    def test_kpi_default_fields(self):
+        self.init()
+        n = self.load()["nodes"]["N1"]
+        self.assertEqual(n["status"], "pending")
+        self.assertEqual(n["attempts"], 0)
+        self.assertEqual(n["deps"], [])
+        self.assertEqual(n["evidence"], "")
+
+    def test_nodes_json_deps_and_domain(self):
+        self.init({"nodes": [
+            {"id": "A", "title": "改界面", "domain": ["ui/"], "accepts": ["能打开"], "deps": []},
+            {"id": "B", "title": "汇总", "domain": ["docs"], "accepts": ["齐了"], "deps": ["A"]},
+        ]})
+        g = self.load()
+        self.assertEqual(g["nodes"]["B"]["deps"], ["A"])
+        self.assertEqual(g["nodes"]["B"]["domain"], ["docs/"])   # 自动补尾斜杠
+
+    def test_nodes_json_max_parallel(self):
+        self.init({"max_parallel": 2, "nodes": [{"id": "A", "accepts": ["x"]}]})
+        self.assertEqual(self.load()["meta"]["max_parallel"], 2)
+
+    def test_init_needs_kpi_or_nodes_json(self):
+        err = bad("init", "--ws", self.ws, "--title", "空手")
+        self.assertIn("--kpi", err)
+
+    def test_duplicate_id_rejected(self):
+        """显式 id 重复必须报错，不能静默覆盖掉前一个节点。"""
+        err = bad("init", "--ws", self.ws, "--title", "重名", "--gid", self.gid,
+                  "--nodes-json", "-",
+                  stdin=json.dumps({"nodes": [
+                      {"id": "A", "accepts": ["一"]},
+                      {"id": "A", "accepts": ["二"]}]}, ensure_ascii=False))
+        self.assertIn("重复", err)
+
+    def test_missing_id_does_not_collide(self):
+        """省略 id 时自动编号要避开已有 id。"""
+        self.init({"nodes": [{"id": "N1", "accepts": ["一"]}, {"accepts": ["二"]}]})
+        self.assertEqual(len(self.load()["nodes"]), 2)
+
+    def test_init_writes_event(self):
+        self.init()
+        self.assertEqual(self.events()[0]["event"], "init")
+
+
+# ── 2 校验硬错 ──────────────────────────────────────────────────────────
+class T2_CheckErrors(Base):
+    def test_cycle(self):
+        self.init({"nodes": [{"id": "A", "deps": ["B"], "accepts": ["x"]},
+                             {"id": "B", "deps": ["A"], "accepts": ["y"]}]})
+        rc, out, _ = _run("check", "--ws", self.ws, "--gid", self.gid)
+        r = json.loads(out)
+        self.assertEqual(rc, 1)
+        self.assertFalse(r["ok"])
+        self.assertTrue(any("环" in e for e in r["errors"]))
+
+    def test_self_cycle(self):
+        self.init({"nodes": [{"id": "A", "deps": ["A"], "accepts": ["x"]}]})
+        r = ok("check", "--ws", self.ws, "--gid", self.gid) if False else \
+            json.loads(_run("check", "--ws", self.ws, "--gid", self.gid)[1])
+        self.assertTrue(any("环" in e for e in r["errors"]))
+
+    def test_dangling_dep(self):
+        self.init({"nodes": [{"id": "A", "deps": ["Z"], "accepts": ["x"]}]})
+        r = json.loads(_run("check", "--ws", self.ws, "--gid", self.gid)[1])
+        self.assertFalse(r["ok"])
+        self.assertTrue(any("Z" in e and "不存在" in e for e in r["errors"]))
+
+    def test_illegal_status(self):
+        self.init()
+        self.state("N1", "weird")
+        r = json.loads(_run("check", "--ws", self.ws, "--gid", self.gid)[1])
+        self.assertFalse(r["ok"])
+        self.assertTrue(any("状态非法" in e for e in r["errors"]))
+
+    def test_clean_graph_ok(self):
+        self.init({"nodes": [{"id": "A", "domain": ["ui/"], "accepts": ["x"]},
+                             {"id": "B", "domain": ["docs/"], "accepts": ["y"], "deps": ["A"]}]})
+        r = ok("check", "--ws", self.ws, "--gid", self.gid)
+        self.assertTrue(r["ok"])
+        self.assertEqual(r["errors"], [])
+        self.assertEqual(r["nodes"], 2)
+
+
+# ── 3 校验警告 ──────────────────────────────────────────────────────────
+class T3_CheckWarnings(Base):
+    def test_no_domain(self):
+        self.init({"nodes": [{"id": "A", "accepts": ["x"]}]})
+        r = ok("check", "--ws", self.ws, "--gid", self.gid)
+        self.assertTrue(r["ok"])
+        self.assertTrue(any("没有声明域" in w for w in r["warnings"]))
+
+    def test_no_accepts(self):
+        self.init({"nodes": [{"id": "A", "domain": ["ui/"]}]})
+        r = ok("check", "--ws", self.ws, "--gid", self.gid)
+        self.assertTrue(any("没有验收判据" in w for w in r["warnings"]))
+
+    def test_overlap_nested(self):
+        self.init({"nodes": [{"id": "A", "domain": ["docs/"], "accepts": ["x"]},
+                             {"id": "B", "domain": ["docs/ui/"], "accepts": ["y"]}]})
+        r = ok("check", "--ws", self.ws, "--gid", self.gid)
+        self.assertTrue(any("重叠" in w for w in r["warnings"]))
+
+    def test_overlap_reverse_nested(self):
+        self.init({"nodes": [{"id": "A", "domain": ["docs/ui/"], "accepts": ["x"]},
+                             {"id": "B", "domain": ["docs/"], "accepts": ["y"]}]})
+        r = ok("check", "--ws", self.ws, "--gid", self.gid)
+        self.assertTrue(any("重叠" in w for w in r["warnings"]))
+
+    def test_disjoint_no_warning(self):
+        self.init({"nodes": [{"id": "A", "domain": ["ui/"], "accepts": ["x"]},
+                             {"id": "B", "domain": ["docs-ui/"], "accepts": ["y"]}]})
+        r = ok("check", "--ws", self.ws, "--gid", self.gid)
+        self.assertFalse(any("重叠" in w for w in r["warnings"]))
+
+
+# ── 4 就绪集 ────────────────────────────────────────────────────────────
+class T4_Ready(Base):
+    def test_all_pending_no_deps(self):
+        self.init({"nodes": [{"id": "A", "domain": ["ui/"], "accepts": ["x"]},
+                             {"id": "B", "domain": ["docs/"], "accepts": ["y"]}]})
+        r = ok("ready", "--ws", self.ws, "--gid", self.gid)
+        self.assertEqual(sorted(r["ready"]), ["A", "B"])
+        self.assertEqual(r["backlog"], [])
+        self.assertEqual(r["running"], [])
+
+    def test_dep_unmet(self):
+        self.init({"nodes": [{"id": "A", "domain": ["ui/"], "accepts": ["x"]},
+                             {"id": "B", "domain": ["docs/"], "accepts": ["y"], "deps": ["A"]}]})
+        r = ok("ready", "--ws", self.ws, "--gid", self.gid)
+        self.assertEqual(r["ready"], ["A"])
+        self.assertTrue(any(w["node"] == "B" and "依赖未完成" in w["why"]
+                            for w in r["waiting"]))
+
+    def test_dep_done_releases(self):
+        self.init({"nodes": [{"id": "A", "domain": ["ui/"], "accepts": ["x"]},
+                             {"id": "B", "domain": ["docs/"], "accepts": ["y"], "deps": ["A"]}]})
+        ok("start", "--ws", self.ws, "--gid", self.gid, "--node", "A")
+        ok("report", "--ws", self.ws, "--gid", self.gid,
+           "--node", "A", "--state", "done", "--evidence", "截图在 ui/shot.png")
+        r = ok("ready", "--ws", self.ws, "--gid", self.gid)
+        self.assertEqual(r["ready"], ["B"])
+
+    def test_domain_clash_with_running(self):
+        self.init({"nodes": [{"id": "A", "domain": ["docs/"], "accepts": ["x"]},
+                             {"id": "B", "domain": ["docs/ui/"], "accepts": ["y"]}]})
+        ok("start", "--ws", self.ws, "--gid", self.gid, "--node", "A")
+        r = ok("ready", "--ws", self.ws, "--gid", self.gid)
+        self.assertNotIn("B", r["ready"])
+        self.assertTrue(any(w["node"] == "B" and "域与在跑的冲突" in w["why"]
+                            for w in r["waiting"]))
+
+    def test_max_parallel_overflow(self):
+        self.init({"max_parallel": 2, "nodes": [
+            {"id": x, "domain": ["%s/" % x.lower()], "accepts": ["ok"]}
+            for x in ("A", "B", "C", "D")]})
+        r = ok("ready", "--ws", self.ws, "--gid", self.gid)
+        self.assertEqual(len(r["ready"]), 2)
+        self.assertEqual(len(r["backlog"]), 2)
+        self.assertEqual(r["cap"], 2)
+
+    def test_max_parallel_counts_running(self):
+        self.init({"max_parallel": 2, "nodes": [
+            {"id": x, "domain": ["%s/" % x.lower()], "accepts": ["ok"]}
+            for x in ("A", "B", "C")]})
+        ok("start", "--ws", self.ws, "--gid", self.gid, "--node", "A")
+        r = ok("ready", "--ws", self.ws, "--gid", self.gid)
+        self.assertEqual(r["slots"], 1)
+        self.assertEqual(len(r["ready"]), 1)
+        self.assertEqual(r["running"], ["A"])
+
+    def test_overlapping_peers_not_in_same_batch(self):
+        """两个域重叠的节点⛔ 不能出现在同一个 ready 批次里（check 说「永不同批」，ready 要兑现）。"""
+        self.init({"nodes": [{"id": "A", "domain": ["docs/"], "accepts": ["x"]},
+                             {"id": "B", "domain": ["docs/ui/"], "accepts": ["y"]}]})
+        r = ok("ready", "--ws", self.ws, "--gid", self.gid)
+        self.assertEqual(len(r["ready"]), 1)
+        left = [w for w in r["waiting"] if "重叠" in w["why"]]
+        self.assertEqual(len(left), 1, r)
+
+    def test_domainless_parallelises(self):
+        """没有域的节点按现状不参与域比较 ⇒ 可与别的节点同批（check 的警告文案须与此一致）。"""
+        self.init({"nodes": [{"id": "A", "accepts": ["x"]},
+                             {"id": "B", "domain": ["docs/"], "accepts": ["y"]}]})
+        r = ok("ready", "--ws", self.ws, "--gid", self.gid)
+        self.assertEqual(sorted(r["ready"]), ["A", "B"])
+
+
+# ── 5 状态流转 ──────────────────────────────────────────────────────────
+class T5_Flow(Base):
+    def test_pending_running_done(self):
+        self.init()
+        r = ok("start", "--ws", self.ws, "--gid", self.gid, "--node", "N1", "--by", "执行会话-x")
+        self.assertEqual(r["status"], "running")
+        self.assertEqual(self.load()["nodes"]["N1"]["status"], "running")
+        r = ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "N1",
+               "--state", "done", "--evidence", "产物在位")
+        self.assertEqual(r["status"], "done")
+        g = self.load()
+        self.assertEqual(g["nodes"]["N1"]["status"], "done")
+        self.assertEqual(g["nodes"]["N1"]["evidence"], "产物在位")
+
+    def test_events_append_only(self):
+        self.init()
+        ok("start", "--ws", self.ws, "--gid", self.gid, "--node", "N1")
+        ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "N1", "--state", "done")
+        ev = self.events()
+        self.assertEqual(ev[0]["event"], "init")
+        self.assertEqual([e.get("to") for e in ev[1:]], ["running", "done"])
+
+    def test_blocked_bumps_attempts(self):
+        self.init()
+        ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "N1", "--state", "blocked")
+        n = self.load()["nodes"]["N1"]
+        self.assertEqual(n["status"], "blocked")
+        self.assertEqual(n["attempts"], 1)
+
+    def test_blocked_twice(self):
+        self.init()
+        ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "N1", "--state", "blocked")
+        ok("start", "--ws", self.ws, "--gid", self.gid, "--node", "N1")
+        ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "N1", "--state", "blocked")
+        self.assertEqual(self.load()["nodes"]["N1"]["attempts"], 2)
+
+    def test_done_does_not_touch_attempts(self):
+        self.init()
+        ok("start", "--ws", self.ws, "--gid", self.gid, "--node", "N1")
+        ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "N1", "--state", "done")
+        self.assertEqual(self.load()["nodes"]["N1"]["attempts"], 0)
+
+    def test_report_returns_next_ready(self):
+        self.init({"nodes": [{"id": "A", "domain": ["ui/"], "accepts": ["x"]},
+                             {"id": "B", "domain": ["docs/"], "accepts": ["y"], "deps": ["A"]}]})
+        ok("start", "--ws", self.ws, "--gid", self.gid, "--node", "A")
+        r = ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "A", "--state", "done")
+        self.assertEqual(r["next_ready"], ["B"])
+
+
+# ── 6 非法操作 ──────────────────────────────────────────────────────────
+class T6_Illegal(Base):
+    def test_start_running_rejected(self):
+        self.init()
+        ok("start", "--ws", self.ws, "--gid", self.gid, "--node", "N1")
+        err = bad("start", "--ws", self.ws, "--gid", self.gid, "--node", "N1")
+        self.assertIn("不能", err)
+
+    def test_start_done_rejected(self):
+        self.init()
+        ok("start", "--ws", self.ws, "--gid", self.gid, "--node", "N1")
+        ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "N1", "--state", "done")
+        bad("start", "--ws", self.ws, "--gid", self.gid, "--node", "N1")
+
+    def test_start_blocked_allowed_for_retry(self):
+        """blocked 要能重开（否则 attempts 白记，且只能手改 JSON 才能重试）。"""
+        self.init()
+        ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "N1", "--state", "blocked")
+        r = ok("start", "--ws", self.ws, "--gid", self.gid, "--node", "N1")
+        self.assertEqual(r["status"], "running")
+        self.assertEqual(self.load()["nodes"]["N1"]["attempts"], 1)
+
+    def test_state_only_done_or_blocked(self):
+        self.init()
+        err = bad("report", "--ws", self.ws, "--gid", self.gid, "--node", "N1", "--state", "pending")
+        self.assertIn("done", err)
+
+    def test_unknown_node_start(self):
+        self.init()
+        err = bad("start", "--ws", self.ws, "--gid", self.gid, "--node", "ZZ")
+        self.assertIn("没有这个节点", err)
+
+    def test_unknown_node_report(self):
+        self.init()
+        err = bad("report", "--ws", self.ws, "--gid", self.gid, "--node", "ZZ", "--state", "done")
+        self.assertIn("没有这个节点", err)
+
+    def test_unknown_gid(self):
+        err = bad("show", "--ws", self.ws, "--gid", "不存在")
+        self.assertIn("找不到图", err)
+
+
+# ── 7 边界 ──────────────────────────────────────────────────────────────
+class T7_Edge(Base):
+    def test_single_node(self):
+        self.init()
+        r = ok("ready", "--ws", self.ws, "--gid", self.gid)
+        self.assertEqual(r["ready"], ["N1"])
+        ok("start", "--ws", self.ws, "--gid", self.gid, "--node", "N1")
+        ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "N1", "--state", "done")
+        r = ok("ready", "--ws", self.ws, "--gid", self.gid)
+        self.assertEqual(r["ready"], [])
+        self.assertEqual(r["waiting"], [])
+
+    def test_empty_graph(self):
+        self.init({"nodes": []})
+        self.assertEqual(self.load()["nodes"], {})
+        self.assertEqual(ok("check", "--ws", self.ws, "--gid", self.gid)["nodes"], 0)
+        r = ok("ready", "--ws", self.ws, "--gid", self.gid)
+        self.assertEqual((r["ready"], r["backlog"], r["waiting"]), ([], [], []))
+        self.assertEqual(len(ok("list", "--ws", self.ws)["graphs"]), 1)
+
+    def test_chinese_gid_explicit(self):
+        ok("init", "--ws", self.ws, "--title", "中文", "--gid", "20261018-2235-界面改造",
+           "--kpi", "看上去对")
+        g = ok("show", "--ws", self.ws, "--gid", "20261018-2235-界面改造")
+        self.assertEqual(g["gid"], "20261018-2235-界面改造")
+        self.assertEqual(len(ok("list", "--ws", self.ws)["graphs"]), 1)
+
+    def test_chinese_title_auto_gid(self):
+        r = ok("init", "--ws", self.ws, "--title", "给三块功能各加一份文档", "--kpi", "一;二")
+        self.assertIn("给三块功能", r["gid"])
+        self.assertEqual(r["nodes"], 2)
+        g = ok("show", "--ws", self.ws, "--gid", r["gid"])
+        self.assertEqual(len(g["nodes"]), 2)
+
+    def test_list_counts_by_status(self):
+        self.init({"nodes": [{"id": "A", "domain": ["a/"], "accepts": ["x"]},
+                             {"id": "B", "domain": ["b/"], "accepts": ["y"]}]})
+        ok("start", "--ws", self.ws, "--gid", self.gid, "--node", "A")
+        row = ok("list", "--ws", self.ws)["graphs"][0]
+        self.assertEqual(row["nodes"], 2)
+        self.assertEqual(row["by_status"], {"running": 1, "pending": 1})
+
+    def test_list_empty_ws(self):
+        self.assertEqual(ok("list", "--ws", self.ws)["graphs"], [])
+
+
+# ── 8 独立性 ────────────────────────────────────────────────────────────
+class T8_Isolation(Base):
+    def _full_lifecycle(self):
+        self.init({"nodes": [{"id": "A", "domain": ["ui/"], "accepts": ["x"]},
+                             {"id": "B", "domain": ["docs/"], "accepts": ["y"], "deps": ["A"]}]})
+        ok("check", "--ws", self.ws, "--gid", self.gid)
+        ok("ready", "--ws", self.ws, "--gid", self.gid)
+        ok("start", "--ws", self.ws, "--gid", self.gid, "--node", "A")
+        ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "A", "--state", "done")
+        ok("show", "--ws", self.ws, "--gid", self.gid)
+        ok("list", "--ws", self.ws)
+
+    def test_sentinel_mechanism_files_untouched(self):
+        """工作区里已有的 goal.json / tasks.json / collabd-state.json 一个字节都不许变。"""
+        sentinels = {
+            "goal.json": '{"goal": "sentinel", "version": 7}\n',
+            "tasks.json": '{"tasks": [{"id": "t1"}]}\n',
+            ".workbuddy/collab/collabd-state.json": '{"mode": "sentinel"}\n',
+        }
+        before = {}
+        for rel, body in sentinels.items():
+            p = self.ws / rel
+            p.parent.mkdir(parents=True, exist_ok=True)
+            p.write_text(body, encoding="utf-8")
+            before[rel] = p.read_bytes()
+
+        self._full_lifecycle()
+
+        for rel in sentinels:
+            self.assertEqual((self.ws / rel).read_bytes(), before[rel], "%s 被改动了" % rel)
+
+    def test_no_mechanism_files_created(self):
+        self._full_lifecycle()
+        for name in ("goal.json", "tasks.json", "collabd-state.json"):
+            hits = [p for p in self.ws.rglob(name)]
+            self.assertEqual(hits, [], "不该出现的文件：%s" % hits)
+
+    def test_writes_only_under_mark_dir(self):
+        """图模式的写入范围只有 <ws>/.workbuddy/graph/。"""
+        self._full_lifecycle()
+        top = sorted(p.name for p in self.ws.iterdir())
+        self.assertEqual(top, [".workbuddy"])
+        self.assertEqual(sorted(p.name for p in (self.ws / ".workbuddy").iterdir()), ["graph"])
+        self.assertEqual(sorted(p.name for p in (self.ws / ".workbuddy" / "graph").iterdir()), [self.gid])
+
+    def test_temp_root_is_under_project_tmp(self):
+        """临时工作区落在项目 tmp 下（跑完由 tearDownModule 删）。"""
+        self.assertIn("tmp", [p.lower() for p in TMP_ROOT.parts])
+        self.assertTrue(str(TMP_ROOT).startswith(str(Path(_DEFAULT_TMP).parent.parent)) or
+                        os.environ.get("GRAPH_TEST_TMP"), TMP_ROOT)
+
+
+def tearDownModule():
+    _drop(TMP_ROOT)
+
+
+_TD_COUNT = 0
+
+
+def _final_cleanup():
+    """兜底：不管 unittest 的模块夹具跑没跑，进程退出前把临时根删干净。"""
+    _drop(TMP_ROOT)
+
+
+atexit.register(_final_cleanup)
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)