From ce05207aa246a4a6f85d14629291ff2cf299232c Mon Sep 17 00:00:00 2001 From: maogeigei Date: Mon, 19 Oct 2026 23:57:40 +0800 Subject: [PATCH] =?UTF-8?q?graph:=20=E8=A1=A5=E9=BD=90=E5=9B=BE=E6=A8=A1?= =?UTF-8?q?=E5=BC=8F=E5=89=A9=E4=BD=99=E5=9B=9B=E4=BB=B6=EF=BC=88=E9=AA=8C?= =?UTF-8?q?=E6=94=B6=E6=9C=BA=E5=88=A4/=E5=9B=9E=E9=80=80=E8=BE=B9/?= =?UTF-8?q?=E8=BF=90=E8=A1=8C=E6=9C=9F=E6=94=B9=E5=9B=BE/=E8=B7=A8?= =?UTF-8?q?=E5=9B=BE=E5=AD=90=E5=9B=BE=EF=BC=89=EF=BC=8C163=20=E6=9D=A1?= =?UTF-8?q?=E6=B5=8B=E8=AF=95=E5=85=A8=E7=BB=BF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - F1 accepts 机判：report done 前按 artifact:/evidence: 逐条校验，缺什么报什么，不许绕过 - F2 rework_to 回退边：判不过回到指定上游重做，attempts/reworks/事件齐备，人审驳回同路 - F3 运行期改图：add-node / remove-edge，目录锁+唯一临时名+原子替换+_rev 回读，并发不丢更新 - F4 跨图子图：subgraph.gid 引用同工作区另一张图，deps/when/rework_to 加前缀，父等收敛；缺图/自引用/跨图环/嵌套均硬报错 - 测试：122 → 163，新增 T20-T23；顺带修测试进程被本机批量删除守卫误杀的假红 --- session-mechanism/graph/README.md | 107 +++- session-mechanism/graph/graphctl.py | 649 ++++++++++++++++++++++- session-mechanism/graph/test_graphctl.py | 410 +++++++++++++- 3 files changed, 1135 insertions(+), 31 deletions(-) diff --git a/session-mechanism/graph/README.md b/session-mechanism/graph/README.md index 7ae04ea..2321196 100644 --- a/session-mechanism/graph/README.md +++ b/session-mechanism/graph/README.md @@ -1,8 +1,10 @@ # 图模式（独立功能） -> 2026-10-18 新增；2026-10-19 补「租约超时 / 重试与触发规则 / 动态展开」三件批量能力。 +> 2026-10-18 新增；2026-10-19 补「租约超时 / 重试与触发规则 / 动态展开」三件批量能力，同日再补「验收机判 / 回退边 / 运行期改图 / 跨图子图」四件。 > **这一套与现有「项目机制」完全并列**：不读写 `goal.json`、`tasks.json`、`collabd-state.json`，不参与六道闸，不接管常驻循环，不改变现有任何会话的行为。 > 停用方式：不调用它就行；它没有常驻进程、没有注册任何钩子、没有改任何现有文件。 +> +> 命令一览：`init` `check` `ready` `start` `report` **`rework`** **`add-node`** **`remove-edge`** `expand` `approve` `cancel` `tick` `plan` `progress` `show` `list`（加粗为 2026-10-19 新增）。 ## 一、它解决什么 @@ -214,6 +216,35 @@ `subgraph` 里还能再套 `subgraph`，前缀会一层层叠上去。 +### 跨图子图：引用同工作区的另一张图 + +除内联外，`subgraph` 还可以写 `{"gid": "<另一张图的 gid>"}`，把那张图的节点定义搬进来当本图的子流程： + + # 先建被引用的那张（同工作区、任何 gid） + python graph/graphctl.py init --ws "" --title "验证流程" --gid gverify --nodes-json - + {"nodes": [{"id": "S1", "domain": ["docs/"], "accepts": ["artifact:docs/a.md"]}, + {"id": "S2", "domain": ["docs/"], "accepts": ["artifact:docs/b.md"], + "deps": ["S1"]}]} + + # 再建引用它的那张 + python graph/graphctl.py init --ws "" --title "主流程" --gid gmain --nodes-json - + {"nodes": [{"id": "P", "subgraph": {"gid": "gverify"}}, + {"id": "Q", "deps": ["P"], "accepts": ["汇总写完"]}]} + + python graph/graphctl.py check --ws "" --gid gmain # 引用在不在、成没成环，这里就报 + +展开后子节点 id 是 `P/S1`、`P/S2`，父节点 `P` 进 `gating`，**等它的子节点全部收敛**才算完（全 `done` ⇒ `done`；有不满意的 ⇒ `blocked`），下游 `Q` 在 `P` 收敛前不派。 + +规则五条： + +1. **只搬定义、不搬状态** —— `title` / `deps` / `domain` / `accepts` / `deps_rule` / `on_dep_fail` / `when` / `retry` / `lease_sec` / `approval` / `rework_to` 都搬过去，被引用图里"跑到哪一步"不带过来（引用即重跑一遍）。子节点上会记 `origin_graph`，看板能看出它是从哪张图来的。 +2. 子图内部的 `deps`、`when.node`、`rework_to` 都加父前缀（`S1` ⇒ `P/S1`），写法与内联子图一致。 +3. **不嵌套**：被引用的图里若有 `map` 或 `subgraph`，`check` 直接报硬错（引用关系已经是一层，再叠会失控）。 +4. **不许成环**：引用自己、`A→B→A`、超过 6 层引用链，`check` 与展开时都报错。 +5. `subgraph` 里 `nodes` 与 `gid` **只能选一种**，同时写报错。 + +⚠️ 被引用的图**改了定义不会回溯**已展开的子节点 —— 要按新定义跑，用 `rework` 把父节点退回去重展开（见 §三之十一）。 + ## 三之七、人审节点与取消 人审：节点写 `"approval": true`，回报 `done` 后不进 `done`，而是进 `awaiting` 等人过目： @@ -264,6 +295,65 @@ 输出 `total` / `by_status` / `percent`（分母排除 `skipped` 与 `cancelled`）、`blocked` / `awaiting`、`critical_path`（**未完成节点之间的最长依赖链**，告诉你哪条链决定工期）。 +## 三之十、验收判据机判（`accepts`） + +节点的 `accepts` 不只是给人看的清单 —— 报 `done` 时**先按它校验产物与证据**，过不了就不许置 `done`，并逐条说明缺什么。 + +两种写法： + + "artifact:docs/*.md" 产物里至少一条匹配这个通配 + "artifact:>=2" 产物至少 2 条 + "evidence:测试通过" 证据里必须出现「测试通过」 + "evidence:全绿,零回归" 多个关键词（`,` 或 `，` 分隔）⇒ 全部都要出现 + "evidence:>=20" 证据至少 20 个字符 + + {"kind": "artifact", "glob": "docs/*.md"} 等价写法 + {"kind": "artifact", "min": 2} + {"kind": "evidence", "contains": ["全绿", "零回归"]} + {"kind": "evidence", "min": 20} + +**纯文本条目（如 `"产物在位"`）＝描述性判据，不参与机判** —— 老图照旧能用，不会被新闸误杀。 + +校验时读的是**本次 ＋ 历史**的合并视图（`--artifact` 是追加语义），路径比较时 `\` 与 `/` 等价。不通过长这样： + + python graph/graphctl.py report --ws "" --gid "" --node A --state done --evidence "全绿" + 节点 A 验收不通过（不置 done）： + - 缺产物 artifact:docs/a.md（现有产物：无） + - 证据缺关键词 evidence:全绿,零回归（现缺：零回归） + 补齐产物／证据后重新 report（accepts 不可绕过） + +`check` 也会校验语法：`"artifact:"` 这种空参数、`{"kind": "nope"}` 这种非法对象，建图时就被拒。 + +## 三之十一、回退边（`rework_to`） + +节点写 `"rework_to": "A"`（也可写列表，取第一个）。判不过时**回到指定的上游节点重做**，而不是只能自己重试： + + python graph/graphctl.py rework --ws "" --gid "" --node Q --reason "输出不合格" + python graph/graphctl.py rework --ws "" --gid "" --node Q --to A # 临时指定目标 + +做的事： + +1. **目标必须是上游**（`Q` 的传递依赖），⛔ 不能是自己、不能是下游 —— `check` 也会提前报。 +2. 目标及其**全部传递下游**复位 `pending`：清租约、`done_at`、`awaiting`、驳回原因、重试等待窗。 +3. 复位时把 `evidence` / `artifacts` 挪进 `prev_evidence` / `prev_artifacts` 再清空 —— 重做之后必须**重新交产物**，免得旧产物把验收闸蒙过去。 +4. 目标 `attempts +1`、全链 `reworks +1`，`rework_from` 记是谁判不过的，并写一条 `event=rework` 的流水（含 `reset` 名单与原因）。 +5. 链上有已展开（`map`／子图）的节点 ⇒ **收回它的子节点、清掉 `expanded`**，下次 `ready`／`tick` 按当时的输入重新展开，不留半截子节点。收回的 id 在输出的 `dropped_children` 里。 +6. 链上有 `running` 节点 ⇒ 报错停手（先 `cancel` 或等它结束），⛔ 不硬抢。 + +人审那边也接上了：`approve --reject` 时若节点带 `rework_to`，**自动走同一套回退**（人审判不过就是判不过），输出里 `rework.auto = true`。 + +## 三之十二、运行期改图（加节点、删边） + +图跑起来以后还能改结构，两条命令： + + python graph/graphctl.py add-node --ws "" --gid "" --node N9 \ + --title "补一份索引" --deps A,B --domain docs/index/ --accepts "artifact:docs/index.md" + python graph/graphctl.py remove-edge --ws "" --gid "" --from A --to B + +- `add-node`：新节点进 `pending`，`deps` 必须都是已存在的节点；`--accepts` 可重复、写法同 §三之十；还能带 `--deps-rule` / `--rework-to` / `--approval`。id 重复、依赖悬空都直接报错。 +- `remove-edge`：删掉「`B` 依赖 `A`」这一条。边不存在就报错（⛔ 不静默空操作），删完立刻按新依赖重算就绪集。 +- **并发安全**：两条命令与 `start` / `report` 共用同一把目录锁（`_Lock` ＋ 唯一临时名 ＋ `_rev` 回读核对），六个进程同时加节点，六个节点全在、`_rev` 前进 6。实测见 `T22_Mutate`。 + ## 四、与现有机制的关系 | 事项 | 现有机制 | 图模式 | @@ -281,7 +371,7 @@ python graph/test_graphctl.py -v -全绿应看到 `Ran 122 tests ... OK`，用时 100 秒上下（每条用例起若干次子进程；52 条是初版基线，40 条是 2026-10-19 为三件批量能力补的，30 条是同日为条件分支／子图／人审／取消／分池／派活／进度补的）。 +全绿应看到 `Ran 163 tests ... OK`，用时 300 秒上下（每条用例起若干次子进程；52 条是初版基线，40 条是 2026-10-19 为三件批量能力补的，30 条是同日的条件分支／子图／人审／取消／分池／派活／进度，41 条是同日为「验收机判／回退边／运行期改图／跨图子图」四件补的）。 测试只往临时目录写：`/graph-test-/<用例名>`，默认 `GRAPH_TEST_TMP` = `AIProject/ai1net-dsh-server/tmp`。每条用例只**新建**自己的目录、跑完不删，整轮收尾由 `tearDownModule` 清一次；这样一次跑下来新建几十个文件、不触发本机沙箱的「批量删除」守卫（那个守卫一次删满 50 个文件就会拦，曾让 20 条用例报 ERROR，而用例本身零失败）。想换落点就设环境变量 `GRAPH_TEST_TMP`。 @@ -302,7 +392,7 @@ rm -rf "E:/ProgramData/AIProject/ai1net-dsh-server/tmp"/graph-test-* -十九个用例类各管一摊： +二十三个用例类各管一摊： 1、`T1_Init` —— 建图：`--kpi` 按条数生成节点（全角分号也认）、`--nodes-json` 带依赖与域、重复 id 报错、自动编号避让、`max_parallel`。 2、`T2_CheckErrors` —— 硬错三条：环（含自环）、依赖指向不存在的节点、状态非法。 @@ -323,15 +413,18 @@ 17、`T17_Pools` —— 分池：同域只放 1 个、`--pool` 落进 `meta.pools`、上限 0 被拒。 18、`T18_Dispatch` —— 派活：`tick` 按并发上限开、`--dry-run` 不动状态、等依赖、`--limit`；`plan` 四条硬规则形态。 19、`T19_Progress` —— 进度：`percent` 与 `by_status`、关键路径跳过已完成、全完成 100%。 +20、`T20_Accepts` —— 验收机判：缺产物／缺关键词／证据太短／产物数量不足都拒收且报缺什么；反斜杠等价；纯文本条目不拦；`blocked` 不受影响；人审节点同样先过验收；`check` 与 `init` 双处拦非法语法。 +21、`T21_Rework` —— 回退边：复位目标与全部下游、`attempts`／`reworks`／`rework_from` 落位、写 `rework` 事件、旧产物挪 `prev_*`、目标必须是上游、不能是自己、没配 `rework_to` 报错、人审驳回自动回退、已展开节点收回子节点后重展开。 +22、`T22_Mutate` —— 运行期改图：`add-node` 字段与事件、重复 id／悬空依赖报错、`remove-edge` 删完即可派、边不存在报错、四进程并发加节点不丢更新（`_rev` = 4）、运行中加节点可被 `ready` 选中。 +23、`T23_SubgraphRef` —— 跨图子图：从另一张图展开、内部依赖加前缀、父等全部子节点收敛、子节点失败父 `blocked`、缺图／自引用／跨图环／`nodes` 与 `gid` 互斥／被引用图里嵌 `map` 全报错、定义搬过来但按新图重跑。 测试对 `graphctl.py` 的唯一要求是「能当命令行跑」，不 import 它的内部函数 —— 所以改内部实现不会连带改测试。 ## 六、还没做的（下一步） -- 判定节点完成：现在靠回报 `--state done`，还没有「看证据够不够」的自动判定。 -- 回退边：`attempts` 与重试已在记，但还没有「判不过回到某个上游节点」的字段（现在的回退只有"自己重试"）。 -- 展开源变化：`map.from` 只在上游 `done` 之后读一次，上游重跑（回退）不会重新展开。 +- 展开源变化：`map.from` 只在上游 `done` 之后读一次，上游重跑（回退）不会自动重新展开（`rework` 会把已展开节点收回去重建，但那是显式动作）。 - 排期落库：`plan` 只出规范行，写进宿主排期表由上层完成。 -- 跨图子图：现在子图只能内联，还不能引用另一张图。 +- 删节点：现在只有 `add-node` / `remove-edge`；删节点要先 `cancel`，再从图里摘结构这一步没做。 +- 跨图引用不带版本：引用的是"那张图当前的定义"，被引用图改了不会回溯已展开的子节点（要按新定义跑得 `rework`）。 这几件都只在本目录内加，不影响现有机制。 diff --git a/session-mechanism/graph/graphctl.py b/session-mechanism/graph/graphctl.py index 113026c..299f5bb 100644 --- a/session-mechanism/graph/graphctl.py +++ b/session-mechanism/graph/graphctl.py @@ -44,6 +44,9 @@ ready --ws --gid 就绪集（含为什么没被选）；顺带回收超时租约、展开 map start --ws --gid --node N1 [--by ...] [--lease-sec S] report --ws --gid --node N1 --state done|blocked [--evidence "..."] [--artifact P ...] + rework --ws --gid --node N1 [--to 上游] [--reason ...] 回退到上游重做 + add-node --ws --gid --node N9 [--deps A,B] [--domain P] [--accepts C] + remove-edge --ws --gid --from A --to B 删掉「B 依赖 A」这条边 expand --ws --gid --node P [--force] 手动触发展开 show --ws --gid list --ws @@ -66,6 +69,52 @@ 分池：`meta.pools: {"域前缀": 并发上限}` —— 同域并发受限，与 `max_parallel` 叠加。 进度：`progress` 给出 by_status／percent／关键路径（未完成节点的最长依赖链）。 +## 验收判据（accepts）怎么机判（2026-10-19 补） + +`report --state done` 时**先按 `accepts` 校验产物与证据**，不满足直接拒收（rc≠0，节点 +不置 done，并在 stderr 逐条列出缺什么）。条目两种写法，**纯文本＝描述性判据、不参与机判**： + + "artifact:docs/*.md" 产物里至少一条匹配该通配（反斜杠与 `/` 等价） + "artifact:>=2" 产物至少 2 条 + "evidence:测试通过" 证据里必须出现「测试通过」 + "evidence:全绿,零回归" 多个关键词用 `,` 分隔 ⇒ 全部都要出现（AND） + "evidence:>=20" 证据至少 20 个字符 + {"kind": "artifact", "glob": "..."} / {"kind": "artifact", "min": 2} + {"kind": "evidence", "contains": ["a", "b"]} / {"kind": "evidence", "min": 20} + +校验读的是**本次 ＋ 历史**的合并视图（`--artifact` 是追加语义）。 + +## 回退边（rework_to，2026-10-19 补） + +节点写 `rework_to: "上游节点"`（或列表，取第一个）。判不过时用 `rework --node X`： + + ① 目标必须是 X 的**上游**（传递依赖），⛔ 不能是自己或下游； + ② 目标及其**全部传递下游**复位 `pending`，清租约／完成时刻／驳回原因； + ③ 复位时把 `evidence` / `artifacts` 挪到 `prev_evidence` / `prev_artifacts` 再清空 + ⇒ 重做之后必须重新交产物（否则验收闸会被旧产物蒙过）； + ④ 目标 `attempts +1`、全链 `reworks +1`，写一条 `rework` 事件。 + ⑤ `approve --reject` 时若节点带 `rework_to`，自动走同一套回退（人审判不过＝判不过）。 + ⑥ 链上有已展开（`map`／子图）的节点 ⇒ **收回它的子节点并清 `expanded`**， + 下次 `ready`／`tick` 按当时的输入重新展开（⛔ 不留半截子节点）。 + +## 运行期改图（2026-10-19 补） + + add-node --node N9 --deps A,B --domain docs/x/ --accepts "artifact:docs/x/a.md" + remove-edge --from A --to B 删掉「B 依赖 A」 +两条都走 `_update`（目录锁 ＋ 唯一临时名 ＋ 原子替换 ＋ `_rev` 回读）⇒ 与并发的 +`start`／`report` 互不覆盖；边不存在、节点重复、依赖悬空 ⇒ **报错**，⛔ 不静默空操作。 + +## 跨图子图（2026-10-19 补） + +`subgraph: {"gid": "<同工作区另一张图的 gid>"}` —— 除内联 `{"nodes": [...]}` 外， +可引用 `/.workbuddy/graph//graph.json`，把它的**节点定义**展开成本图的 +`<父>/<子id>`，父节点照旧 `gating`、等子节点全部收敛。 + ① 只搬定义（title／deps／domain／accepts／deps_rule／on_dep_fail／when／retry／ + lease_sec／approval／rework_to），⛔ 不搬运行状态（引用即重跑一遍）； + ② 子图内部依赖与被引用图的 `when`／`rework_to` 引用都加父前缀； + ③ ⛔ 被引用的图里不允许再有 `map`／`subgraph`（不嵌套），`check` 报硬错； + ④ 自引用、A→B→A 成环 ⇒ `check` 与展开时都报错。 + ## 就绪的判据 ① 依赖按 `deps_rule` 满足（默认 all_success） @@ -94,6 +143,7 @@ from __future__ import annotations import argparse +import fnmatch import io import json import math @@ -283,6 +333,285 @@ def _out(obj) -> None: sys.stdout.buffer.write(data.encode("utf-8")) +# ── 验收判据（accepts 机判）────────────────────────────────────────────── +def _accepts_problem(vals): + """校验 accepts 语法；有问题返回一句人读的原因，没问题返回 None。""" + for raw in (vals or []): + if isinstance(raw, dict): + kind = str(raw.get("kind") or "").strip() + if kind not in ("artifact", "evidence"): + return 'accepts 的对象条目要写 kind=artifact 或 evidence（实际 %r）' % (raw,) + if kind == "artifact": + if raw.get("glob") in (None, "") and raw.get("min") in (None, ""): + return 'accepts 的 artifact 条目要写 glob 或 min：%r' % (raw,) + if raw.get("min") not in (None, ""): + try: + if int(raw["min"]) <= 0: + return "accepts 的 min 必须 ≥1：%r" % (raw,) + except (TypeError, ValueError): + return "accepts 的 min 不是整数：%r" % (raw,) + else: + cs = raw.get("contains") + if cs is not None and not isinstance(cs, list): + return 'accepts 的 evidence.contains 要写成列表：%r' % (raw,) + if not cs and raw.get("min") in (None, ""): + return 'accepts 的 evidence 条目要写 contains 或 min：%r' % (raw,) + if raw.get("min") not in (None, ""): + try: + if int(raw["min"]) <= 0: + return "accepts 的 min 必须 ≥1：%r" % (raw,) + except (TypeError, ValueError): + return "accepts 的 min 不是整数：%r" % (raw,) + continue + if not isinstance(raw, str): + return "accepts 条目只认字符串或对象（实际 %r）" % (raw,) + s = raw.strip() + low = s.lower() + if not (low.startswith("artifact:") or low.startswith("evidence:")): + continue # 纯文本 ＝ 描述性判据，不机判 + head, _, arg = s.partition(":") + arg = arg.strip() + if not arg: + return "accepts 条目缺参数：%r（写成 %s:<通配|>=N|关键词>）" % (s, head) + if arg.startswith(">="): + try: + if int(arg[2:]) <= 0: + return "accepts 的 >=N 要 ≥1：%r" % s + except ValueError: + return "accepts 的 >=N 不是整数：%r" % s + elif low.startswith("evidence:") and not any( + x.strip() for x in arg.replace("，", ",").split(",")): + return "accepts 的 evidence 关键词为空：%r" % s + return None + + +def _norm_accepts(vals) -> list: + p = _accepts_problem(vals) + if p: + raise SystemExit(p) + out = [] + for raw in (vals or []): + if isinstance(raw, dict): + kind = str(raw.get("kind")).strip() + it = {"kind": kind} + if kind == "artifact": + if raw.get("glob") not in (None, ""): + it["glob"] = str(raw["glob"]).strip() + if raw.get("min") not in (None, ""): + it["min"] = int(raw["min"]) + else: + if raw.get("contains"): + it["contains"] = [str(x) for x in raw["contains"]] + if raw.get("min") not in (None, ""): + it["min"] = int(raw["min"]) + out.append(it) + else: + out.append(str(raw).strip()) + return out + + +def _norm_path(s) -> str: + return str(s or "").replace("\\", "/").strip() + + +def _accepts_missing(accepts, artifacts, evidence) -> list: + """逐条判 accepts，返回没达标的描述（空列表＝全过）。""" + arts = [_norm_path(x) for x in (artifacts or [])] + ev = str(evidence or "") + miss = [] + for raw in (accepts or []): + if isinstance(raw, dict): + kind = str(raw.get("kind") or "") + glob = raw.get("glob") + mn = raw.get("min") + need = [str(x) for x in (raw.get("contains") or [])] + else: + s = str(raw).strip() + head, _, arg = s.partition(":") + low = head.strip().lower() + arg = arg.strip() + if low not in ("artifact", "evidence"): + continue # 描述性判据，不机判 + kind = low + glob = None + mn = None + need = [] + if arg.startswith(">="): + mn = int(arg[2:]) + elif kind == "artifact": + glob = arg + else: + need = [x.strip() for x in arg.replace("，", ",").split(",") if x.strip()] + if kind == "artifact": + if glob: + if not any(fnmatch.fnmatch(a, _norm_path(glob)) for a in arts): + miss.append("缺产物 artifact:%s（现有产物：%s）" + % (glob, "、".join(arts) if arts else "无")) + if mn is not None and len(arts) < mn: + miss.append("产物数量不足 artifact:>=%d（现有 %d 条）" % (mn, len(arts))) + elif kind == "evidence": + if mn is not None and len(ev.strip()) < mn: + miss.append("证据太短 evidence:>=%d（现有 %d 字）" % (mn, len(ev.strip()))) + if need: + absent = [k for k in need if k not in ev] + if absent: + miss.append("证据缺关键词 evidence:%s（现缺：%s）" + % (",".join(need), "、".join(absent))) + return miss + + +# ── 回退边（rework_to）────────────────────────────────────────────────── +def _ancestors(nodes: dict, nid: str) -> set: + seen, stack = set(), list(_deps(nodes.get(nid) or {})) + while stack: + x = stack.pop() + if x in seen or x not in nodes: + continue + seen.add(x) + stack.extend(_deps(nodes[x])) + return seen + + +def _descendants(nodes: dict, nid: str) -> set: + """所有传递下游（不含自己）。""" + kids = {} + for k, n in nodes.items(): + for d in _deps(n): + kids.setdefault(d, []).append(k) + seen, stack = set(), list(kids.get(nid) or []) + while stack: + x = stack.pop() + if x in seen or x not in nodes: + continue + seen.add(x) + stack.extend(kids.get(x) or []) + return seen + + +def _rework_target(n: dict): + """节点自己的 rework_to（字符串或列表）⇒ 第一个目标 id（可能为空）。""" + rt = n.get("rework_to") + if isinstance(rt, (list, tuple)): + for x in rt: + if str(x).strip(): + return str(x).strip() + return "" + return str(rt or "").strip() + + +def _drop_subtree(nodes: dict, nid: str) -> list: + """收回一个已展开节点的全部子节点（展开产物，重新展开时按需重建）。""" + out, stack = [], list((nodes.get(nid) or {}).get("children") or []) + while stack: + c = stack.pop() + if c not in nodes: + continue + out.append(c) + stack.extend((nodes[c] or {}).get("children") or []) + for c in out: + nodes.pop(c, None) + return out + + +def _rework_reset(nodes: dict, from_nid: str, tgt: str, reason: str = "") -> dict: + """把 tgt 及其全部传递下游复位成 pending（调用方保证已通过前置校验）。 + + 链上有已展开（map／subgraph）的节点 ⇒ **收回它的子节点、清掉 expanded**， + 下次 ready／tick 会按当时的输入重新展开（⛔ 不留半截子节点）。 + """ + down = _descendants(nodes, tgt) | {tgt} + bad = [x for x in down if str((nodes.get(x) or {}).get("status")) == "running"] + if bad: + raise SystemExit("回退链上有正在跑的节点：%s（先 cancel 或等它结束）" + % ", ".join(sorted(bad))) + dropped = [] + for x in sorted(down): + m = nodes.get(x) + if m is not None and (m.get("expanded") or str(m.get("status")) == "gating"): + dropped.extend(_drop_subtree(nodes, x)) + now = _now() + reset = [] + for x in sorted(down): + m = nodes.get(x) + if m is None: # 已被当作子节点收回 + continue + if m.get("artifacts"): + m["prev_artifacts"] = list(m["artifacts"]) + m["artifacts"] = [] + if m.get("evidence"): + m["prev_evidence"] = m["evidence"] + m["evidence"] = "" + m["status"] = "pending" + for k in ("lease_at", "lease_at_ts", "lease_sec", "done_at", "awaiting_at", + "approved_at", "approved_by", "not_before", "retry_at", + "retry_exhausted", "rejected_at", "blocked_by", "blocked_reason", + "skipped_at", "skipped_reason", "completed_at", + "expanded", "expanded_at", "item_count", "children"): + m.pop(k, None) + m["reworked_at"] = now + m["rework_of"] = tgt + m["reworks"] = int(m.get("reworks") or 0) + 1 + reset.append(x) + nodes[tgt]["attempts"] = int(nodes[tgt].get("attempts") or 0) + 1 + nodes[tgt]["rework_from"] = from_nid + if reason: + nodes[tgt]["rework_reason"] = reason + res = {"target": tgt, "reset": reset, + "attempts": nodes[tgt]["attempts"], + "reworks": nodes[tgt]["reworks"]} + if dropped: + res["dropped_children"] = dropped + return res + + +# ── 跨图子图（subgraph.gid）───────────────────────────────────────────── +def _sg_chain(g: dict, nid: str) -> list: + """从自己往上走 parent 链，收集沿途声明的 subgraph.gid（防跨图成环）。""" + nodes = g.get("nodes") or {} + out, seen, cur = [], set(), nid + while cur and cur in nodes and cur not in seen: + seen.add(cur) + sg = (nodes.get(cur) or {}).get("subgraph") + if isinstance(sg, dict) and str(sg.get("gid") or "").strip(): + out.append(str(sg["gid"]).strip()) + cur = (nodes.get(cur) or {}).get("parent") + return out + + +def _sg_specs_from_gid(ws: str, ref_gid: str, chain: list) -> list: + """读另一张图，把它的节点定义翻成内联子图规格。""" + if not ws: + raise SystemExit("内部错误：展开跨图子图时缺工作区路径（gid=%s）" % ref_gid) + if ref_gid in chain: + raise SystemExit("子图引用成环：%s" % " → ".join(list(chain) + [ref_gid])) + p = _gdir(ws, ref_gid) / "graph.json" + if not p.is_file(): + raise SystemExit("子图引用的图不存在：%s（gid=%s）" % (p, ref_gid)) + sg = json.loads(p.read_text(encoding="utf-8")) + specs = [] + for sid, sn in (sg.get("nodes") or {}).items(): + if not isinstance(sn, dict): + continue + if sn.get("map") or sn.get("subgraph"): + raise SystemExit("子图 %s 的节点 %s 自己又声明了 map／subgraph ⇒ 不支持嵌套" + % (ref_gid, sid)) + specs.append({ + "id": sid, + "title": sn.get("title") or sid, + "deps": list(sn.get("deps") or []), + "domain": _norm_dom(sn.get("domain")), + "accepts": list(sn.get("accepts") or []), + "deps_rule": sn.get("deps_rule"), + "on_dep_fail": sn.get("on_dep_fail"), + "when": sn.get("when"), + "retry": sn.get("retry"), + "lease_sec": sn.get("lease_sec"), + "approval": sn.get("approval"), + "rework_to": sn.get("rework_to"), + }) + return specs + + # ── 域与依赖 ──────────────────────────────────────────────────────────── def _norm_dom(d) -> list: if isinstance(d, str): @@ -467,6 +796,45 @@ def _find_cycle(nodes: dict) -> list: return [] +def _ref_graph(ws: str, gid: str): + p = _gdir(ws, gid) / "graph.json" + if not p.is_file(): + return None + try: + return json.loads(p.read_text(encoding="utf-8")) + except Exception: # noqa: BLE001 + return None + + +def _check_sg_ref(ws: str, chain: list, nid: str, ref: str, errs: list, + depth: int = 0) -> list: + """校验 `subgraph.gid`：存在性／自引用／跨图成环／不许嵌套；返回子图节点规格。""" + if ref in chain: + errs.append("%s 的 subgraph 引用成环：%s" % (nid, " → ".join(chain + [ref]))) + return [] + if depth > 6: + errs.append("%s 的 subgraph 引用链超过 6 层（gid=%s）" % (nid, ref)) + return [] + rg = _ref_graph(ws, ref) + if rg is None: + errs.append("%s 的 subgraph 引用的图不存在：gid=%s（%s）" + % (nid, ref, _gdir(ws, ref) / "graph.json")) + return [] + rnodes = {k: v for k, v in (rg.get("nodes") or {}).items() if isinstance(v, dict)} + out = [] + for sid, sn in rnodes.items(): + if sn.get("map") or sn.get("subgraph"): + errs.append("%s 引用的图 %s 里节点 %s 又声明了 map／subgraph ⇒ 不支持嵌套" + % (nid, ref, sid)) + out.append({"id": sid, "deps": [str(x) for x in (sn.get("deps") or [])]}) + for sid, sn in rnodes.items(): + sub2 = sn.get("subgraph") + g2 = str((sub2 or {}).get("gid") or "").strip() if isinstance(sub2, dict) else "" + if g2: + _check_sg_ref(ws, chain + [ref], "%s/%s" % (nid, sid), g2, errs, depth + 1) + return out + + def cmd_check(a) -> int: g = _load(a.ws, a.gid) nodes = g.get("nodes") or {} @@ -499,9 +867,20 @@ def cmd_check(a) -> int: sg = n.get("subgraph") if sg is not None: kids = sg.get("nodes") if isinstance(sg, dict) else None - if not isinstance(kids, list): - errs.append('%s 的 subgraph 要写 {"nodes": [...]}' % nid) + ref = str((sg or {}).get("gid") or "").strip() if isinstance(sg, dict) else "" + if isinstance(kids, list) and ref: + errs.append("%s 的 subgraph 同时给了 nodes 与 gid（只能选一种）" % nid) + if not isinstance(kids, list) and not ref: + errs.append('%s 的 subgraph 要写 {"nodes": [...]} 或 {"gid": "..."}' % nid) kids = [] + if ref: + kids = _check_sg_ref(a.ws, [a.gid], nid, ref, errs) + extra = [s["id"] for s in kids] + for s in kids: + for d in s["deps"]: + if d not in extra: + errs.append("%s 引用的图 %s 里 %s 依赖 %s 不在那张图里" + % (nid, ref, s["id"], d)) seen = set() for s in kids: sid = str((s or {}).get("id") or "") @@ -537,6 +916,18 @@ def cmd_check(a) -> int: "保守做法是让它单独跑" % nid) if str(n.get("status")) not in STATUSES: errs.append("%s 的状态非法：%r" % (nid, n.get("status"))) + ap = _accepts_problem(n.get("accepts")) + if ap: + errs.append("%s：%s" % (nid, ap)) + rt = _rework_target(n) + if rt: + if rt not in nodes: + errs.append("%s 的 rework_to 指向不存在的节点：%s" % (nid, rt)) + elif rt == nid: + errs.append("%s 的 rework_to 指向自己" % nid) + elif rt not in _ancestors(nodes, nid): + errs.append("%s 的 rework_to %s 不是它的上游（回退只能回到上游）" + % (nid, rt)) if not (n.get("accepts") or []) and not n.get("map") and not n.get("parent"): warns.append("%s 没有验收判据 ⇒ 判不出完成" % nid) @@ -820,7 +1211,7 @@ def _apply_conditions(g: dict, acts: list) -> int: # ── 子图（节点内部再拆子流程）─────────────────────────────────────────── -def _expand_subgraphs(g: dict, acts: list) -> int: +def _expand_subgraphs(g: dict, acts: list, ws: str = "") -> int: nodes = g.get("nodes") or {} changed = 0 for nid, n in list(nodes.items()): @@ -828,16 +1219,28 @@ def _expand_subgraphs(g: dict, acts: list) -> int: continue if not _deps_ok(nodes, n)[0]: continue - acts.append(_expand_subgraph(g, nodes, nid)) + acts.append(_expand_subgraph(g, nodes, nid, ws)) changed += 1 return changed -def _expand_subgraph(g: dict, nodes: dict, nid: str) -> dict: - """内联子图 ⇒ 子节点 `<父>/<子id>`，父节点转 gating，子节点全成功才 done。""" +def _expand_subgraph(g: dict, nodes: dict, nid: str, ws: str = "") -> dict: + """子图 ⇒ 子节点 `<父>/<子id>`，父节点转 gating，子节点全成功才 done。 + + `subgraph.nodes` 内联；`subgraph.gid` 引用同工作区另一张图的节点定义。 + """ n = nodes[nid] sub = n.get("subgraph") or {} - specs = [s for s in (sub.get("nodes") or []) if isinstance(s, dict)] + ref_gid = str(sub.get("gid") or "").strip() + if ref_gid: + if ref_gid == str(g.get("gid") or ""): + raise SystemExit("子图引用了当前图自己：%s" % ref_gid) + par = n.get("parent") + chain = list(reversed(_sg_chain(g, par))) if par else [] + specs = _sg_specs_from_gid(ws, ref_gid, chain) + n["subgraph_ref"] = ref_gid + else: + specs = [s for s in (sub.get("nodes") or []) if isinstance(s, dict)] prefix = nid + "/" local, kids = {}, [] for i, s in enumerate(specs, 1): @@ -868,19 +1271,30 @@ def _expand_subgraph(g: dict, nodes: dict, nid: str) -> dict: cd = sub.get("child_domain") dom = _norm_dom(str(cd).replace("{id}", cid[len(prefix):])) if cd else list(base) kid = {"title": str(s.get("title") or cid), "deps": deps, "domain": dom, - "accepts": [str(x) for x in (s.get("accepts") or [])], + "accepts": _norm_accepts(s.get("accepts") or []), "status": "pending", "evidence": "", "artifacts": [], "attempts": 0, "parent": nid, "sub": True} - for k in ("deps_rule", "on_dep_fail", "approval", "lease_sec", "when"): + for k in ("deps_rule", "on_dep_fail", "approval", "lease_sec"): if s.get(k) not in (None, "", []): kid[k] = s[k] + # `when` / `rework_to` 里的节点引用要跟着加父前缀，否则指向不存在的 id + if s.get("when"): + kid["when"] = [{"node": local.get(str(c.get("node")), str(c.get("node"))), + "status": str(c.get("status") or "done")} + for c in s["when"] if isinstance(c, dict)] + rt = _rework_target(s) + if rt: + kid["rework_to"] = local.get(rt, rt) if s.get("retry"): kid["retry"] = _retry_cfg({"meta": {}}, {"retry": s["retry"]}) + if ref_gid: + kid["origin_graph"] = ref_gid nodes[cid] = kid n["status"] = "gating" n["children"] = [c for c, _ in kids] return {"node": nid, "event": "expand", "kind": "subgraph", - "count": len(kids), "children": n["children"], "to": "gating"} + "count": len(kids), "children": n["children"], "to": "gating", + **({"from_graph": ref_gid} if ref_gid else {})} # ── 按域分池 ──────────────────────────────────────────────────────────── @@ -919,7 +1333,7 @@ def _pool_full(g: dict, n: dict, occupied: list) -> dict: return {} -def _advance(g: dict) -> list: +def _advance(g: dict, ws: str = "") -> list: """把「超时回收／子节点收敛／展开／条件判定」推到不动点；返回动了哪些（供写事件）。""" acts: list = [] for _ in range(6): @@ -927,7 +1341,7 @@ def _advance(g: dict) -> list: moved += _reclaim_leases(g, acts) moved += _complete_parents(g, acts) moved += _expand_maps(g, acts) - moved += _expand_subgraphs(g, acts) + moved += _expand_subgraphs(g, acts, ws) moved += _apply_conditions(g, acts) if not moved: break @@ -972,7 +1386,7 @@ def _advance_graph(ws: str, gid: str, g: dict): return g, acts def _mut(gg): - acts.extend(_advance(gg)) + acts.extend(_advance(gg, ws)) g2 = _update(ws, gid, _mut) for x in acts: @@ -1120,6 +1534,14 @@ def cmd_report(a) -> int: n["evidence"] = a.evidence for art in (a.artifact or []): n.setdefault("artifacts", []).append(str(art)) + if a.state == "done": # F1：报 done 先过验收判据 + gaps = _accepts_missing(n.get("accepts"), n.get("artifacts"), n.get("evidence")) + if gaps: + raise SystemExit("节点 %s 验收不通过（不置 done）：\n - %s\n" + "补齐产物／证据后重新 report（accepts 不可绕过）" + % (a.node, "\n - ".join(gaps))) + n["accepts_checked_at"] = _now() + n["accepts_gaps"] = [] for k in ("lease_at", "lease_at_ts", "lease_sec"): n.pop(k, None) if a.state == "blocked": @@ -1209,15 +1631,34 @@ def cmd_expand(a) -> int: # ── 人审 / 取消 / 派活 / 排期 / 进度 ───────────────────────────────────── def cmd_approve(a) -> int: - """人审节点：`awaiting` ⇒ done（默认）或 blocked（--reject）。""" + """人审节点：`awaiting` ⇒ done（默认）或 blocked（--reject）。 + + 驳回时若节点带 `rework_to` ⇒ 直接走回退（人审判不过 ＝ 判不过）。 + """ + rw = {} + def _mut(g): - n = (g.get("nodes") or {}).get(a.node) + nodes = g.get("nodes") or {} + n = nodes.get(a.node) if not n: raise SystemExit("没有这个节点：%s" % a.node) st = str(n.get("status")) if st != "awaiting": raise SystemExit("节点 %s 不在等人审（当前 %s）" % (a.node, st)) if a.reject: + tgt = _rework_target(n) + if tgt: + if tgt not in nodes: + raise SystemExit("回退目标不存在：%s" % tgt) + if tgt not in _ancestors(nodes, a.node): + raise SystemExit("回退目标 %s 不是 %s 的上游" % (tgt, a.node)) + n["status"] = "blocked" + n["blocked_by"] = "rejected" + n["blocked_reason"] = a.reason or "人审驳回" + n["rejected_at"] = _now() + rw.update(_rework_reset(nodes, a.node, tgt, a.reason or "人审驳回")) + rw["auto"] = True + return n["status"] = "blocked" n["blocked_by"] = "rejected" n["blocked_reason"] = a.reason or "人审驳回" @@ -1231,11 +1672,12 @@ def cmd_approve(a) -> int: g = _update(a.ws, a.gid, _mut) st = _st(g.get("nodes") or {}, a.node) _event(a.ws, a.gid, {"node": a.node, "to": st, "by": a.by or "", - "reject": bool(a.reject), "reason": a.reason or ""}) + "reject": bool(a.reject), "reason": a.reason or "", + **({"rework": rw} if rw else {})}) g, _acts = _advance_graph(a.ws, a.gid, g) r = _ready(g) _out({"ok": True, "node": a.node, "status": _st(g.get("nodes") or {}, a.node), - "next_ready": r["ready"]}) + "next_ready": r["ready"], "rework": rw or None}) return 0 @@ -1276,6 +1718,120 @@ def cmd_cancel(a) -> int: return 0 +def cmd_rework(a) -> int: + """回退到上游重做：目标及其全部传递下游复位 pending，记 attempts／reworks 与事件。""" + res = {} + + def _mut(g): + nodes = g.get("nodes") or {} + n = nodes.get(a.node) + if not n: + raise SystemExit("没有这个节点：%s" % a.node) + tgt = str(a.to or "").strip() or _rework_target(n) + if not tgt: + raise SystemExit("节点 %s 没有 rework_to，也没给 --to" % a.node) + if tgt not in nodes: + raise SystemExit("回退目标不存在：%s" % tgt) + if tgt == a.node: + raise SystemExit("回退目标不能是自己（要重做自己直接 start 该节点）") + if tgt not in _ancestors(nodes, a.node): + raise SystemExit("回退目标 %s 不是 %s 的上游（回退只能回到上游）" + % (tgt, a.node)) + res.update(_rework_reset(nodes, a.node, tgt, a.reason or "")) + res["node"] = a.node + + g = _update(a.ws, a.gid, _mut) + _event(a.ws, a.gid, {"event": "rework", "node": a.node, "to": res.get("target"), + "reset": res.get("reset") or [], "reason": a.reason or "", + "by": a.by or "", "attempts": res.get("attempts"), + "reworks": res.get("reworks")}) + g, acts = _advance_graph(a.ws, a.gid, g) + r = _ready(g) + _out({"ok": True, "node": a.node, "target": res.get("target"), + "reset": res.get("reset") or [], + "dropped_children": res.get("dropped_children") or [], + "attempts": res.get("attempts"), + "reworks": res.get("reworks"), + "skipped_by_rework": [x["node"] for x in acts if x.get("event") == "skip"], + "next_ready": r["ready"]}) + return 0 + + +def cmd_add_node(a) -> int: + """运行期加节点（与并发的 start／report 共用同一把目录锁，不会互相覆盖）。""" + nid = str(a.node or "").strip() + if not nid: + raise SystemExit("--node 不能为空") + deps = [x.strip() for x in str(a.deps or "").replace("，", ",").split(",") if x.strip()] + rule = str(a.deps_rule or "all_success") + if rule not in DEP_RULES: + raise SystemExit("--deps-rule 只认 %s" % " / ".join(DEP_RULES)) + accepts = _norm_accepts(a.accepts or []) + rework_to = str(a.rework_to or "").strip() + + def _mut(g): + nodes = g.get("nodes") or {} + if nid in nodes: + raise SystemExit("节点已存在：%s（换一个 id 或先 remove-edge）" % nid) + miss = [d for d in deps if d not in nodes] + if miss: + raise SystemExit("依赖不存在：%s" % ", ".join(miss)) + if rework_to and rework_to not in nodes: + raise SystemExit("rework_to 指向的节点不存在：%s" % rework_to) + if rework_to == nid: + raise SystemExit("rework_to 不能指向自己") + nodes[nid] = { + "title": str(a.title or nid), "deps": deps, + "domain": _norm_dom(a.domain or []), "accepts": accepts, + "status": "pending", "evidence": "", "artifacts": [], "attempts": 0, + "deps_rule": rule, "added_at": _now(), + } + if rework_to: + nodes[nid]["rework_to"] = rework_to + if a.approval: + nodes[nid]["approval"] = True + g["nodes"] = nodes + + g = _update(a.ws, a.gid, _mut) + _event(a.ws, a.gid, {"event": "add_node", "node": nid, "deps": deps, + "by": a.by or "", "rev": g.get("_rev")}) + g, _acts = _advance_graph(a.ws, a.gid, g) + r = _ready(g) + _out({"ok": True, "node": nid, "deps": deps, + "status": _st(g.get("nodes") or {}, nid), + "rev": g.get("_rev"), "next_ready": r["ready"]}) + return 0 + + +def cmd_remove_edge(a) -> int: + """删边：删掉「--to 依赖 --from」这一条依赖（边不存在即报错，⛔ 不静默空操作）。""" + src, dst = str(a.src or "").strip(), str(a.to or "").strip() + if not src or not dst: + raise SystemExit("要给 --from <上游> 和 --to <下游>") + + def _mut(g): + nodes = g.get("nodes") or {} + for x in (src, dst): + if x not in nodes: + raise SystemExit("没有这个节点：%s" % x) + d = _deps(nodes[dst]) + if src not in d: + raise SystemExit("边不存在：%s → %s（%s 的 deps=%s 里没有 %s）" + % (src, dst, dst, d, src)) + nodes[dst]["deps"] = [x for x in d if x != src] + nodes[dst]["edge_removed_at"] = _now() + + g = _update(a.ws, a.gid, _mut) + _event(a.ws, a.gid, {"event": "remove_edge", "from": src, "to": dst, + "by": a.by or "", "rev": g.get("_rev")}) + g, _acts = _advance_graph(a.ws, a.gid, g) + r = _ready(g) + _out({"ok": True, "from": src, "to": dst, "deps": _deps( + (g.get("nodes") or {}).get(dst) or {}), "rev": g.get("_rev"), + "next_ready": r["ready"]}) + return 0 + + def cmd_tick(a) -> int: """自动派活：推进到不动点 → 算就绪集 → 直接把就绪节点开成 running。""" started, missed = [], [] @@ -1467,7 +2023,7 @@ def cmd_init(a) -> int: "title": str(n.get("title") or nid), "deps": [str(x) for x in (n.get("deps") or [])], "domain": _norm_dom(n.get("domain")), - "accepts": [str(x) for x in (n.get("accepts") or [])], + "accepts": _norm_accepts(n.get("accepts") or []), "status": "pending", "evidence": "", "artifacts": [], "attempts": 0, "deps_rule": rule, } @@ -1485,14 +2041,33 @@ def cmd_init(a) -> int: nodes[nid]["map"] = _norm_map(n["map"]) if n.get("subgraph"): sg = n["subgraph"] - if not isinstance(sg, dict) or not isinstance(sg.get("nodes"), list): - raise SystemExit('%s 的 subgraph 要写 {"nodes": [...]}' % nid) + if not isinstance(sg, dict): + raise SystemExit('%s 的 subgraph 要写 {"nodes": [...]} 或 {"gid": "另一张图"}' + % nid) + has_nodes = isinstance(sg.get("nodes"), list) + has_gid = bool(str(sg.get("gid") or "").strip()) + if has_nodes and has_gid: + raise SystemExit("%s 的 subgraph 只能选一种：内联 nodes 或引用 gid" % nid) + if not has_nodes and not has_gid: + raise SystemExit('%s 的 subgraph 要写 {"nodes": [...]} 或 {"gid": "..."}' + % nid) if n.get("map"): raise SystemExit("%s 同时声明了 map 与 subgraph（只能选一个）" % nid) - sub = {"nodes": [dict(s) for s in sg["nodes"] if isinstance(s, dict)]} + sub = {} + if has_nodes: + sub["nodes"] = [dict(s) for s in sg["nodes"] if isinstance(s, dict)] + else: + sub["gid"] = str(sg["gid"]).strip() if sg.get("child_domain"): sub["child_domain"] = str(sg["child_domain"]) nodes[nid]["subgraph"] = sub + if n.get("rework_to"): + rt = _rework_target({"rework_to": n["rework_to"]}) + if not rt: + raise SystemExit("%s 的 rework_to 不能为空" % nid) + if rt == nid: + raise SystemExit("%s 的 rework_to 不能指向自己" % nid) + nodes[nid]["rework_to"] = rt if n.get("when"): if not isinstance(n["when"], list): raise SystemExit("%s 的 when 必须是列表" % nid) @@ -1632,6 +2207,38 @@ def main(argv=None) -> int: p.add_argument("--by", default="") p.set_defaults(fn=cmd_report) + p = sub.add_parser("rework", help="回退到上游重做（目标及其下游复位 pending）") + p.add_argument("--ws", required=True) + p.add_argument("--gid", required=True) + p.add_argument("--node", required=True) + p.add_argument("--to", default="", help="回退目标（默认取节点自己的 rework_to）") + p.add_argument("--reason", default="") + p.add_argument("--by", default="") + p.set_defaults(fn=cmd_rework) + + p = sub.add_parser("add-node", help="运行期加节点") + p.add_argument("--ws", required=True) + p.add_argument("--gid", required=True) + p.add_argument("--node", required=True) + p.add_argument("--title", default="") + p.add_argument("--deps", default="", help="依赖节点，逗号分隔") + p.add_argument("--domain", action="append", default=[], help="域前缀，可重复") + p.add_argument("--accepts", action="append", default=[], + help="验收判据，可重复（artifact:… / evidence:… / 纯文本）") + p.add_argument("--deps-rule", default="all_success") + p.add_argument("--rework-to", default="") + p.add_argument("--approval", action="store_true") + p.add_argument("--by", default="") + p.set_defaults(fn=cmd_add_node) + + p = sub.add_parser("remove-edge", help="删边：删掉「--to 依赖 --from」") + p.add_argument("--ws", required=True) + p.add_argument("--gid", required=True) + p.add_argument("--from", dest="src", required=True, help="上游节点") + p.add_argument("--to", required=True, help="下游节点（它的 deps 里删掉上游）") + p.add_argument("--by", default="") + p.set_defaults(fn=cmd_remove_edge) + p = sub.add_parser("expand", help="手动触发展开") p.add_argument("--ws", required=True) p.add_argument("--gid", required=True) diff --git a/session-mechanism/graph/test_graphctl.py b/session-mechanism/graph/test_graphctl.py index 572c1d7..4aa9eab 100644 --- a/session-mechanism/graph/test_graphctl.py +++ b/session-mechanism/graph/test_graphctl.py @@ -2,7 +2,7 @@ # -*- coding: utf-8 -*- """test_graphctl.py —— 「图模式」自动化测试（2026-10-18）。 -## 覆盖十九类 +## 覆盖二十三类 1 建图（--kpi / --nodes-json） 7 边界（单节点 / 空节点 / gid 含中文 / 图不存在） 2 校验硬错（环 / 悬空依赖 / 状态非法） 8 独立性（只写临时目录，不碰现有机制文件） 3 校验警告（无域 / 无判据 / 域重叠） 9 并发写（两进程同时写不丢状态） @@ -13,6 +13,10 @@ 14 子图（内联展开 / 子节点收敛 / 互斥） 18 自动派活（tick / plan 四条硬规则） 15 人审（awaiting / approve / reject） 19 进度与关键路径（progress） 16 取消（连坐子节点 / --all / 下游跳过） +20 验收判据机判（artifact / evidence 两种条目、缺什么报什么、纯文本不拦、人审同样受限） +21 回退边（rework_to 上游校验 / 复位链 / attempts 与事件 / 人审驳回自动回退 / 收回已展开子节点） +22 运行期改图（add-node / remove-edge / 并发加节点不丢更新 / 运行中加节点可被派） +23 跨图子图（subgraph.gid 引用另一张图 / 内部依赖加前缀 / 父等收敛 / 缺图·自引用·跨图环·嵌套全报错） ## 跑法 python graph/test_graphctl.py -v @@ -43,6 +47,17 @@ TMP_ROOT = Path(os.environ.get("GRAPH_TEST_TMP") or _DEFAULT_TMP) / ("graph-test _ENV = dict(os.environ) _ENV["PYTHONIOENCODING"] = "utf-8" +# 🔴 2026-10-19 实测（根因）：本机沙箱的「批量删除」守卫按 **一次工具调用累计删除数** 计数 +# （`CODEBUDDY_SAFE_DELETE_BULK_STATE_DIR` + `CODEBUDDY_TOOL_CALL_ID`，阈值 +# `CODEBUDDY_SAFE_DELETE_BULK_THRESHOLD`，宿主给 50）。一轮全量用例会起上百次 +# graphctl，每次释放目录锁都要删一个 `owner`，累计到 50 条就 SystemExit(1) —— +# 于是**从第 50 次调用之后，所有用例都开始假红**（单跑某几条却全 OK）。 +# 逐个 `unlink` 也躲不掉：守卫数的是累计量，不是单批大小。 +# ⇒ 测试进程把自己的删除预算调大：这套用例只删自己建的 `graph-test-`， +# 与交互式误删保护无关；生产调用不设这个变量，守卫照旧。 +_ENV["CODEBUDDY_SAFE_DELETE_BULK_THRESHOLD"] = "1000000" +os.environ["CODEBUDDY_SAFE_DELETE_BULK_THRESHOLD"] = "1000000" + # ── 基础设施 ──────────────────────────────────────────────────────────── def _run(*args, stdin: str | None = None): @@ -71,12 +86,18 @@ def bad(*args, stdin: str | None = None): return err +def chk(*args) -> dict: + """跑 `check` 并解析结果 —— 校验失败时 rc=1 是正常返回，所以不要求 rc=0。""" + return json.loads(_run(*args)[1]) + + def _drop(p: Path, tries: int = 5) -> None: """删临时目录：**逐个文件删**，失败重试几次。 ⚠️ 2026-10-19 实测：整棵树一次 `shutil.rmtree` 会命中本机沙箱的「批量删除」守卫 - （一次删满 50 个文件即 `SAFE_DELETE_BULK_CONFIRM_REQUIRED`，进程直接 SystemExit）， - 用例全过也会被它判成失败。逐个 `unlink` 不触发该守卫。 + （`SAFE_DELETE_BULK_CONFIRM_REQUIRED` ⇒ 进程直接 SystemExit），用例全过也会被判失败。 + 逐个 `unlink` 更好读日志、也便于重试残留句柄，但**仍然计入同一个累计额度** —— + 真正的解法是模块头把 `CODEBUDDY_SAFE_DELETE_BULK_THRESHOLD` 调大（见上方说明）。 """ if not p.exists(): return @@ -1291,6 +1312,389 @@ class T19_Progress(Base): self.assertEqual(r["critical_path"], []) +# ── 20 验收判据机判（F1）─────────────────────────────────────────────── +class T20_Accepts(Base): + def test_missing_artifact_refuses_done(self): + self.init({"nodes": [{"id": "A", "domain": ["a/"], + "accepts": ["artifact:out/a.md"]}]}) + err = bad("report", "--ws", self.ws, "--gid", self.gid, + "--node", "A", "--state", "done") + self.assertIn("验收不通过", err) + self.assertIn("out/a.md", err) + self.assertEqual(self.load()["nodes"]["A"]["status"], "pending") + + def test_artifact_present_passes(self): + self.init({"nodes": [{"id": "A", "domain": ["a/"], + "accepts": ["artifact:out/*.md"]}]}) + ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "A", + "--state", "done", "--artifact", "out/a.md") + n = self.load()["nodes"]["A"] + self.assertEqual(n["status"], "done") + self.assertEqual(n["accepts_gaps"], []) + + def test_backslash_artifact_matches_glob(self): + self.init({"nodes": [{"id": "A", "domain": ["a/"], + "accepts": ["artifact:out/*.md"]}]}) + ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "A", + "--state", "done", "--artifact", "out\\a.md") + self.assertEqual(self.load()["nodes"]["A"]["status"], "done") + + def test_evidence_keywords_all_required(self): + self.init({"nodes": [{"id": "A", "domain": ["a/"], + "accepts": ["evidence:全绿,零回归"]}]}) + err = bad("report", "--ws", self.ws, "--gid", self.gid, "--node", "A", + "--state", "done", "--evidence", "全绿") + self.assertIn("零回归", err) + ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "A", + "--state", "done", "--evidence", "全绿，零回归") + self.assertEqual(self.load()["nodes"]["A"]["status"], "done") + + def test_evidence_min_length(self): + self.init({"nodes": [{"id": "A", "domain": ["a/"], "accepts": ["evidence:>=10"]}]}) + err = bad("report", "--ws", self.ws, "--gid", self.gid, "--node", "A", + "--state", "done", "--evidence", "短") + self.assertIn("证据太短", err) + ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "A", + "--state", "done", "--evidence", "0123456789") + self.assertEqual(self.load()["nodes"]["A"]["status"], "done") + + def test_artifact_min_count(self): + self.init({"nodes": [{"id": "A", "domain": ["a/"], "accepts": ["artifact:>=2"]}]}) + err = bad("report", "--ws", self.ws, "--gid", self.gid, "--node", "A", + "--state", "done", "--artifact", "a.md") + self.assertIn("数量不足", err) + ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "A", + "--state", "done", "--artifact", "a.md", "--artifact", "b.md") + self.assertEqual(self.load()["nodes"]["A"]["status"], "done") + + def test_object_form(self): + self.init({"nodes": [{"id": "A", "domain": ["a/"], "accepts": [ + {"kind": "artifact", "glob": "out/*.md"}, + {"kind": "evidence", "contains": ["跑过"]}]}]}) + err = bad("report", "--ws", self.ws, "--gid", self.gid, "--node", "A", + "--state", "done", "--evidence", "跑过") + self.assertIn("out/*.md", err) + ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "A", + "--state", "done", "--evidence", "跑过", "--artifact", "out/x.md") + self.assertEqual(self.load()["nodes"]["A"]["status"], "done") + + def test_plain_text_accepts_not_enforced(self): + self.init({"nodes": [{"id": "A", "domain": ["a/"], "accepts": ["产物在位"]}]}) + ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "A", "--state", "done") + self.assertEqual(self.load()["nodes"]["A"]["status"], "done") + + def test_blocked_ignores_accepts(self): + self.init({"nodes": [{"id": "A", "domain": ["a/"], + "accepts": ["artifact:out/a.md"]}]}) + ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "A", + "--state", "blocked", "--evidence", "没做出来") + self.assertEqual(self.load()["nodes"]["A"]["status"], "blocked") + + def test_approval_still_gated_by_accepts(self): + self.init({"nodes": [{"id": "A", "domain": ["a/"], "approval": True, + "accepts": ["artifact:out/a.md"]}]}) + bad("report", "--ws", self.ws, "--gid", self.gid, "--node", "A", "--state", "done") + ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "A", + "--state", "done", "--artifact", "out/a.md") + self.assertEqual(self.load()["nodes"]["A"]["status"], "awaiting") + + def test_check_rejects_bad_grammar(self): + self.init({"nodes": [{"id": "A", "domain": ["a/"], "accepts": ["产物在位"]}]}) + g = self.load() # 手改图（模拟早期写坏的图） + g["nodes"]["A"]["accepts"] = ["artifact:"] + self.dump(g) + r = chk("check", "--ws", self.ws, "--gid", self.gid) + self.assertFalse(r["ok"]) + self.assertTrue(any("accepts" in e for e in r["errors"])) + + def test_init_rejects_bad_accepts(self): + err = bad("init", "--ws", self.ws, "--title", "t", "--gid", "g1", + "--nodes-json", "-", + stdin='{"nodes": [{"id": "A", "accepts": ["evidence:"]}]}') + self.assertIn("accepts", err) + + def test_init_rejects_bad_object(self): + err = bad("init", "--ws", self.ws, "--title", "t", "--gid", "g1", + "--nodes-json", "-", + stdin='{"nodes": [{"id": "A", "accepts": [{"kind": "nope"}]}]}') + self.assertIn("kind=artifact", err) + + +# ── 21 回退边（F2）────────────────────────────────────────────────────── +class T21_Rework(Base): + def _chain(self, **kw): + b = {"id": "B", "domain": ["b/"], "accepts": ["y"], "deps": ["A"]} + b.update(kw) + self.init({"nodes": [{"id": "A", "domain": ["a/"], "accepts": ["x"]}, b]}) + self.state("A", "done") + + def test_resets_target_and_downstream(self): + self._chain(rework_to="A") + r = ok("rework", "--ws", self.ws, "--gid", self.gid, "--node", "B") + self.assertEqual(r["target"], "A") + self.assertEqual(sorted(r["reset"]), ["A", "B"]) + g = self.load()["nodes"] + self.assertEqual(g["A"]["status"], "pending") + self.assertEqual(g["B"]["status"], "pending") + + def test_attempts_and_reworks_recorded(self): + self._chain(rework_to="A") + r = ok("rework", "--ws", self.ws, "--gid", self.gid, "--node", "B") + self.assertEqual(r["attempts"], 1) + g = self.load()["nodes"] + self.assertEqual(g["A"]["attempts"], 1) + self.assertEqual(g["A"]["reworks"], 1) + self.assertEqual(g["A"]["rework_from"], "B") + + def test_event_written(self): + self._chain(rework_to="A") + ok("rework", "--ws", self.ws, "--gid", self.gid, "--node", "B", + "--reason", "产物不合格") + ev = [e for e in self.events() if e.get("event") == "rework"] + self.assertEqual(len(ev), 1) + self.assertEqual(ev[0]["to"], "A") + self.assertEqual(ev[0]["reason"], "产物不合格") + + def test_evidence_and_artifacts_moved(self): + self._chain(rework_to="A") + ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "B", + "--state", "blocked", "--evidence", "不行", "--artifact", "b.md") + ok("rework", "--ws", self.ws, "--gid", self.gid, "--node", "B") + g = self.load()["nodes"] + self.assertEqual(g["A"]["evidence"], "") + self.assertEqual(g["B"]["prev_evidence"], "不行") + self.assertEqual(g["B"]["artifacts"], []) + self.assertEqual(g["B"]["prev_artifacts"], ["b.md"]) + + def test_target_must_be_upstream(self): + self.init({"nodes": [{"id": "A", "domain": ["a/"], "accepts": ["x"]}, + {"id": "B", "domain": ["b/"], "accepts": ["y"], + "deps": ["A"]}]}) + err = bad("rework", "--ws", self.ws, "--gid", self.gid, "--node", "A", + "--to", "B") + self.assertIn("上游", err) + + def test_target_cannot_be_self(self): + self._chain(rework_to="A") + err = bad("rework", "--ws", self.ws, "--gid", self.gid, "--node", "B", + "--to", "B") + self.assertIn("自己", err) + + def test_no_target_given(self): + self._chain() + err = bad("rework", "--ws", self.ws, "--gid", self.gid, "--node", "B") + self.assertIn("rework_to", err) + + def test_approve_reject_auto_reworks(self): + self._chain(rework_to="A", approval=True) + ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "B", "--state", "done") + self.assertEqual(self.load()["nodes"]["B"]["status"], "awaiting") + r = ok("approve", "--ws", self.ws, "--gid", self.gid, "--node", "B", "--reject") + self.assertTrue(r["rework"]["auto"]) + g = self.load()["nodes"] + self.assertEqual(g["A"]["status"], "pending") + self.assertEqual(g["B"]["status"], "pending") + + def test_check_flags_non_upstream_rework_to(self): + self.init({"nodes": [{"id": "A", "domain": ["a/"], "accepts": ["x"], + "rework_to": "B"}, + {"id": "B", "domain": ["b/"], "accepts": ["y"], + "deps": ["A"]}]}) + r = chk("check", "--ws", self.ws, "--gid", self.gid) + self.assertFalse(r["ok"]) + self.assertTrue(any("rework_to" in e for e in r["errors"])) + + def test_rework_expanded_node_drops_children(self): + ok("init", "--ws", self.ws, "--title", "子", "--gid", "gsub", "--nodes-json", "-", + stdin='{"nodes": [{"id": "S1", "domain": ["s/"], "accepts": ["s"]}]}') + self.init({"nodes": [{"id": "P", "domain": ["p/"], "accepts": ["p"], + "subgraph": {"gid": "gsub"}}, + {"id": "Q", "domain": ["q/"], "accepts": ["q"], + "deps": ["P"], "rework_to": "P"}]}) + ok("tick", "--ws", self.ws, "--gid", self.gid) + ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "P/S1", + "--state", "done", "--artifact", "s/x.md") + r = ok("rework", "--ws", self.ws, "--gid", self.gid, "--node", "Q") + self.assertEqual(r["dropped_children"], ["P/S1"]) # 旧子节点被收回 + g = self.load()["nodes"] + self.assertEqual(g["P"]["reworks"], 1) + self.assertEqual(g["P"]["rework_from"], "Q") + # 收回后按当时输入**重新展开** ⇒ 新子节点是干净的一轮（不带旧产物） + self.assertEqual(g["P"]["status"], "gating") + self.assertEqual(g["P/S1"]["artifacts"], []) + self.assertEqual(g["P/S1"]["attempts"], 0) + + +# ── 22 运行期改图（F3）────────────────────────────────────────────────── +class T22_Mutate(Base): + def test_add_node_fields(self): + self.init({"nodes": [{"id": "A", "domain": ["a/"], "accepts": ["x"]}]}) + r = ok("add-node", "--ws", self.ws, "--gid", self.gid, "--node", "B", + "--title", "新增", "--deps", "A", "--domain", "b/", + "--accepts", "artifact:b/*.md") + self.assertEqual(r["status"], "pending") + n = self.load()["nodes"]["B"] + self.assertEqual(n["deps"], ["A"]) + self.assertEqual(n["domain"], ["b/"]) + self.assertEqual(n["accepts"], ["artifact:b/*.md"]) + self.assertEqual(n["attempts"], 0) + + def test_add_node_duplicate_rejected(self): + self.init({"nodes": [{"id": "A", "domain": ["a/"], "accepts": ["x"]}]}) + err = bad("add-node", "--ws", self.ws, "--gid", self.gid, "--node", "A") + self.assertIn("已存在", err) + + def test_add_node_dangling_dep_rejected(self): + self.init({"nodes": [{"id": "A", "domain": ["a/"], "accepts": ["x"]}]}) + err = bad("add-node", "--ws", self.ws, "--gid", self.gid, "--node", "B", + "--deps", "ZZ") + self.assertIn("依赖不存在", err) + + def test_add_node_writes_event(self): + self.init({"nodes": [{"id": "A", "domain": ["a/"], "accepts": ["x"]}]}) + ok("add-node", "--ws", self.ws, "--gid", self.gid, "--node", "B") + ev = [e for e in self.events() if e.get("event") == "add_node"] + self.assertEqual(len(ev), 1) + self.assertEqual(ev[0]["node"], "B") + + def test_remove_edge_frees_downstream(self): + self.init({"nodes": [{"id": "A", "domain": ["a/"], "accepts": ["x"]}, + {"id": "B", "domain": ["b/"], "accepts": ["y"], + "deps": ["A"]}]}) + r = ok("remove-edge", "--ws", self.ws, "--gid", self.gid, + "--from", "A", "--to", "B") + self.assertEqual(r["deps"], []) + self.assertIn("B", r["next_ready"]) + + def test_remove_missing_edge_rejected(self): + self.init({"nodes": [{"id": "A", "domain": ["a/"], "accepts": ["x"]}, + {"id": "B", "domain": ["b/"], "accepts": ["y"]}]}) + err = bad("remove-edge", "--ws", self.ws, "--gid", self.gid, + "--from", "A", "--to", "B") + self.assertIn("边不存在", err) + + def test_concurrent_add_node_no_lost_update(self): + self.init({"nodes": [{"id": "A", "domain": ["a/"], "accepts": ["x"]}]}) + procs = [] + for i in range(4): + procs.append(subprocess.Popen( + [PY, str(GRAPHCTL), "add-node", "--ws", str(self.ws), "--gid", self.gid, + "--node", "C%d" % i, "--domain", "c%d/" % i], + stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, env=_ENV)) + for p in procs: + _o, e = p.communicate() + self.assertEqual(p.returncode, 0, e.decode("utf-8", "replace")) + g = self.load() + for i in range(4): + self.assertIn("C%d" % i, g["nodes"]) + self.assertEqual(int(g.get("_rev") or 0), 4) + + def test_add_node_during_run_is_pickable(self): + self.init({"nodes": [{"id": "A", "domain": ["a/"], "accepts": ["x"]}]}) + ok("start", "--ws", self.ws, "--gid", self.gid, "--node", "A") + ok("add-node", "--ws", self.ws, "--gid", self.gid, "--node", "B", + "--domain", "b/") + r = ok("ready", "--ws", self.ws, "--gid", self.gid) + self.assertIn("B", r["ready"]) + self.assertEqual(self.load()["nodes"]["A"]["status"], "running") + + +# ── 23 跨图子图（F4）─────────────────────────────────────────────────── +class T23_SubgraphRef(Base): + SUB = {"nodes": [{"id": "S1", "domain": ["s/"], "accepts": ["s1"]}, + {"id": "S2", "domain": ["t/"], "accepts": ["s2"], + "deps": ["S1"]}]} + + def _both(self, sub=None): + ok("init", "--ws", self.ws, "--title", "子图", "--gid", "gsub", "--nodes-json", "-", + stdin=json.dumps(sub or self.SUB, ensure_ascii=False)) + self.init({"nodes": [{"id": "P", "domain": ["p/"], "accepts": ["p"], + "subgraph": {"gid": "gsub"}}, + {"id": "Q", "domain": ["q/"], "accepts": ["q"], + "deps": ["P"]}]}) + + def test_check_ok(self): + self._both() + r = ok("check", "--ws", self.ws, "--gid", self.gid) + self.assertTrue(r["ok"], r["errors"]) + + def test_expands_from_other_graph(self): + self._both() + ok("tick", "--ws", self.ws, "--gid", self.gid) + g = self.load()["nodes"] + self.assertEqual(g["P"]["status"], "gating") + self.assertEqual(sorted(g["P"]["children"]), ["P/S1", "P/S2"]) + self.assertEqual(g["P/S1"]["origin_graph"], "gsub") + self.assertEqual(g["P/S1"]["deps"], []) # 父节点无外依赖 + self.assertEqual(g["P/S2"]["deps"], ["P/S1"]) # 内部依赖加父前缀 + + def test_parent_waits_for_all_children(self): + self._both() + ok("tick", "--ws", self.ws, "--gid", self.gid) + ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "P/S1", + "--state", "done") + self.assertEqual(self.load()["nodes"]["P"]["status"], "gating") + ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "P/S2", + "--state", "done") + g = self.load()["nodes"] + self.assertEqual(g["P"]["status"], "done") + self.assertIn("Q", ok("ready", "--ws", self.ws, "--gid", self.gid)["ready"]) + + def test_child_failure_blocks_parent(self): + self._both() + ok("tick", "--ws", self.ws, "--gid", self.gid) + ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "P/S1", + "--state", "blocked", "--evidence", "失败") + ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "P/S2", + "--state", "blocked", "--evidence", "也没成") + self.assertEqual(self.load()["nodes"]["P"]["status"], "blocked") + + def test_missing_ref_graph(self): + self.init({"nodes": [{"id": "P", "domain": ["p/"], "accepts": ["p"], + "subgraph": {"gid": "nope"}}]}) + r = chk("check", "--ws", self.ws, "--gid", self.gid) + self.assertFalse(r["ok"]) + self.assertTrue(any("不存在" in e for e in r["errors"])) + + def test_self_reference(self): + self.init({"nodes": [{"id": "P", "domain": ["p/"], "accepts": ["p"], + "subgraph": {"gid": self.gid}}]}) + r = chk("check", "--ws", self.ws, "--gid", self.gid) + self.assertFalse(r["ok"]) + self.assertTrue(any("成环" in e for e in r["errors"])) + + def test_cross_graph_cycle(self): + ok("init", "--ws", self.ws, "--title", "b", "--gid", "gb", "--nodes-json", "-", + stdin='{"nodes": [{"id": "M1", "domain": ["m/"], "subgraph": {"gid": "ga"}}]}') + ok("init", "--ws", self.ws, "--title", "a", "--gid", "ga", "--nodes-json", "-", + stdin='{"nodes": [{"id": "M2", "domain": ["n/"], "subgraph": {"gid": "gb"}}]}') + r = chk("check", "--ws", self.ws, "--gid", "ga") + self.assertFalse(r["ok"]) + self.assertTrue(any("成环" in e for e in r["errors"])) + + def test_nodes_and_gid_exclusive(self): + err = bad("init", "--ws", self.ws, "--title", "t", "--gid", "g1", + "--nodes-json", "-", + stdin='{"nodes": [{"id": "P", "subgraph": ' + '{"nodes": [{"id": "S1"}], "gid": "gsub"}}]}') + self.assertIn("只能选一种", err) + + def test_nested_map_in_ref_graph_rejected(self): + self._both({"nodes": [{"id": "S1", "domain": ["s/"], + "map": {"over": ["a"]}}]}) + r = chk("check", "--ws", self.ws, "--gid", self.gid) + self.assertFalse(r["ok"]) + self.assertTrue(any("不支持嵌套" in e for e in r["errors"])) + + def test_ref_definitions_copied_not_status(self): + sub = {"nodes": [{"id": "S1", "domain": ["s/"], "accepts": ["artifact:s/x.md"]}]} + self._both(sub) + ok("tick", "--ws", self.ws, "--gid", self.gid) + kid = self.load()["nodes"]["P/S1"] + self.assertEqual(kid["accepts"], ["artifact:s/x.md"]) + self.assertEqual(kid["status"], "running") # 定义搬过来、按新图重跑 + + def tearDownModule(): _drop(TMP_ROOT)