diff --git a/session-mechanism/graph/README.md b/session-mechanism/graph/README.md index ee49a15..a94e6a9 100644 --- a/session-mechanism/graph/README.md +++ b/session-mechanism/graph/README.md @@ -1,6 +1,7 @@ # 图模式(独立功能) -> 2026-10-18 新增。**这一套与现有「项目机制」完全并列**:不读写 `goal.json`、`tasks.json`、`collabd-state.json`,不参与六道闸,不接管常驻循环,不改变现有任何会话的行为。 +> 2026-10-18 新增;2026-10-19 补「租约超时 / 重试与触发规则 / 动态展开」三件批量能力。 +> **这一套与现有「项目机制」完全并列**:不读写 `goal.json`、`tasks.json`、`collabd-state.json`,不参与六道闸,不接管常驻循环,不改变现有任何会话的行为。 > 停用方式:不调用它就行;它没有常驻进程、没有注册任何钩子、没有改任何现有文件。 ## 一、它解决什么 @@ -52,22 +53,131 @@ ### 4、开一个节点、回报 python graph/graphctl.py start --ws "" --gid "" --node A --by "执行会话-xxx" - python graph/graphctl.py report --ws "" --gid "" --node A --state done --evidence "页面 1440 宽截图在 ui/shot.png" + python graph/graphctl.py report --ws "" --gid "" --node A --state done \ + --evidence "页面 1440 宽截图在 ui/shot.png" --artifact ui/shot.png -回报 `done` 后会立刻返回下一批的 `ready`,这就是图在往前走的信号。 +`start` 会给节点记租约(见 §三之二);`report` 可带多个 `--artifact`(产物清单,供展开节点读取,见 §三之四)。 + +回报 `done` 后会立刻返回下一批的 `ready`,这就是图在往前走的信号。回报 `blocked` 时若节点配了重试策略,会自动排下一次(见 §三之三)。 ### 5、看状态 python graph/graphctl.py list --ws "" python graph/graphctl.py show --ws "" --gid "" +### 6、手动展开 + + python graph/graphctl.py expand --ws "" --gid "" --node P [--force] + +一般不用手动调:`ready` 命中条件会自动展开。手动 `expand` 用于提前铺开、或排查"为什么没展开"。 + ## 三、就绪的四个条件 -1. 依赖全部 `done`。 -2. 节点自己是 `pending`。 +1. 依赖按 `deps_rule` 满足(默认 `all_success`)。 +2. 节点自己是 `pending`,且不在重试等待窗内(`not_before` 未到就先等)。 3. 域不与任何 `running` 节点重叠(`docs/` 与 `docs/ui/` 算重叠)。 4. 加上在跑的节点后,不超过 `max_parallel`(默认 3)。 +## 三之二、节点租约与超时 + +解决的问题:一个节点开工后卡死、没人回报,会一直占着并发名额,这一批就永远推不动。 + +**start 时记租约**:`start` 会给节点写 `lease_at`(ISO)、`lease_at_ts`(epoch)、`starts`(开工次数)。租约期限由 `--lease-sec`(全局)或节点自己的 `lease_sec` 决定,节点级优先。 + +**ready 时收超时**:`ready` 会先把超过租约的 `running` 节点按策略处置,再算就绪集。所以在拿"这一批派谁"的同时就把僵死节点清掉了,不需要额外的守护进程。 + +字段写法(建图时给全局): + + python graph/graphctl.py init --ws "" --title "…" \ + --lease-sec 3600 --on-timeout requeue --nodes-json - <<'JSON' + {"nodes": [ + {"id": "A", "accepts": ["…"], "lease_sec": 120}, + {"id": "B", "accepts": ["…"], "on_timeout": "block"} + ]} + JSON + +| 字段 | 默认值 | 含义 | +|---|---|---| +| `meta.lease.timeout_sec` / `--lease-sec` | `3600` | 多少秒没回报算超时;`0` = 不判超时 | +| `meta.lease.on_timeout` / `--on-timeout` | `requeue` | 超时怎么处置 | +| 节点 `lease_sec` | 跟随全局 | 这个节点单独的租约秒数 | +| 节点 `on_timeout` | 跟随全局 | 这个节点单独的处置策略 | + +三种策略: + +1. `requeue`(默认)—— 回 `pending`,`attempts+1`,清掉租约,名额立刻还回来。 +2. `block` —— 进 `blocked`,`blocked_by=lease_timeout`,`attempts+1`,等人工处理。 +3. `keep` —— 只提示不动;`ready` 的输出里会多一条 `lease_notices`。 + +**注意两点**:没有 `lease_at` 的 `running` 节点(手改出来的旧图)不判超时,保持兼容;回收动作会写一条 `lease_timeout` 事件,`attempts` 会涨,所以"自动回收"和"真失败"在计数上是合并的——看 `blocked_by` 与事件区分。 + +## 三之三、失败策略与依赖触发规则 + +### 1、节点级重试 + +`report --state blocked` 之后怎么办,由节点自己的 `retry` 决定: + + {"id": "A", "accepts": ["…"], + "retry": {"max": 2, "delay_sec": 30, "backoff": "exponential", "factor": 2}} + +| 字段 | 默认值 | 含义 | +|---|---|---| +| `retry.max` | `0` | 最多自动重试几次;`0` = 不自动重试(`blocked` 立刻停住) | +| `retry.delay_sec` | `0` | 基础间隔秒 | +| `retry.backoff` | `fixed` | `fixed` 固定/`linear` 线性递增/`exponential` 指数 | +| `retry.factor` | `2` | 指数退避的倍数 | + +语义:`attempts` 是失败次数。`attempts <= max` 时回 `pending` 并写 `not_before`(下次可开工时刻,`retry_at` 是同值的可读形式);超过就进 `blocked` 并标 `retry_exhausted=true`,不再自动重试。间隔算法:`fixed` = `delay_sec`,`linear` = `delay_sec × attempts`,`exponential` = `delay_sec × factor^(attempts-1)`。 + +等待窗内 `ready` 不会派它,`waiting` 里会写「重试等待中:还需 N 秒」。 + +全局默认也能在 init 上给:`--retry-max` / `--retry-delay` / `--retry-backoff` / `--retry-factor`;节点字段覆盖全局。 + +### 2、依赖触发规则(`deps_rule`) + +| 值 | 放行条件 | +|---|---| +| `all_success`(默认) | 上游**全部 done** | +| `all_done` | 上游**全部跑完**(done 或 blocked 都算),不看成败 | +| `one_success` | 上游**任一 done** 即可,不等其余的 | + +写法:节点上写 `"deps_rule": "all_done"`,或建图时统一给 `--deps-rule all_done`。非法值两处都会拦:`init` 直接报错,手改 JSON 的由 `check` 报硬错。 + +## 三之四、动态展开(map / foreach) + +一个节点声明成"按 N 项展开",`ready` 命中时生成 `P#1 … P#N` 子节点,父节点转 `gating`;子节点全部 `done` 后父节点自动转 `done`,`evidence` 里写「展开 N 项全部完成」,子节点的 `artifacts` 会汇总到父节点上(方便再上一层继续展开)。 + +### 1、建图时给列表 + + {"id": "P", "title": "逐页改文案", "accepts": ["每页都改完"], + "map": {"over": ["a.md", "b.md", "c.md"], "child_domain": "out/{i}/"}} + +### 2、运行时从上游产物 / 证据读 + + {"id": "U", "title": "扫描", "accepts": ["清单在"]}, + {"id": "P", "title": "逐个处理", "accepts": ["都处理完"], "deps": ["U"], + "map": {"from": {"node": "U", "pick": "artifacts"}}} + +| 字段 | 默认值 | 含义 | +|---|---|---| +| `map.over` | — | 建图时直接给的列表(元素可以是字符串或对象) | +| `map.from.node` | — | 运行时读哪个上游节点 | +| `map.from.pick` | `artifacts` | 读什么:`artifacts`(产物清单)/`evidence`(按行或分号切,JSON 数组也认)/`children`(上游展开出来的子节点) | +| `map.child_domain` | 空(不继承) | 子节点域的模板,`{i}` 序号、`{item}` 该项,可给字符串或列表 | +| `map.max_items` | 不限制 | 最多展开几项(防爆炸) | + +产物清单要在回报时给:`report … --state done --artifact a.md --artifact b.md`。 + +三个必须知道的取舍: + +1. **子节点默认不继承父域**。同一批展开通常写同一批文件,继承父域会让域比较把它们互相挡死,N 项只能一个个跑;要分片写盘就写 `child_domain` 模板(`out/{i}/`),域不重叠就能并行。 +2. **N=0 立即完成**:父节点直接 `done`、不生成子节点,下游可以马上走;空清单不算失败。 +3. **N 很大由 `max_parallel` 兜住**:10 项、cap=2 时 `ready` 只放 2 个,其余进 `backlog`,跑完一个补一个。 + +父节点不能 `start`、也不能 `report`(它是容器);要提前手动展开用 `expand` 命令: + + python graph/graphctl.py expand --ws "" --gid "" --node P [--force] + ## 四、与现有机制的关系 | 事项 | 现有机制 | 图模式 | @@ -85,17 +195,20 @@ python graph/test_graphctl.py -v -全绿应看到 `Ran 52 tests ... OK`,用时 15 秒上下(每条用例起一次子进程)。 +全绿应看到 `Ran 92 tests ... OK`,用时 70 秒上下(每条用例起若干次子进程;52 条是原有基线,40 条是 2026-10-19 为三件批量能力补的)。 测试只往临时目录写:`/graph-test-/<用例名>`,默认 `GRAPH_TEST_TMP` = `AIProject/ai1net-dsh-server/tmp`。每条用例只**新建**自己的目录、跑完不删,整轮收尾由 `tearDownModule` 清一次;这样一次跑下来新建几十个文件、不触发本机沙箱的「批量删除」守卫(那个守卫一次删满 50 个文件就会拦,曾让 20 条用例报 ERROR,而用例本身零失败)。想换落点就设环境变量 `GRAPH_TEST_TMP`。 ## 五之二、并发写是怎么保证的 -两个进程可以同时对同一张图动手(比如两个节点同时开工),做法是三层: +两个进程可以同时对同一张图动手(比如两个节点同时开工),做法是四层: -1. 临时文件名带进程号与随机数,两个进程写字不会撞同一个名字(早期版本用固定名 `graph.json.tmp`,并发时直接报 `PermissionError`)。 -2. 每次写都先读最新磁盘状态、在它上面重放本次变更,再原子替换——避免「后写的整份覆盖前写的」导致节点状态静默消失。 -3. 图里带 `_rev` 版本号:写完回读核对,发现自己不是最新一版就重读重做,连续三次仍被覆盖才报错退出(要的是「宁可报错,也不静默丢状态」)。 +1. **写串行化**:每次读-改-写先进 `/.lock`(`os.mkdir` 原子建目录,⛔ 不用 OS 文件锁,本机会卡死),写完释放;锁目录超过 30 秒没动静认定持锁进程已死、直接接管,等 10 秒仍拿不到则报错退出(带手工清理提示)。 +2. 临时文件名带进程号与随机数,两个进程写字不会撞同一个名字(早期版本用固定名 `graph.json.tmp`,并发时直接报 `PermissionError`)。 +3. 每次写都在最新磁盘状态上重放本次变更,再原子替换——避免「后写的整份覆盖前写的」导致节点状态静默消失。 +4. 图里带 `_rev` 与 `_token`:写完回读核对,对不上就重读重做,连续三次仍被覆盖才报错退出(「宁可报错,也不静默丢状态」)。 + +为什么多加第 1 层:2026-10-19 实测只靠第 4 层仍会丢更新 —— A 写完回读确认后、B 再整份覆盖,B 回读看到的也是自己的 token,两边都"成功",但 A 的节点状态没了(表现为并发用例 `pending` != `running`,四轮抖一轮)。加上目录锁后连跑三轮 92 条全绿。 这条路径由 `T9_Concurrency` 的三个用例守着:并发开工两个节点、并发回报两个节点、版本号每次写都前进。 @@ -103,7 +216,7 @@ rm -rf "E:/ProgramData/AIProject/ai1net-dsh-server/tmp"/graph-test-* -八个用例类各管一摊: +十二个用例类各管一摊: 1、`T1_Init` —— 建图:`--kpi` 按条数生成节点(全角分号也认)、`--nodes-json` 带依赖与域、重复 id 报错、自动编号避让、`max_parallel`。 2、`T2_CheckErrors` —— 硬错三条:环(含自环)、依赖指向不存在的节点、状态非法。 @@ -113,6 +226,10 @@ 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` 跑完字节不变,且不新建这类文件,写入范围只有 `/.workbuddy/graph/`。 +9、`T9_Concurrency` —— 并发写:两进程同时开工/同时回报都不许丢状态,`_rev` 每次写都前进。 +10、`T10_Lease` —— 租约:start 记租约、超时回收(回 pending / block / keep 三种策略)、回收后名额还给别人、节点级 `lease_sec` 覆盖全局、`--lease-sec 0` 关闭、手改的无租约 running 不误杀、回收写事件。 +11、`T11_Retry` —— 重试与触发:重排写 `not_before`、等待窗内不派、`fixed`/`linear`/`exponential` 三种间隔、超限进 blocked 且 `retry_exhausted`、默认不自动重试、`all_success`/`all_done`/`one_success` 三种触发、非法 `deps_rule` 双处拦截。 +12、`T12_Map` —— 展开:`over` 列表、N=0 立即完成、从 evidence/artifacts 读 N 项、子节点全完成收敛父节点、N=10 但 cap=2 只放 2 个、`child_domain` 模板、默认不继承域、父节点不许 start/report、`expand` 命令与 `--force`、子节点 attempts 独立。 测试对 `graphctl.py` 的唯一要求是「能当命令行跑」,不 import 它的内部函数 —— 所以改内部实现不会连带改测试。 @@ -120,6 +237,7 @@ - 自动派活:读 `ready` 后自动建排期(要复用现有排期硬规则:cwds 正斜杠、workspace_scope、next_run_at 毫秒、scheduled_at ISO)。 - 判定节点完成:现在靠回报 `--state done`,还没有「看证据够不够」的自动判定。 -- 回退边:`attempts` 已经在记,还没有「判不过回到哪个节点」的字段与规则。 +- 回退边:`attempts` 与重试已在记,但还没有「判不过回到某个上游节点」的字段(现在的回退只有"自己重试")。 +- 展开源变化:`map.from` 只在上游 `done` 之后读一次,上游重跑(回退)不会重新展开。 -这三件都只在本目录内加,不影响现有机制。 +这几件都只在本目录内加,不影响现有机制。 diff --git a/session-mechanism/graph/graphctl.py b/session-mechanism/graph/graphctl.py index c88513c..640312f 100644 --- a/session-mechanism/graph/graphctl.py +++ b/session-mechanism/graph/graphctl.py @@ -1,6 +1,6 @@ #!/usr/bin/env python3 # -*- coding: utf-8 -*- -"""graphctl.py —— 「图模式」的独立实现(2026-10-18)。 +"""graphctl.py —— 「图模式」的独立实现(2026-10-18;2026-10-19 补批量能力)。 ## 它是什么 @@ -19,49 +19,66 @@ "gid": "20261018-2230-ui改造", "title": "给三块功能各加一份文档", "created_at": "2026-10-18T22:30:00", - "meta": {"max_parallel": 3}, + "meta": {"max_parallel": 3, + "lease": {"timeout_sec": 3600, "on_timeout": "requeue"}, + "retry": {"max": 0, "delay_sec": 0, "backoff": "fixed", "factor": 2}}, "nodes": { "N1": {"title": "...", "deps": [], "domain": ["docs/ui/"], "accepts": ["产物在位"], "status": "pending", - "evidence": "", "attempts": 0} + "evidence": "", "artifacts": [], "attempts": 0, + "deps_rule": "all_success", + "retry": {"max": 2, "delay_sec": 30, "backoff": "exponential"}, + "map": {"over": ["a", "b"], "child_domain": "docs/ui/{i}/"}} } } -`status` 四态:`pending` → `running` → `done` / `blocked`。 +`status` 五态:`pending` → `running` → `done` / `blocked`; +另有 `gating`——展开出来的父节点,等子节点全部完成(父节点自动转 `done`)。 ## 命令 init --ws <工作区> --title <目标> (--kpi "判据A;判据B" | --nodes-json <文件|->) - check --ws --gid 环/悬空依赖/空域/同批域冲突 - ready --ws --gid 就绪集(含为什么没被选) - start --ws --gid --node N1 - report --ws --gid --node N1 --state done|blocked [--evidence "..."] + [--max-parallel N] [--lease-sec S] [--on-timeout requeue|block|keep] + [--retry-max N] [--retry-delay S] [--retry-backoff fixed|linear|exponential] + check --ws --gid 环/悬空依赖/空域/同批域冲突/非法规则 + ready --ws --gid 就绪集(含为什么没被选);顺带回收超时租约、展开 map + start --ws --gid --node N1 [--by ...] [--lease-sec S] + report --ws --gid --node N1 --state done|blocked [--evidence "..."] [--artifact P ...] + expand --ws --gid --node P [--force] 手动触发展开 show --ws --gid list --ws -## 就绪的判据(四条全中) +## 就绪的判据 - ① 依赖全部 done ② 自己还是 pending + ① 依赖按 `deps_rule` 满足(默认 all_success) + ② 自己还是 pending,且不在重试等待窗内(`not_before` 未到 ⇒ 等) ③ 域不与任何 running 节点重叠,也不与本批已选节点重叠 ④ 不超 max_parallel(running 占名额) -## 状态的走法 +## 三件批量能力(2026-10-19 补) - pending --start--> running --report done--> done - |--report blocked--> blocked --start--> running(重试,attempts 已累加) - -`start` 只拒 `running`(已在跑)与 `done`(已完成);`blocked` 可重开。 + 租约:start 记 `lease_at` / `lease_at_ts`;ready 时超过 `lease.timeout_sec` + 的 running 节点按 `on_timeout` 处置 —— requeue(默认,回 pending 且 + attempts+1)/block(进 blocked)/keep(只提示不动)。 + 重试:`retry.max` 给了才自动重试,report blocked 后 attempts≤max ⇒ 回 pending + 并写 `not_before`(按 backoff 算),超过 ⇒ blocked 且 `retry_exhausted`。 + 默认 `max=0` ⇒ 与旧行为一致(report blocked 直接 blocked)。 + 展开:`map.over`(建图给列表)或 `map.from`(运行时读上游 artifacts / evidence), + ready 命中即生成子节点 `P#1..P#N`,父节点转 `gating`,子节点全 done ⇒ 父 done。 ## 域的用法 `domain` 是一组路径前缀(相对工作区根)。两节点域重叠 ⇒ 不能同批跑。 判定:任一前缀互为前缀即视为重叠(`docs/` 与 `docs/ui/` 算重叠)。 +展开出的子节点**默认不继承父域**(同一批展开通常写同一批文件,继承会把并发挡死); +需要分片写盘时用 `map.child_domain` 模板(`{i}` / `{item}`)。 """ from __future__ import annotations import argparse import io import json +import math import os import random import sys @@ -69,7 +86,12 @@ import time from pathlib import Path MARK = ".workbuddy/graph" -STATUSES = ("pending", "running", "done", "blocked") +LOCK_STALE_SEC = 30.0 # 锁目录超过这么久没动静 ⇒ 认定持锁进程已死,接管 +STATUSES = ("pending", "running", "done", "blocked", "gating") +DEP_RULES = ("all_success", "all_done", "one_success") +TIMEOUT_POLICIES = ("requeue", "block", "keep") +BACKOFFS = ("fixed", "linear", "exponential") +DEFAULT_LEASE_SEC = 3600.0 # ── 基础设施 ──────────────────────────────────────────────────────────── @@ -77,6 +99,26 @@ def _now() -> str: return time.strftime("%Y-%m-%dT%H:%M:%S") +def _iso(ts: float) -> str: + return time.strftime("%Y-%m-%dT%H:%M:%S", time.localtime(ts)) + + +def _parse_ts(v): + """ISO 字符串 → epoch;失败返回 None(⛔ 不抛,坏数据当作"没有租约")。""" + if v in (None, ""): + return None + try: + return float(v) + except (TypeError, ValueError): + pass + for fmt in ("%Y-%m-%dT%H:%M:%S", "%Y-%m-%d %H:%M:%S"): + try: + return time.mktime(time.strptime(str(v), fmt)) + except ValueError: + continue + return None + + def _root(ws: str) -> Path: return Path(ws).resolve() / MARK @@ -101,36 +143,110 @@ def _save(ws: str, gid: str, g: dict) -> None: os.replace(tmp, p) # 原子替换,避免半截文件 -def _update(ws: str, gid: str, mutate, retries: int = 3) -> dict: - """读-改-写,带版本号乐观重试(2026-10-19 修两处实测缺陷)。 +class _Lock: + """写串行化:`os.mkdir` 是原子的,用它当互斥量(⛔ 不用 OS 文件锁,本机会卡死)。 - ① 临时名原来固定成 `graph.json.tmp` ⇒ 两进程同时写直接撞 `PermissionError` - (桌面测试报告第 14 条实测:同时对 C、D 执行 start 就复现)。 - ② 更隐蔽的一层:就算临时名不撞,两人同时读-改-写会**丢更新** —— - 后写的整份覆盖前写的,节点状态静默消失,连报错都没有。 - ③ 对策:图里带 `_rev`,每次在**最新磁盘状态**上重放变更,写完回读核对; - 被别的进程覆盖就重读重做,最多 `retries` 次,仍失败则**报错退出**(⛔ 不静默)。 - ⚠️ 本机不用 OS 文件锁(实测会卡死),靠「唯一临时名 + 原子替换 + 回读核对」。 + ⚠️ 2026-10-19 实测:只靠 `_rev` + token 回读核对仍然丢更新 —— + A 写完回读确认后、B 再覆盖,B 回读看到的也是自己的 token,两边都"成功"。 + 那次实测里 4 轮测试抖 1 轮(`pending` != `running`)。所以写必须串行。 + ⚠️ 僵死锁:持锁进程崩了锁目录会留着 ⇒ 超过 `LOCK_STALE_SEC` 直接接管, + ⛔ 不会永久卡住;等 `wait` 秒仍拿不到则**报错退出**(带手工清理提示)。 + """ + + def __init__(self, p: Path, wait: float = 10.0): + self.p = p + self.wait = wait + self.held = False + + def __enter__(self): + t0 = time.time() + while True: + try: + os.mkdir(self.p) + self.held = True + try: + (self.p / "owner").write_text(str(os.getpid()), encoding="utf-8") + except OSError: + pass + return self + except FileExistsError: + try: + age = time.time() - self.p.stat().st_mtime + except OSError: + age = 0.0 + if age > LOCK_STALE_SEC: + for f in ("owner",): + try: + (self.p / f).unlink() + except OSError: + pass + try: + os.rmdir(self.p) + except OSError: + pass + continue + if time.time() - t0 > self.wait: + raise SystemExit( + "写锁等待超时:%s(等了 %.0f 秒;确认没有僵死进程后可直接删这个目录)" + % (self.p, self.wait)) + time.sleep(0.02) + + def __exit__(self, *exc): + if self.held: + try: + (self.p / "owner").unlink() + except OSError: + pass + try: + os.rmdir(self.p) + except OSError: + pass + return False + + +def _update(ws: str, gid: str, mutate, retries: int = 3) -> dict: + """读-改-写:**先拿目录锁再动文件**,锁内重放变更并原子替换。 + + ① 临时名带 pid + 随机数 ⇒ 两进程不会撞同一个名字(早期固定名会 `PermissionError`)。 + ② 写整体串行(`_Lock`)⇒ 不会"后写的整份覆盖前写的"导致节点状态静默消失。 + ③ `_rev` + `_token` 是第二道网:写后回读核对,对不上就重做; + 连续 `retries` 次仍被覆盖则**报错退出**(⛔ 不静默丢状态)。 + ④ Windows 上 `os.replace` 与别的进程的读并发时会短暂 `PermissionError` + ⇒ 替换动作本身小步重试,⛔ 不把它当成"并发冲突"直接失败。 """ p = _gdir(ws, gid) / "graph.json" if not p.is_file(): raise SystemExit("找不到图:%s" % p) - for _ in range(retries): - g = json.loads(p.read_text(encoding="utf-8")) - rev = int(g.get("_rev") or 0) - mutate(g) - g["_rev"] = rev + 1 - tmp = p.with_suffix(".%d.%d.json.tmp" % (os.getpid(), random.randint(0, 1 << 30))) - tmp.write_text(json.dumps(g, ensure_ascii=False, indent=2) + "\n", - encoding="utf-8", newline="\n") - os.replace(tmp, p) - try: - back = json.loads(p.read_text(encoding="utf-8")) - except Exception: # noqa: BLE001 - back = {} - if int(back.get("_rev") or 0) == rev + 1: - return back - raise SystemExit("并发冲突:%s 连续 %d 次被别的进程覆盖,稍后重试" % (p, retries)) + with _Lock(p.parent / ".lock"): + for _ in range(retries): + g = json.loads(p.read_text(encoding="utf-8")) + rev = int(g.get("_rev") or 0) + mutate(g) + token = "%d-%d" % (os.getpid(), random.randint(0, 1 << 30)) + g["_rev"] = rev + 1 + g["_token"] = token + tmp = p.with_suffix(".%d.%d.json.tmp" % (os.getpid(), random.randint(0, 1 << 30))) + tmp.write_text(json.dumps(g, ensure_ascii=False, indent=2) + "\n", + encoding="utf-8", newline="\n") + for _try in range(6): # 目标被读打开时 replace 会短暂被拒 + try: + os.replace(tmp, p) + break + except OSError: + time.sleep(0.05) + else: + try: + tmp.unlink() + except OSError: + pass + continue + try: + back = json.loads(p.read_text(encoding="utf-8")) + except Exception: # noqa: BLE001 + back = {} + if int(back.get("_rev") or 0) == rev + 1 and back.get("_token") == token: + return back + raise SystemExit("并发冲突:%s 连续 %d 次被别的进程覆盖,稍后重试" % (p, retries)) def _event(ws: str, gid: str, obj: dict) -> None: @@ -176,6 +292,79 @@ def _is_done(nodes: dict, nid: str) -> bool: return str((nodes.get(nid) or {}).get("status")) == "done" +def _deps_ok(nodes: dict, n: dict): + """按 `deps_rule` 判依赖是否放行,返回 (bool, 没满足的原因)。""" + deps = _deps(n) + if not deps: + return True, "" + rule = str(n.get("deps_rule") or "all_success") + st = {d: str((nodes.get(d) or {}).get("status")) for d in deps} + if rule == "all_done": + lack = [d for d, s in st.items() if s not in ("done", "blocked")] + return (not lack), ("依赖未跑完:%s" % ", ".join(lack)) + if rule == "one_success": + if any(s == "done" for s in st.values()): + return True, "" + return False, ("等待任一上游成功:%s" % ", ".join( + "%s(%s)" % (d, s) for d, s in st.items())) + lack = [d for d, s in st.items() if s != "done"] + return (not lack), ("依赖未完成:%s" % ", ".join("%s(%s)" % (d, st[d]) for d in lack)) + + +# ── 三件批量能力的配置读法 ────────────────────────────────────────────── +def _meta(g: dict) -> dict: + m = g.get("meta") + return m if isinstance(m, dict) else {} + + +def _lease_cfg(g: dict) -> dict: + d = _meta(g).get("lease") + d = d if isinstance(d, dict) else {} + try: + sec = float(d.get("timeout_sec", DEFAULT_LEASE_SEC)) + except (TypeError, ValueError): + sec = DEFAULT_LEASE_SEC + pol = str(d.get("on_timeout") or "requeue") + return {"timeout_sec": sec, "on_timeout": pol if pol in TIMEOUT_POLICIES else "requeue"} + + +def _node_lease_sec(g: dict, n: dict) -> float: + """节点级 `lease_sec` 优先于全局;0 表示该节点不判超时。""" + v = n.get("lease_sec") + if v in (None, ""): + return _lease_cfg(g)["timeout_sec"] + try: + return float(v) + except (TypeError, ValueError): + return _lease_cfg(g)["timeout_sec"] + + +def _retry_cfg(g: dict, n: dict) -> dict: + d = {} + for src in (_meta(g).get("retry"), n.get("retry")): + if isinstance(src, dict): + d.update(src) + def _num(k, dv): + try: + return float(d.get(k, dv)) + except (TypeError, ValueError): + return float(dv) + bk = str(d.get("backoff") or "fixed") + return {"max": max(0, int(_num("max", 0))), + "delay_sec": max(0.0, _num("delay_sec", 0)), + "backoff": bk if bk in BACKOFFS else "fixed", + "factor": max(1.0, _num("factor", 2))} + + +def _delay_of(rc: dict, attempts: int) -> float: + k = max(1, int(attempts)) + if rc["backoff"] == "linear": + return rc["delay_sec"] * k + if rc["backoff"] == "exponential": + return rc["delay_sec"] * (rc["factor"] ** (k - 1)) + return rc["delay_sec"] + + # ── 校验 ──────────────────────────────────────────────────────────────── def _find_cycle(nodes: dict) -> list: color, stack, hit = {}, [], [] @@ -215,12 +404,30 @@ def cmd_check(a) -> int: for d in _deps(n): if d not in nodes: errs.append("%s 的依赖 %s 不存在" % (nid, d)) - if not _norm_dom(n.get("domain")): + if str(n.get("deps_rule") or "all_success") not in DEP_RULES: + errs.append("%s 的 deps_rule 非法:%r(只认 %s)" + % (nid, n.get("deps_rule"), " / ".join(DEP_RULES))) + m = n.get("map") + if m is not None: + if not isinstance(m, dict) or not ("over" in m or "from" in m): + errs.append("%s 的 map 需要 over(建图给列表)或 from(运行时读上游)" % nid) + else: + src = (m.get("from") or {}).get("node") if isinstance(m.get("from"), dict) else None + if src and src not in nodes: + errs.append("%s 的 map.from.node %s 不存在" % (nid, src)) + elif src and src not in _deps(n): + warns.append("%s 的展开源 %s 不在 deps 里 ⇒ 上游没完成时会被当成" + "「展开源未就绪」空等" % (nid, src)) + if n.get("expanded"): + for cid in (n.get("children") or []): + if cid not in nodes: + errs.append("%s 的子节点 %s 不见了(图被手改过?)" % (nid, cid)) + if not _norm_dom(n.get("domain")) and not n.get("parent"): 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 []): + if not (n.get("accepts") or []) and not n.get("map") and not n.get("parent"): warns.append("%s 没有验收判据 ⇒ 判不出完成" % nid) ids = list(nodes) @@ -237,21 +444,295 @@ def cmd_check(a) -> int: return 0 if not errs else 1 +# ── 租约超时回收 ──────────────────────────────────────────────────────── +def _lease_over(g: dict, n: dict, now: float): + """超时则返回超时秒数,否则 None。没有租约时刻 ⇒ 不判超时(兼容手改的旧图)。""" + if str(n.get("status")) != "running": + return None + sec = _node_lease_sec(g, n) + if sec <= 0: + return None + ts = _parse_ts(n.get("lease_at_ts")) + if ts is None: + ts = _parse_ts(n.get("lease_at")) + if ts is None: + return None + over = now - ts - sec + return over if over > 0 else None + + +def _reclaim_leases(g: dict, acts: list) -> int: + nodes = g.get("nodes") or {} + lc = _lease_cfg(g) + now = time.time() + changed = 0 + for nid, n in list(nodes.items()): + over = _lease_over(g, n, now) + if over is None: + continue + pol = str(n.get("on_timeout") or lc["on_timeout"]) + if pol == "keep": + continue # 只提示,不动(notice 由外层补) + n["attempts"] = int(n.get("attempts") or 0) + 1 + n["reclaimed_at"] = _now() + n["reclaim_count"] = int(n.get("reclaim_count") or 0) + 1 + for k in ("lease_at", "lease_at_ts", "lease_sec"): + n.pop(k, None) + if pol == "block": + n["status"] = "blocked" + n["blocked_by"] = "lease_timeout" + else: + n["status"] = "pending" + acts.append({"node": nid, "event": "lease_timeout", "to": n["status"], + "over_sec": int(over), "attempts": n["attempts"], + "policy": pol}) + changed += 1 + return changed + + +def _lease_notices(g: dict) -> list: + lc = _lease_cfg(g) + if lc["on_timeout"] != "keep": + return [] + now = time.time() + out = [] + for nid, n in (g.get("nodes") or {}).items(): + over = _lease_over(g, n, now) + if over is not None: + out.append({"node": nid, "over_sec": int(over), + "why": "租约超时但策略为 keep(只提示,不回收)"}) + return out + + +# ── 动态展开(map / foreach)───────────────────────────────────────────── +def _as_text(x) -> str: + if isinstance(x, (dict, list)): + return json.dumps(x, ensure_ascii=False) + return str(x) + + +def _parse_list(v) -> list: + """证据/产物 → 列表。JSON 数组优先,否则按行与分号切。""" + if isinstance(v, list): + return list(v) + s = str(v or "").strip() + if not s: + return [] + if s[0] in "[(": + try: + got = json.loads(s) + if isinstance(got, list): + return list(got) + except Exception: # noqa: BLE001 + pass + out = [] + for line in s.replace("\r\n", "\n").replace(";", "\n").replace(";", "\n").split("\n"): + t = line.strip().lstrip("-*").strip() + if t: + out.append(t) + return out + + +def _map_items(g: dict, nodes: dict, nid: str): + """取展开项;源还没就绪 ⇒ None(区别于空列表 = N=0)。""" + n = nodes.get(nid) or {} + m = n.get("map") + if not isinstance(m, dict): + return None + if "over" in m: + items = list(m.get("over") or []) + else: + f = m.get("from") or {} + up = nodes.get(str(f.get("node") or "")) + if not up or str(up.get("status")) != "done": + return None + pick = str(f.get("pick") or "artifacts") + if pick == "evidence": + items = _parse_list(up.get("evidence")) + elif pick == "children": + items = list(up.get("children") or []) + else: + items = list(up.get("artifacts") or []) + cap = int(m.get("max_items") or 0) + if cap > 0: + items = items[:cap] + return items + + +def _child_domain(m: dict, i: int, item) -> list: + cd = m.get("child_domain") + if not cd: + return [] + if isinstance(cd, str): + cd = [cd] + out = [] + for x in cd: + s = str(x).replace("{i}", str(i)).replace("{item}", _as_text(item)) + out.append(s) + return _norm_dom(out) + + +def _expand_one(g: dict, nodes: dict, nid: str, items: list) -> dict: + n = nodes[nid] + if items and int((n.get("map") or {}).get("max_items") or 0) > 0: + items = items[:int((n.get("map") or {}).get("max_items"))] + n["expanded"] = True + n["expanded_at"] = _now() + n["item_count"] = len(items) + if not items: # N=0 ⇒ 无需执行,父节点直接完成 + n["status"] = "done" + n["children"] = [] + n["evidence"] = n.get("evidence") or "展开 0 项,无需执行" + return {"node": nid, "event": "expand", "count": 0, "children": [], + "to": "done"} + kids = [] + for i, it in enumerate(items, 1): + cid = "%s#%d" % (nid, i) + if cid in nodes: + raise SystemExit("子节点 id 已被占用:%s(父节点 %s)" % (cid, nid)) + title = "%s[%d]" % (str(n.get("title") or nid), i) + if not isinstance(it, (dict, list)): + title = "%s %s" % (title, str(it)[:40]) + nodes[cid] = { + "title": title, + "deps": _deps(n), + "domain": _child_domain(n.get("map") or {}, i, it), + "accepts": list(n.get("accepts") or []), + "status": "pending", "evidence": "", "artifacts": [], "attempts": 0, + "parent": nid, "index": i, "item": it, + } + kids.append(cid) + n["status"] = "gating" + n["children"] = kids + return {"node": nid, "event": "expand", "count": len(kids), "children": kids, + "to": "gating"} + + +def _expand_maps(g: dict, acts: list) -> int: + nodes = g.get("nodes") or {} + changed = 0 + for nid, n in list(nodes.items()): + st = str(n.get("status")) + if st not in ("pending", "blocked") or n.get("expanded") or not n.get("map"): + continue + if st == "blocked": + continue # 失败节点不自动展开,等人工 start/重试 + okd, _why = _deps_ok(nodes, n) + if not okd: + continue + items = _map_items(g, nodes, nid) + if items is None: + continue + acts.append(_expand_one(g, nodes, nid, items)) + changed += 1 + return changed + + +def _complete_parents(g: dict, acts: list) -> int: + nodes = g.get("nodes") or {} + changed = 0 + for nid, n in list(nodes.items()): + if str(n.get("status")) != "gating": + continue + kids = [c for c in (n.get("children") or []) if c in nodes] + if not kids or not all(_is_done(nodes, c) for c in kids): + continue + n["status"] = "done" + n["completed_at"] = _now() + arts = [] + for c in kids: + arts.extend(list((nodes.get(c) or {}).get("artifacts") or [])) + if arts: + n["artifacts"] = arts + n["evidence"] = n.get("evidence") or ("展开 %d 项全部完成" % len(kids)) + acts.append({"node": nid, "event": "children_done", "to": "done", + "count": len(kids)}) + changed += 1 + return changed + + +def _advance(g: dict) -> list: + """把「超时回收 / 子节点收敛 / 展开」推到不动点;返回动了哪些(供写事件)。""" + acts: list = [] + for _ in range(4): + moved = 0 + moved += _reclaim_leases(g, acts) + moved += _complete_parents(g, acts) + moved += _expand_maps(g, acts) + if not moved: + break + return acts + + +def _maybe_advance(g: dict) -> bool: + """便宜的预判:没有可推进的事就不写盘(保持 ready 的只读语义)。""" + nodes = g.get("nodes") or {} + lc = _lease_cfg(g) + now = time.time() + if lc["on_timeout"] != "keep" and lc["timeout_sec"] > 0: + for n in nodes.values(): + if _lease_over(g, n, now) is not None: + return True + for nid, n in nodes.items(): + st = str(n.get("status")) + if st == "gating": + kids = [c for c in (n.get("children") or []) if c in nodes] + if kids and all(_is_done(nodes, c) for c in kids): + return True + elif st == "pending" and n.get("map") and not n.get("expanded"): + if _deps_ok(nodes, n)[0] and _map_items(g, nodes, nid) is not None: + return True + return False + + +def _advance_graph(ws: str, gid: str, g: dict): + acts: list = [] + if not _maybe_advance(g): + return g, acts + + def _mut(gg): + acts.extend(_advance(gg)) + + g2 = _update(ws, gid, _mut) + for x in acts: + _event(ws, gid, x) + return g2, acts + + # ── 就绪集 ────────────────────────────────────────────────────────────── def _ready(g: dict) -> dict: nodes = g.get("nodes") or {} - cap = int((g.get("meta") or {}).get("max_parallel") or 3) + cap = int(_meta(g).get("max_parallel") or 3) running = [k for k, n in nodes.items() if str(n.get("status")) == "running"] ready, waiting = [], [] + now = time.time() for nid, n in nodes.items(): st = str(n.get("status")) + if st == "gating": + kids = [c for c in (n.get("children") or []) if c in nodes] + left = [c for c in kids if not _is_done(nodes, c)] + waiting.append({"node": nid, "why": "等待 %d 个子节点完成(还剩 %d)" + % (len(kids), len(left))}) + continue if st != "pending": continue - nd = _deps(n) - lack = [d for d in nd if not _is_done(nodes, d)] - if lack: - waiting.append({"node": nid, "why": "依赖未完成:%s" % ", ".join(lack)}) + if n.get("map") and not n.get("expanded"): + waiting.append({"node": nid, "why": "展开源未就绪(上游没完成或没有产物)"}) + continue + nb = n.get("not_before") + if nb not in (None, ""): + try: + left = float(nb) - now + except (TypeError, ValueError): + left = 0 + if left > 0: + waiting.append({"node": nid, "why": "重试等待中:还需 %d 秒" + % int(math.ceil(left))}) + continue + okd, why = _deps_ok(nodes, n) + if not okd: + waiting.append({"node": nid, "why": why}) continue dom = _norm_dom(n.get("domain")) clash = [r for r in running @@ -286,7 +767,13 @@ def _ready(g: dict) -> dict: def cmd_ready(a) -> int: g = _load(a.ws, a.gid) - _out(_ready(g)) + g, acts = _advance_graph(a.ws, a.gid, g) + r = _ready(g) + r["reclaimed"] = [x for x in acts if x.get("event") == "lease_timeout"] + r["expanded"] = [x for x in acts if x.get("event") == "expand"] + r["completed_parents"] = [x["node"] for x in acts if x.get("event") == "children_done"] + r["lease_notices"] = _lease_notices(g) + _out(r) return 0 @@ -296,38 +783,125 @@ 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")) == "running": + st = str(n.get("status")) + if st == "running": raise SystemExit("节点 %s 已在跑,不能重复开工" % a.node) - if str(n.get("status")) == "done": + if st == "done": raise SystemExit("节点 %s 已完成,不能开工" % a.node) + if st == "gating" or (n.get("map") and not n.get("expanded")): + raise SystemExit("节点 %s 是展开节点,不能直接开工(由 ready 展开后跑子节点)" + % a.node) # pending 与 blocked 都可开工:blocked 是「这轮没过」,重试时 attempts 保留 n["status"] = "running" - _update(a.ws, a.gid, _mut) + now = time.time() + n["lease_at"] = _iso(now) + n["lease_at_ts"] = now + if a.lease_sec is not None: + n["lease_sec"] = float(a.lease_sec) + n["starts"] = int(n.get("starts") or 0) + 1 + n.pop("not_before", None) + n.pop("retry_at", None) + g = _update(a.ws, a.gid, _mut) _event(a.ws, a.gid, {"node": a.node, "to": "running", "by": a.by or ""}) - _out({"ok": True, "node": a.node, "status": "running"}) + n = (g.get("nodes") or {}).get(a.node) or {} + _out({"ok": True, "node": a.node, "status": "running", + "lease_sec": _node_lease_sec(g, n)}) return 0 def cmd_report(a) -> int: if a.state not in ("done", "blocked"): raise SystemExit("--state 只认 done / blocked") + retry_info = {} def _mut(g): n = (g.get("nodes") or {}).get(a.node) if not n: raise SystemExit("没有这个节点:%s" % a.node) + if n.get("map") or str(n.get("status")) == "gating": + raise SystemExit("节点 %s 是展开父节点,本身不回报(子节点全完成自动 done)" + % a.node) if a.evidence: n["evidence"] = a.evidence - n["status"] = a.state + for art in (a.artifact or []): + n.setdefault("artifacts", []).append(str(art)) + for k in ("lease_at", "lease_at_ts", "lease_sec"): + n.pop(k, None) if a.state == "blocked": n["attempts"] = int(n.get("attempts") or 0) + 1 + n["blocked_by"] = "report" + rc = _retry_cfg(g, n) + left = rc["max"] - n["attempts"] + 1 # 还能再试几次 + if rc["max"] > 0 and n["attempts"] <= rc["max"]: + delay = _delay_of(rc, n["attempts"]) + at = time.time() + delay + n["status"] = "pending" + n["not_before"] = at + n["retry_at"] = _iso(at) + n["retry_exhausted"] = False + retry_info.update({"policy_max": rc["max"], "attempts": n["attempts"], + "requeued": True, "delay_sec": delay, + "retry_at": n["retry_at"], + "retries_left": max(0, left - 1)}) + else: + n["status"] = "blocked" + n["retry_exhausted"] = rc["max"] > 0 + n.pop("not_before", None) + n.pop("retry_at", None) + retry_info.update({"policy_max": rc["max"], "attempts": n["attempts"], + "requeued": False, "retry_at": None, + "exhausted": True}) + else: + n["status"] = "done" + n["done_at"] = _now() + n.pop("not_before", None) + n.pop("retry_at", None) g = _update(a.ws, a.gid, _mut) - _event(a.ws, a.gid, {"node": a.node, "to": a.state, - "evidence": a.evidence or "", "by": a.by or ""}) + _event(a.ws, a.gid, {"node": a.node, "to": a.state if a.state == "done" else "blocked", + "evidence": a.evidence or "", "by": a.by or "", + "artifacts": list(a.artifact or [])}) + g, acts = _advance_graph(a.ws, a.gid, g) r = _ready(g) - _out({"ok": True, "node": a.node, "status": a.state, - "next_ready": r["ready"]}) + _out({"ok": True, "node": a.node, + "status": str((g.get("nodes") or {}).get(a.node, {}).get("status")), + "next_ready": r["ready"], + "retry": retry_info or None, + "completed_parents": [x["node"] for x in acts + if x.get("event") == "children_done"]}) + return 0 + + +def cmd_expand(a) -> int: + acts = [] + + def _mut(g): + nodes = g.get("nodes") or {} + n = nodes.get(a.node) + if not n: + raise SystemExit("没有这个节点:%s" % a.node) + if not n.get("map"): + raise SystemExit("节点 %s 没有 map 声明,无可展开" % a.node) + if n.get("expanded"): + raise SystemExit("节点 %s 已展开过(children=%s)" + % (a.node, n.get("children"))) + okd, why = _deps_ok(nodes, n) + if not okd and not a.force: + raise SystemExit("展开源未就绪:%s(要强制展开加 --force)" % why) + items = _map_items(g, nodes, a.node) + if items is None: + raise SystemExit("展开源未就绪:上游节点没完成或没有产物") + acts.append(_expand_one(g, nodes, a.node, items)) + + g = _update(a.ws, a.gid, _mut) + for x in acts: + _event(a.ws, a.gid, x) + g, more = _advance_graph(a.ws, a.gid, g) + acts.extend(more) + n = (g.get("nodes") or {}).get(a.node) or {} + _out({"ok": True, "node": a.node, "status": n.get("status"), + "children": n.get("children") or [], + "count": len(n.get("children") or [])}) return 0 @@ -337,6 +911,31 @@ def _gid_of(title: str) -> str: return "%s-%s" % (time.strftime("%Y%m%d-%H%M"), safe or "graph") +def _norm_map(m) -> dict: + if not isinstance(m, dict): + raise SystemExit("map 必须是对象") + out = {} + if "over" in m: + if not isinstance(m["over"], list): + raise SystemExit("map.over 必须是列表") + out["over"] = list(m["over"]) + if "from" in m: + f = m["from"] + if not isinstance(f, dict) or not str(f.get("node") or "").strip(): + raise SystemExit('map.from 需要 {"node": "上游节点"} 形态') + out["from"] = {"node": str(f["node"]).strip(), + "pick": str(f.get("pick") or "artifacts")} + if not out: + raise SystemExit("map 需要 over(建图给列表)或 from(运行时读上游)") + if m.get("child_domain"): + out["child_domain"] = m["child_domain"] + if m.get("max_items"): + out["max_items"] = int(m["max_items"]) + if m.get("as"): + out["as"] = str(m["as"]) + return out + + def cmd_init(a) -> int: if a.nodes_json: raw = sys.stdin.read() if a.nodes_json == "-" else \ @@ -351,6 +950,11 @@ def cmd_init(a) -> int: else: raise SystemExit("要给 --kpi 或 --nodes-json") + if a.on_timeout not in TIMEOUT_POLICIES: + raise SystemExit("--on-timeout 只认 %s" % " / ".join(TIMEOUT_POLICIES)) + if a.retry_backoff not in BACKOFFS: + raise SystemExit("--retry-backoff 只认 %s" % " / ".join(BACKOFFS)) + nodes = {} for i, n in enumerate(spec.get("nodes") or []): raw = n.get("id") @@ -362,16 +966,40 @@ def cmd_init(a) -> int: nid = "N%d" % k if nid in nodes: # 显式重复 ⇒ 硬错,⛔ 不静默覆盖前一个 raise SystemExit("节点 id 重复:%s" % nid) + rule = str(n.get("deps_rule") or a.deps_rule or "all_success") + if rule not in DEP_RULES: + raise SystemExit("%s 的 deps_rule 非法:%r(只认 %s)" + % (nid, rule, " / ".join(DEP_RULES))) nodes[nid] = { "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 [])], - "status": "pending", "evidence": "", "attempts": 0, + "status": "pending", "evidence": "", "artifacts": [], "attempts": 0, + "deps_rule": rule, } + if n.get("retry"): + if not isinstance(n["retry"], dict): + raise SystemExit("%s 的 retry 必须是对象" % nid) + nodes[nid]["retry"] = _retry_cfg({"meta": {}}, {"retry": n["retry"]}) + if n.get("lease_sec") not in (None, ""): + nodes[nid]["lease_sec"] = float(n["lease_sec"]) + if n.get("on_timeout"): + if str(n["on_timeout"]) not in TIMEOUT_POLICIES: + raise SystemExit("%s 的 on_timeout 非法:%r" % (nid, n["on_timeout"])) + nodes[nid]["on_timeout"] = str(n["on_timeout"]) + if n.get("map"): + nodes[nid]["map"] = _norm_map(n["map"]) gid = a.gid or _gid_of(a.title) g = {"gid": gid, "title": a.title, "created_at": _now(), - "meta": {"max_parallel": int(spec.get("max_parallel") or a.max_parallel or 3)}, + "meta": {"max_parallel": int(spec.get("max_parallel") or a.max_parallel or 3), + "lease": {"timeout_sec": float(a.lease_sec) if a.lease_sec is not None + else DEFAULT_LEASE_SEC, + "on_timeout": a.on_timeout}, + "retry": {"max": int(a.retry_max or 0), + "delay_sec": float(a.retry_delay or 0), + "backoff": a.retry_backoff, + "factor": float(a.retry_factor or 2)}}, "nodes": nodes} d = _gdir(a.ws, gid) d.mkdir(parents=True, exist_ok=True) @@ -422,20 +1050,30 @@ def main(argv=None) -> int: p.add_argument("--kpi", default="", help="判据,分号分隔;每条生成一个节点") p.add_argument("--nodes-json", default="", help="节点定义文件(- 表示标准输入)") p.add_argument("--max-parallel", type=int, default=3) + p.add_argument("--lease-sec", type=float, default=None, + help="租约多少秒算超时(默认 3600;0 = 不判超时)") + p.add_argument("--on-timeout", default="requeue", + help="超时处置:requeue(默认)/block/keep") + p.add_argument("--retry-max", type=int, default=0, help="默认自动重试次数(默认 0)") + p.add_argument("--retry-delay", type=float, default=0, help="重试基础间隔秒(默认 0)") + p.add_argument("--retry-backoff", default="fixed", help="fixed(默认)/linear/exponential") + p.add_argument("--retry-factor", type=float, default=2, help="指数退避倍数(默认 2)") + p.add_argument("--deps-rule", default="all_success", + help="默认依赖规则:all_success(默认)/all_done/one_success") p.set_defaults(fn=cmd_init) - for name, fn, extra in (("check", cmd_check, None), ("ready", cmd_ready, None), - ("show", cmd_show, None)): + for name, fn in (("check", cmd_check), ("ready", cmd_ready), ("show", cmd_show)): p = sub.add_parser(name) p.add_argument("--ws", required=True) p.add_argument("--gid", required=True) p.set_defaults(fn=fn) - p = sub.add_parser("start", help="节点开工") + p = sub.add_parser("start", help="节点开工(记租约)") p.add_argument("--ws", required=True) p.add_argument("--gid", required=True) p.add_argument("--node", required=True) p.add_argument("--by", default="") + p.add_argument("--lease-sec", type=float, default=None, help="本次开工的租约秒数(覆盖默认)") p.set_defaults(fn=cmd_start) p = sub.add_parser("report", help="节点回报") @@ -444,9 +1082,18 @@ def main(argv=None) -> int: p.add_argument("--node", required=True) p.add_argument("--state", required=True) p.add_argument("--evidence", default="") + p.add_argument("--artifact", action="append", default=[], + help="产物清单,可重复(供 map.from pick=artifacts 读取)") p.add_argument("--by", default="") p.set_defaults(fn=cmd_report) + p = sub.add_parser("expand", help="手动触发展开") + p.add_argument("--ws", required=True) + p.add_argument("--gid", required=True) + p.add_argument("--node", required=True) + p.add_argument("--force", action="store_true") + p.set_defaults(fn=cmd_expand) + p = sub.add_parser("list") p.add_argument("--ws", required=True) p.set_defaults(fn=cmd_list) diff --git a/session-mechanism/graph/test_graphctl.py b/session-mechanism/graph/test_graphctl.py index 0485caa..fbaa1ab 100644 --- a/session-mechanism/graph/test_graphctl.py +++ b/session-mechanism/graph/test_graphctl.py @@ -2,11 +2,13 @@ # -*- 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 独立性(只写临时目录,不碰现有机制文件) +## 覆盖十二类 +1 建图(--kpi / --nodes-json) 7 边界(单节点 / 空节点 / gid 含中文 / 图不存在) +2 校验硬错(环 / 悬空依赖 / 状态非法) 8 独立性(只写临时目录,不碰现有机制文件) +3 校验警告(无域 / 无判据 / 域重叠) 9 并发写(两进程同时写不丢状态) +4 就绪集(依赖 / 域冲突 / 并发上限) 10 租约(超时回收三种策略 / 名额释放 / 节点级覆盖) +5 状态流转(pending→running→done) 11 重试与触发规则(三种退避 / 三档 deps_rule) +6 非法操作(重复开工 / 非法 state) 12 动态展开(over / from / N=0 / N 大 / child_domain) ## 跑法 python graph/test_graphctl.py -v @@ -17,7 +19,6 @@ """ from __future__ import annotations -import atexit import atexit import json import os @@ -574,6 +575,423 @@ class T9_Concurrency(Base): self.assertGreater(self.load().get("_rev"), r0) +# ── 10 租约与超时 ─────────────────────────────────────────────────────── +class T10_Lease(Base): + """卡死的节点不能永远占着并发名额(2026-10-19 补)。""" + + def backdate(self, node, sec=10000, gid=None): + """把租约时刻往前拨(等价于"这节点已经跑了 sec 秒没回报")。""" + g = self.load(gid) + n = g["nodes"][node] + n["lease_at_ts"] = time.time() - sec + n["lease_at"] = time.strftime("%Y-%m-%dT%H:%M:%S", + time.localtime(time.time() - sec)) + self.dump(g, gid) + + def test_start_records_lease(self): + self.init() + ok("start", "--ws", self.ws, "--gid", self.gid, "--node", "N1") + n = self.load()["nodes"]["N1"] + self.assertIn("lease_at", n) + self.assertGreater(n["lease_at_ts"], 0) + self.assertEqual(n["starts"], 1) + + def test_timeout_requeues_by_default(self): + self.init({"nodes": [{"id": "A", "domain": ["a/"], "accepts": ["x"]}]}) + ok("start", "--ws", self.ws, "--gid", self.gid, "--node", "A") + self.backdate("A") + r = ok("ready", "--ws", self.ws, "--gid", self.gid) + self.assertEqual([x["node"] for x in r["reclaimed"]], ["A"]) + n = self.load()["nodes"]["A"] + self.assertEqual(n["status"], "pending") + self.assertEqual(n["attempts"], 1) + self.assertNotIn("lease_at_ts", n) + + def test_timeout_frees_capacity(self): + """cap=1,A 卡死占着名额 ⇒ 回收后名额必须还给别人。""" + self.init({"max_parallel": 1, "nodes": [ + {"id": "A", "domain": ["a/"], "accepts": ["x"]}, + {"id": "B", "domain": ["b/"], "accepts": ["y"]}]}) + ok("start", "--ws", self.ws, "--gid", self.gid, "--node", "A") + r = ok("ready", "--ws", self.ws, "--gid", self.gid) + self.assertEqual(r["slots"], 0) # 先确认确实被占着 + self.backdate("A") + r = ok("ready", "--ws", self.ws, "--gid", self.gid) + self.assertEqual(r["running"], []) + self.assertEqual(r["slots"], 1) + self.assertEqual(len(r["ready"]), 1) + + def test_policy_block(self): + self.init({"nodes": [{"id": "A", "domain": ["a/"], "accepts": ["x"]}]}) + ok("init", "--ws", self.ws, "--title", "阻塞", "--gid", "b1", + "--on-timeout", "block", "--nodes-json", "-", + stdin=json.dumps({"nodes": [{"id": "A", "domain": ["a/"], "accepts": ["x"]}]})) + ok("start", "--ws", self.ws, "--gid", "b1", "--node", "A") + self.backdate("A", gid="b1") + ok("ready", "--ws", self.ws, "--gid", "b1") + n = self.load("b1")["nodes"]["A"] + self.assertEqual(n["status"], "blocked") + self.assertEqual(n["blocked_by"], "lease_timeout") + self.assertEqual(n["attempts"], 1) + + def test_policy_keep_only_notices(self): + ok("init", "--ws", self.ws, "--title", "只提示", "--gid", "k1", + "--on-timeout", "keep", "--nodes-json", "-", + stdin=json.dumps({"nodes": [{"id": "A", "domain": ["a/"], "accepts": ["x"]}]})) + ok("start", "--ws", self.ws, "--gid", "k1", "--node", "A") + self.backdate("A", gid="k1") + r = ok("ready", "--ws", self.ws, "--gid", "k1") + self.assertEqual(r["reclaimed"], []) + self.assertEqual([x["node"] for x in r["lease_notices"]], ["A"]) + n = self.load("k1")["nodes"]["A"] + self.assertEqual(n["status"], "running") + self.assertEqual(n["attempts"], 0) + + def test_node_level_lease_override(self): + """全局 3600 秒,节点自己声明 60 秒 ⇒ 按节点的算。""" + self.init({"nodes": [{"id": "A", "domain": ["a/"], "accepts": ["x"], + "lease_sec": 60}, + {"id": "B", "domain": ["b/"], "accepts": ["y"]}]}) + ok("start", "--ws", self.ws, "--gid", self.gid, "--node", "A") + ok("start", "--ws", self.ws, "--gid", self.gid, "--node", "B") + self.backdate("A", 120) + self.backdate("B", 120) # B 用全局 3600,不该被回收 + r = ok("ready", "--ws", self.ws, "--gid", self.gid) + self.assertEqual([x["node"] for x in r["reclaimed"]], ["A"]) + + def test_no_lease_field_is_not_reclaimed(self): + """手改出来的 running(没有租约时刻)不判超时 —— 兼容旧图。""" + self.init({"nodes": [{"id": "A", "domain": ["a/"], "accepts": ["x"]}]}) + self.state("A", "running") + r = ok("ready", "--ws", self.ws, "--gid", self.gid) + self.assertEqual(r["reclaimed"], []) + self.assertEqual(self.load()["nodes"]["A"]["status"], "running") + + def test_lease_zero_disables(self): + ok("init", "--ws", self.ws, "--title", "关闭", "--gid", "z1", "--lease-sec", "0", + "--nodes-json", "-", + stdin=json.dumps({"nodes": [{"id": "A", "domain": ["a/"], "accepts": ["x"]}]})) + ok("start", "--ws", self.ws, "--gid", "z1", "--node", "A") + self.backdate("A", 99999, gid="z1") + r = ok("ready", "--ws", self.ws, "--gid", "z1") + self.assertEqual(r["reclaimed"], []) + + def test_reclaim_writes_event(self): + self.init({"nodes": [{"id": "A", "domain": ["a/"], "accepts": ["x"]}]}) + ok("start", "--ws", self.ws, "--gid", self.gid, "--node", "A") + self.backdate("A") + ok("ready", "--ws", self.ws, "--gid", self.gid) + evs = [e for e in self.events() if e.get("event") == "lease_timeout"] + self.assertEqual(len(evs), 1) + self.assertEqual(evs[0]["node"], "A") + + +# ── 11 重试策略与依赖触发规则 ─────────────────────────────────────────── +class T11_Retry(Base): + def _one(self, retry=None, rule=None): + n = {"id": "A", "domain": ["a/"], "accepts": ["x"]} + if retry: + n["retry"] = retry + if rule: + n["deps_rule"] = rule + self.init({"nodes": [n]}) + + def test_retry_requeues_with_delay(self): + self._one({"max": 2, "delay_sec": 30}) + r = ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "A", + "--state", "blocked") + self.assertEqual(r["status"], "pending") + self.assertTrue(r["retry"]["requeued"]) + self.assertEqual(r["retry"]["delay_sec"], 30) + n = self.load()["nodes"]["A"] + self.assertEqual(n["status"], "pending") + self.assertEqual(n["attempts"], 1) + self.assertGreater(n["not_before"], time.time() + 20) + + def test_retry_waits_until_not_before(self): + self._one({"max": 2, "delay_sec": 600}) + ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "A", "--state", "blocked") + r = ok("ready", "--ws", self.ws, "--gid", self.gid) + self.assertNotIn("A", r["ready"]) + self.assertTrue(any(w["node"] == "A" and "重试等待中" in w["why"] + for w in r["waiting"])) + + def test_retry_ready_after_delay(self): + self._one({"max": 2, "delay_sec": 600}) + ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "A", "--state", "blocked") + g = self.load() + g["nodes"]["A"]["not_before"] = time.time() - 1 + self.dump(g) + r = ok("ready", "--ws", self.ws, "--gid", self.gid) + self.assertIn("A", r["ready"]) + + def test_retry_exhausted_blocks(self): + self._one({"max": 1, "delay_sec": 0}) + r1 = ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "A", + "--state", "blocked") + self.assertEqual(r1["status"], "pending") # 第 1 次失败 ⇒ 还给 1 次 + r2 = ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "A", + "--state", "blocked") + self.assertEqual(r2["status"], "blocked") + n = self.load()["nodes"]["A"] + self.assertEqual(n["attempts"], 2) + self.assertTrue(n["retry_exhausted"]) + + def test_backoff_exponential_grows(self): + self._one({"max": 3, "delay_sec": 10, "backoff": "exponential", "factor": 2}) + r1 = ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "A", + "--state", "blocked") + self.assertEqual(r1["retry"]["delay_sec"], 10) + g = self.load() + g["nodes"]["A"]["not_before"] = time.time() - 1 + self.dump(g) + r2 = ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "A", + "--state", "blocked") + self.assertEqual(r2["retry"]["delay_sec"], 20) + + def test_backoff_linear(self): + self._one({"max": 3, "delay_sec": 10, "backoff": "linear"}) + r = ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "A", + "--state", "blocked") + self.assertEqual(r["retry"]["delay_sec"], 10) + g = self.load() + g["nodes"]["A"]["not_before"] = time.time() - 1 + self.dump(g) + r2 = ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "A", + "--state", "blocked") + self.assertEqual(r2["retry"]["delay_sec"], 20) # 10 × 2 + + def test_default_no_auto_retry(self): + """没写 retry ⇒ 与旧行为一致:blocked 立刻 blocked(⛔ 不许偷偷重排)。""" + self._one() + r = ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "A", + "--state", "blocked") + self.assertEqual(r["status"], "blocked") + self.assertEqual(self.load()["nodes"]["A"]["attempts"], 1) + + def _dep_case(self, rule): + child = {"id": "C", "domain": ["c/"], "accepts": ["x"], "deps": ["A", "B"]} + if rule: + child["deps_rule"] = rule + self.init({"nodes": [ + {"id": "A", "domain": ["a/"], "accepts": ["x"]}, + {"id": "B", "domain": ["b/"], "accepts": ["y"]}, + child]}) + + def test_deps_rule_default_all_success(self): + self._dep_case(None) + ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "A", "--state", "done") + ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "B", "--state", "blocked") + r = ok("ready", "--ws", self.ws, "--gid", self.gid) + self.assertNotIn("C", r["ready"]) + + def test_deps_rule_all_done(self): + """all_done:上游跑完就算(成败不论)。""" + self._dep_case("all_done") + ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "A", "--state", "done") + ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "B", "--state", "blocked") + r = ok("ready", "--ws", self.ws, "--gid", self.gid) + self.assertIn("C", r["ready"]) + + def test_deps_rule_all_done_waits_pending(self): + self._dep_case("all_done") + ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "A", "--state", "done") + r = ok("ready", "--ws", self.ws, "--gid", self.gid) + self.assertNotIn("C", r["ready"]) + self.assertTrue(any(w["node"] == "C" and "依赖未跑完" in w["why"] + for w in r["waiting"])) + + def test_deps_rule_one_success(self): + self._dep_case("one_success") + ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "A", "--state", "done") + ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "B", "--state", "blocked") + r = ok("ready", "--ws", self.ws, "--gid", self.gid) + self.assertIn("C", r["ready"]) + + def test_deps_rule_one_success_none_yet(self): + self._dep_case("one_success") + r = ok("ready", "--ws", self.ws, "--gid", self.gid) + self.assertNotIn("C", r["ready"]) + self.assertTrue(any(w["node"] == "C" and "任一上游成功" in w["why"] + for w in r["waiting"])) + + def test_illegal_deps_rule_rejected(self): + err = bad("init", "--ws", self.ws, "--title", "坏规则", "--gid", "bad", + "--nodes-json", "-", + stdin=json.dumps({"nodes": [{"id": "A", "deps_rule": "any"}]})) + self.assertIn("deps_rule", err) + + def test_illegal_deps_rule_caught_by_check(self): + self.init({"nodes": [{"id": "A", "accepts": ["x"]}]}) + g = self.load() + g["nodes"]["A"]["deps_rule"] = "any" + self.dump(g) + r = json.loads(_run("check", "--ws", self.ws, "--gid", self.gid)[1]) + self.assertFalse(r["ok"]) + self.assertTrue(any("deps_rule" in e for e in r["errors"])) + + +# ── 12 动态展开(map / foreach)──────────────────────────────────────── +class T12_Map(Base): + def test_expand_from_list(self): + self.init({"max_parallel": 5, "nodes": [ + {"id": "P", "title": "逐页改", "accepts": ["都改完"], + "map": {"over": ["a.md", "b.md", "c.md"]}}]}) + r = ok("ready", "--ws", self.ws, "--gid", self.gid) + self.assertEqual(r["expanded"][0]["node"], "P") + self.assertEqual(r["expanded"][0]["count"], 3) + g = self.load() + self.assertEqual(g["nodes"]["P"]["status"], "gating") + self.assertEqual(g["nodes"]["P"]["children"], ["P#1", "P#2", "P#3"]) + self.assertEqual(g["nodes"]["P#2"]["item"], "b.md") + self.assertEqual(sorted(r["ready"]), ["P#1", "P#2", "P#3"]) + + def test_children_done_completes_parent(self): + self.init({"max_parallel": 5, "nodes": [ + {"id": "P", "accepts": ["齐"], "map": {"over": ["x", "y"]}}, + {"id": "Z", "domain": ["z/"], "accepts": ["收尾"], "deps": ["P"]}]}) + ok("ready", "--ws", self.ws, "--gid", self.gid) + ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "P#1", "--state", "done") + r = ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "P#2", "--state", "done") + g = self.load() + self.assertEqual(g["nodes"]["P"]["status"], "done") + self.assertEqual(r["completed_parents"], ["P"]) + self.assertIn("Z", r["next_ready"]) + + def test_parent_not_done_while_children_left(self): + self.init({"max_parallel": 5, "nodes": [ + {"id": "P", "accepts": ["齐"], "map": {"over": ["x", "y"]}}]}) + ok("ready", "--ws", self.ws, "--gid", self.gid) + ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "P#1", "--state", "done") + g = self.load() + self.assertEqual(g["nodes"]["P"]["status"], "gating") + r = ok("ready", "--ws", self.ws, "--gid", self.gid) + self.assertTrue(any(w["node"] == "P" and "子节点" in w["why"] for w in r["waiting"])) + + def test_zero_items_completes_immediately(self): + self.init({"nodes": [{"id": "P", "accepts": ["齐"], + "map": {"over": []}}]}) + r = ok("ready", "--ws", self.ws, "--gid", self.gid) + g = self.load() + self.assertEqual(g["nodes"]["P"]["status"], "done") + self.assertEqual(g["nodes"]["P"]["children"], []) + self.assertEqual([x for x in g["nodes"]], ["P"]) + self.assertEqual(r["ready"], []) + + def test_expand_from_evidence_lines(self): + self.init({"max_parallel": 4, "nodes": [ + {"id": "U", "domain": ["u/"], "accepts": ["清单在"]}, + {"id": "P", "accepts": ["都处理"], + "deps": ["U"], "map": {"from": {"node": "U", "pick": "evidence"}}}]}) + ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "U", "--state", "done", + "--evidence", "a.md\nb.md;c.md") + ok("ready", "--ws", self.ws, "--gid", self.gid) + self.assertEqual(self.load()["nodes"]["P"]["children"], ["P#1", "P#2", "P#3"]) + + def test_expand_from_artifacts(self): + self.init({"max_parallel": 4, "nodes": [ + {"id": "U", "domain": ["u/"], "accepts": ["清单在"]}, + {"id": "P", "accepts": ["都处理"], + "deps": ["U"], "map": {"from": {"node": "U", "pick": "artifacts"}}}]}) + ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "U", "--state", "done", + "--artifact", "a.md", "--artifact", "b.md") + ok("ready", "--ws", self.ws, "--gid", self.gid) + self.assertEqual(self.load()["nodes"]["P"]["children"], ["P#1", "P#2"]) + + def test_expand_source_not_ready_waits(self): + self.init({"nodes": [ + {"id": "U", "domain": ["u/"], "accepts": ["清单在"]}, + {"id": "P", "accepts": ["都处理"], + "deps": ["U"], "map": {"from": {"node": "U", "pick": "evidence"}}}]}) + r = ok("ready", "--ws", self.ws, "--gid", self.gid) + self.assertNotIn("P", r["ready"]) + self.assertEqual(self.load()["nodes"]["P"]["status"], "pending") + + def test_large_n_blocked_by_max_parallel(self): + """N=10 但 cap=2 ⇒ 这一批只放 2 个,其余进 backlog(⛔ 不许一次性全放出去)。""" + self.init({"max_parallel": 2, "nodes": [ + {"id": "P", "accepts": ["齐"], "map": {"over": ["i%d" % i for i in range(1, 11)]}}]}) + r = ok("ready", "--ws", self.ws, "--gid", self.gid) + self.assertEqual(len(self.load()["nodes"]["P"]["children"]), 10) + self.assertEqual(len(r["ready"]), 2) + self.assertEqual(len(r["backlog"]), 8) + + def test_child_domain_template_allows_parallel(self): + self.init({"max_parallel": 3, "nodes": [ + {"id": "P", "accepts": ["齐"], + "map": {"over": ["a", "b"], "child_domain": "out/{i}/"}}]}) + r = ok("ready", "--ws", self.ws, "--gid", self.gid) + g = self.load() + self.assertEqual(g["nodes"]["P#1"]["domain"], ["out/1/"]) + self.assertEqual(g["nodes"]["P#2"]["domain"], ["out/2/"]) + self.assertEqual(sorted(r["ready"]), ["P#1", "P#2"]) + + def test_children_default_have_no_domain(self): + """默认不继承父域:同一批展开通常写同一批文件,继承会把并发挡死。""" + self.init({"max_parallel": 3, "nodes": [ + {"id": "P", "domain": ["docs/"], "accepts": ["齐"], + "map": {"over": ["a", "b"]}}]}) + ok("ready", "--ws", self.ws, "--gid", self.gid) + g = self.load() + self.assertEqual(g["nodes"]["P#1"]["domain"], []) + self.assertEqual(len(ok("ready", "--ws", self.ws, "--gid", self.gid)["ready"]), 2) + + def test_start_map_node_rejected(self): + self.init({"nodes": [{"id": "P", "accepts": ["齐"], + "map": {"over": ["a"]}}]}) + err = bad("start", "--ws", self.ws, "--gid", self.gid, "--node", "P") + self.assertIn("展开节点", err) + + def test_report_on_map_node_rejected(self): + self.init({"nodes": [{"id": "P", "accepts": ["齐"], + "map": {"over": ["a"]}}]}) + ok("ready", "--ws", self.ws, "--gid", self.gid) + err = bad("report", "--ws", self.ws, "--gid", self.gid, "--node", "P", + "--state", "done") + self.assertIn("展开父节点", err) + + def test_check_invalid_map(self): + self.init({"nodes": [{"id": "P", "accepts": ["齐"]}]}) + g = self.load() + g["nodes"]["P"]["map"] = {"as": "item"} + self.dump(g) + r = json.loads(_run("check", "--ws", self.ws, "--gid", self.gid)[1]) + self.assertFalse(r["ok"]) + self.assertTrue(any("map" in e for e in r["errors"])) + + def test_init_invalid_map_rejected(self): + err = bad("init", "--ws", self.ws, "--title", "坏展开", "--gid", "bad", + "--nodes-json", "-", + stdin=json.dumps({"nodes": [{"id": "P", "map": {"over": "不是列表"}}]})) + self.assertIn("over", err) + + def test_expand_command_manual(self): + self.init({"nodes": [{"id": "P", "accepts": ["齐"], + "map": {"over": ["a", "b"]}}]}) + r = ok("expand", "--ws", self.ws, "--gid", self.gid, "--node", "P") + self.assertEqual(r["count"], 2) + err = bad("expand", "--ws", self.ws, "--gid", self.gid, "--node", "P") + self.assertIn("已展开", err) + + def test_expand_waits_for_deps(self): + """有依赖的展开节点:上游没完成时手动 expand 要拒绝(除非 --force)。""" + self.init({"nodes": [{"id": "U", "domain": ["u/"], "accepts": ["x"]}, + {"id": "P", "accepts": ["齐"], "deps": ["U"], + "map": {"over": ["a"]}}]}) + err = bad("expand", "--ws", self.ws, "--gid", self.gid, "--node", "P") + self.assertIn("未就绪", err) + + def test_child_attempts_independent(self): + self.init({"max_parallel": 3, "nodes": [ + {"id": "P", "accepts": ["齐"], "map": {"over": ["a", "b"]}}]}) + ok("ready", "--ws", self.ws, "--gid", self.gid) + ok("report", "--ws", self.ws, "--gid", self.gid, "--node", "P#1", "--state", "blocked") + g = self.load() + self.assertEqual(g["nodes"]["P#1"]["attempts"], 1) + self.assertEqual(g["nodes"]["P#2"]["attempts"], 0) + self.assertEqual(g["nodes"]["P"]["attempts"], 0) + + def tearDownModule(): _drop(TMP_ROOT)