1691 lines
69 KiB
Python
1691 lines
69 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""graphctl.py —— 「图模式」的独立实现（2026-10-18；2026-10-19 补批量能力）。
|
||
|
||
## 它是什么
|
||
|
||
把目标拆成节点、按依赖推进的**独立**运行模式。**⛔ 不修改任何现有机制**：
|
||
不读也不写 `goal.json` / `tasks.json` / `collabd-state.json`，不参与六道闸，
|
||
不接管常驻循环，不改变现有会话的任何行为。
|
||
|
||
## 数据落点（独立目录，与现有状态完全分开）
|
||
|
||
<ws>/.workbuddy/graph/<gid>/graph.json 图的唯一真源
|
||
<ws>/.workbuddy/graph/<gid>/events.jsonl 每次状态变更的流水
|
||
|
||
## 图的结构
|
||
|
||
{
|
||
"gid": "20261018-2230-ui改造",
|
||
"title": "给三块功能各加一份文档",
|
||
"created_at": "2026-10-18T22:30:00",
|
||
"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": "", "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`；
|
||
另有 `gating`——展开出来的父节点，等子节点全部完成（父节点自动转 `done`）。
|
||
|
||
## 命令
|
||
|
||
init --ws <工作区> --title <目标> (--kpi "判据A;判据B" | --nodes-json <文件|->)
|
||
[--max-parallel N] [--lease-sec S] [--on-timeout requeue|block|keep]
|
||
[--retry-max N] [--retry-delay S] [--retry-backoff fixed|linear|exponential]
|
||
check --ws <ws> --gid <gid> 环／悬空依赖／空域／同批域冲突／非法规则
|
||
ready --ws <ws> --gid <gid> 就绪集（含为什么没被选）；顺带回收超时租约、展开 map
|
||
start --ws <ws> --gid <gid> --node N1 [--by ...] [--lease-sec S]
|
||
report --ws <ws> --gid <gid> --node N1 --state done|blocked [--evidence "..."] [--artifact P ...]
|
||
expand --ws <ws> --gid <gid> --node P [--force] 手动触发展开
|
||
show --ws <ws> --gid <gid>
|
||
list --ws <ws>
|
||
tick --ws <ws> --gid <gid> [--by ...] [--limit N] [--dry-run] 自动派活（算完就开）
|
||
approve --ws <ws> --gid <gid> --node N [--reject] [--reason ...] 人审
|
||
cancel --ws <ws> --gid <gid> (--node N | --all) [--reason ...] 取消
|
||
plan --ws <ws> --gid <gid> [--scope ...] [--after-sec N] 出一行排期
|
||
progress --ws <ws> --gid <gid> 进度／关键路径
|
||
|
||
## 复杂任务能力（2026-10-19 补，C 方案）
|
||
|
||
条件分支：节点 `when: [{"node": "A", "status": "blocked"}]` —— 条件被证伪即
|
||
`skipped`；`on_dep_fail: wait|skip|block` 决定上游跑不成时下游怎么走
|
||
（默认 wait，与旧行为一致）。
|
||
子图：节点 `subgraph: {"nodes": [...]}` —— 内联子流程展开成 `<父>/<子id>`，
|
||
父节点 `gating`，子节点全成功才 `done`，有不成功的 ⇒ 父 `blocked`。
|
||
人审：节点 `approval: true` —— report done 后进 `awaiting`，`approve` 才放行
|
||
（`--reject` ⇒ blocked）。
|
||
取消：`cancelled` 是终态，连坐子节点；下游按 `on_dep_fail` 处置。
|
||
分池：`meta.pools: {"域前缀": 并发上限}` —— 同域并发受限，与 `max_parallel` 叠加。
|
||
进度：`progress` 给出 by_status／percent／关键路径（未完成节点的最长依赖链）。
|
||
|
||
## 就绪的判据
|
||
|
||
① 依赖按 `deps_rule` 满足（默认 all_success）
|
||
② 自己还是 pending，且不在重试等待窗内（`not_before` 未到 ⇒ 等）
|
||
③ 域不与任何 running 节点重叠，也不与本批已选节点重叠
|
||
④ 不超 max_parallel（running 占名额）
|
||
|
||
## 三件批量能力（2026-10-19 补）
|
||
|
||
租约：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
|
||
import time
|
||
from pathlib import Path
|
||
|
||
MARK = ".workbuddy/graph"
|
||
LOCK_STALE_SEC = 30.0 # 锁目录超过这么久没动静 ⇒ 认定持锁进程已死，接管
|
||
STATUSES = ("pending", "running", "done", "blocked", "gating",
|
||
"awaiting", "skipped", "cancelled")
|
||
SETTLED = ("done", "blocked", "skipped", "cancelled") # 不会再自己变了
|
||
DEP_RULES = ("all_success", "all_done", "one_success")
|
||
TIMEOUT_POLICIES = ("requeue", "block", "keep")
|
||
BACKOFFS = ("fixed", "linear", "exponential")
|
||
ON_DEP_FAIL = ("wait", "skip", "block") # 上游跑不成时下游怎么办
|
||
DEFAULT_LEASE_SEC = 3600.0
|
||
|
||
|
||
# ── 基础设施 ────────────────────────────────────────────────────────────
|
||
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
|
||
|
||
|
||
def _gdir(ws: str, gid: str) -> Path:
|
||
return _root(ws) / gid
|
||
|
||
|
||
def _load(ws: str, gid: str) -> dict:
|
||
p = _gdir(ws, gid) / "graph.json"
|
||
if not p.is_file():
|
||
raise SystemExit("找不到图：%s" % p)
|
||
return json.loads(p.read_text(encoding="utf-8"))
|
||
|
||
|
||
def _save(ws: str, gid: str, g: dict) -> None:
|
||
"""整份写回（新建图时用）。临时名带 pid ＋ 随机数 ⇒ 两进程不会撞同一个名字。"""
|
||
p = _gdir(ws, gid) / "graph.json"
|
||
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) # 原子替换，避免半截文件
|
||
|
||
|
||
class _Lock:
|
||
"""写串行化：`os.mkdir` 是原子的，用它当互斥量（⛔ 不用 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)
|
||
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:
|
||
p = _gdir(ws, gid) / "events.jsonl"
|
||
obj = dict(obj)
|
||
obj.setdefault("at", _now())
|
||
with io.open(p, "a", encoding="utf-8", newline="\n") as f:
|
||
f.write(json.dumps(obj, ensure_ascii=False) + "\n")
|
||
|
||
|
||
def _out(obj) -> None:
|
||
data = json.dumps(obj, ensure_ascii=False, indent=2) + "\n"
|
||
sys.stdout.buffer.write(data.encode("utf-8"))
|
||
|
||
|
||
# ── 域与依赖 ────────────────────────────────────────────────────────────
|
||
def _norm_dom(d) -> list:
|
||
if isinstance(d, str):
|
||
d = [d]
|
||
out = []
|
||
for x in (d or []):
|
||
s = str(x).strip().replace("\\", "/").lstrip("./")
|
||
if s:
|
||
out.append(s if s.endswith("/") else s + "/")
|
||
return out
|
||
|
||
|
||
def _dom_overlap(a, b) -> bool:
|
||
"""任一前缀互为前缀 ⇒ 重叠（docs/ 与 docs/ui/ 算重叠）。"""
|
||
for x in a:
|
||
for y in b:
|
||
if x.startswith(y) or y.startswith(x):
|
||
return True
|
||
return False
|
||
|
||
|
||
def _deps(n: dict) -> list:
|
||
v = n.get("deps") or []
|
||
return [str(x) for x in v] if isinstance(v, list) else []
|
||
|
||
|
||
def _is_done(nodes: dict, nid: str) -> bool:
|
||
return str((nodes.get(nid) or {}).get("status")) == "done"
|
||
|
||
|
||
def _st(nodes: dict, nid: str) -> str:
|
||
return str((nodes.get(nid) or {}).get("status"))
|
||
|
||
|
||
def _dep_verdict(nodes: dict, n: dict) -> dict:
|
||
"""依赖判定 ⇒ {"verdict": go|wait|skip|block, "why": 说明}
|
||
|
||
① `when` 条件子句先判（条件分支）：全部成立 ⇒ 继续；任一被证伪 ⇒ skip。
|
||
② 再按 `deps_rule` 判：all_success（默认）／all_done／one_success。
|
||
③ 上游停在 skipped / cancelled / blocked 时按 `on_dep_fail` 处置：
|
||
wait（默认，与旧行为一致）／skip（连坐跳过）／block（整枝停住）。
|
||
"""
|
||
deps = _deps(n)
|
||
rule = str(n.get("deps_rule") or "all_success")
|
||
on_fail = str(n.get("on_dep_fail") or "wait")
|
||
if on_fail not in ON_DEP_FAIL:
|
||
on_fail = "wait"
|
||
st = {d: _st(nodes, d) for d in deps}
|
||
|
||
for c in (n.get("when") or []):
|
||
if not isinstance(c, dict):
|
||
continue
|
||
cnode = str(c.get("node") or "")
|
||
want = str(c.get("status") or "done")
|
||
got = _st(nodes, cnode)
|
||
if got == want:
|
||
continue
|
||
if got in SETTLED:
|
||
return {"verdict": "skip",
|
||
"why": "条件不成立：%s 是 %s，要 %s" % (cnode, got, want)}
|
||
if not deps:
|
||
return {"verdict": "wait", "why": "条件节点 %s 还没定论（%s）" % (cnode, got)}
|
||
if not deps:
|
||
return {"verdict": "go", "why": ""}
|
||
|
||
dead = [(d, s) for d, s in st.items() if s in ("skipped", "cancelled")] \
|
||
or [(d, s) for d, s in st.items() if s == "blocked"]
|
||
|
||
if rule == "all_done":
|
||
lack = [d for d, s in st.items() if s not in SETTLED]
|
||
if lack:
|
||
return {"verdict": "wait", "why": "依赖未跑完：%s" % ", ".join(lack)}
|
||
return {"verdict": "go", "why": ""}
|
||
|
||
if rule == "one_success":
|
||
if any(s == "done" for s in st.values()):
|
||
return {"verdict": "go", "why": ""}
|
||
if dead and all(s in SETTLED for s in st.values()):
|
||
return {"verdict": on_fail if on_fail != "wait" else "wait",
|
||
"why": "没有任一上游成功：%s"
|
||
% ", ".join("%s(%s)" % (d, s) for d, s in st.items())}
|
||
return {"verdict": "wait",
|
||
"why": "等待任一上游成功：%s"
|
||
% ", ".join("%s(%s)" % (d, s) for d, s in st.items())}
|
||
|
||
lack = [d for d, s in st.items() if s != "done"]
|
||
if not lack:
|
||
return {"verdict": "go", "why": ""}
|
||
if dead:
|
||
why = "上游不会成功了：%s" % ", ".join("%s(%s)" % (d, s) for d, s in dead)
|
||
if on_fail != "wait":
|
||
return {"verdict": on_fail, "why": why}
|
||
return {"verdict": "wait",
|
||
"why": "依赖未完成：%s" % ", ".join("%s(%s)" % (d, st[d]) for d in lack)}
|
||
return {"verdict": "wait",
|
||
"why": "依赖未完成：%s" % ", ".join("%s(%s)" % (d, st[d]) for d in lack)}
|
||
|
||
|
||
def _deps_ok(nodes: dict, n: dict):
|
||
"""按 `deps_rule` 判依赖是否放行，返回 (bool, 没满足的原因)。"""
|
||
v = _dep_verdict(nodes, n)
|
||
return v["verdict"] == "go", v["why"]
|
||
|
||
|
||
# ── 三件批量能力的配置读法 ──────────────────────────────────────────────
|
||
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 = {}, [], []
|
||
|
||
def dfs(u):
|
||
color[u] = 1
|
||
stack.append(u)
|
||
for v in _deps(nodes.get(u) or {}):
|
||
if v not in nodes:
|
||
continue
|
||
if color.get(v) == 1:
|
||
i = stack.index(v)
|
||
hit.extend(stack[i:] + [v])
|
||
return True
|
||
if color.get(v, 0) == 0 and dfs(v):
|
||
return True
|
||
stack.pop()
|
||
color[u] = 2
|
||
return False
|
||
|
||
for k in nodes:
|
||
if color.get(k, 0) == 0 and dfs(k):
|
||
return hit
|
||
return []
|
||
|
||
|
||
def cmd_check(a) -> int:
|
||
g = _load(a.ws, a.gid)
|
||
nodes = g.get("nodes") or {}
|
||
errs, warns = [], []
|
||
|
||
cyc = _find_cycle(nodes)
|
||
if cyc:
|
||
errs.append("存在环：%s" % " → ".join(cyc))
|
||
|
||
for nid, n in nodes.items():
|
||
for d in _deps(n):
|
||
if d not in nodes:
|
||
errs.append("%s 的依赖 %s 不存在" % (nid, d))
|
||
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("map") and n.get("subgraph"):
|
||
errs.append("%s 同时声明了 map 与 subgraph（只能选一个）" % nid)
|
||
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)
|
||
kids = []
|
||
seen = set()
|
||
for s in kids:
|
||
sid = str((s or {}).get("id") or "")
|
||
if not sid:
|
||
warns.append("%s 的子图节点缺 id ⇒ 会按顺序补 S1/S2，deps 引用会指不到"
|
||
% nid)
|
||
elif sid in seen:
|
||
errs.append("%s 的子图里有重复 id：%s" % (nid, sid))
|
||
else:
|
||
seen.add(sid)
|
||
for s in kids:
|
||
for d in ((s or {}).get("deps") or []):
|
||
if str(d) not in seen and str(d) not in nodes:
|
||
errs.append("%s/%s 的依赖 %s 既不在子图里也不在图里"
|
||
% (nid, (s or {}).get("id") or "?", d))
|
||
if str(n.get("on_dep_fail") or "wait") not in ON_DEP_FAIL:
|
||
errs.append("%s 的 on_dep_fail 非法：%r（只认 %s）"
|
||
% (nid, n.get("on_dep_fail"), " / ".join(ON_DEP_FAIL)))
|
||
for c in (n.get("when") or []):
|
||
if not isinstance(c, dict) or not str(c.get("node") or ""):
|
||
errs.append('%s 的 when 子句要写 {"node": "A", "status": "done"}' % nid)
|
||
continue
|
||
if str(c["node"]) not in nodes:
|
||
errs.append("%s 的 when 条件节点 %s 不存在" % (nid, c["node"]))
|
||
if str(c.get("status") or "done") not in STATUSES:
|
||
errs.append("%s 的 when 期望状态非法：%r" % (nid, c.get("status")))
|
||
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 []) and not n.get("map") and not n.get("parent"):
|
||
warns.append("%s 没有验收判据 ⇒ 判不出完成" % nid)
|
||
|
||
pools = _meta(g).get("pools")
|
||
if pools is not None:
|
||
if not isinstance(pools, dict):
|
||
errs.append('meta.pools 要写成 {"域前缀": 并发上限}')
|
||
else:
|
||
doms = [(_norm_dom(n.get("domain")), nid) for nid, n in nodes.items()]
|
||
for k, v in pools.items():
|
||
try:
|
||
cap = int(v)
|
||
except (TypeError, ValueError):
|
||
errs.append("meta.pools[%s] 不是数字：%r" % (k, v))
|
||
continue
|
||
if cap <= 0:
|
||
errs.append("meta.pools[%s] 必须是正整数" % k)
|
||
pre = _norm_dom(k)
|
||
if not pre:
|
||
errs.append("meta.pools 的域不能为空")
|
||
elif not any(d and _dom_overlap(pre, d) for d, _n in doms):
|
||
warns.append("域池 %s 没有匹配任何节点的域" % k)
|
||
|
||
ids = list(nodes)
|
||
for i in range(len(ids)):
|
||
for j in range(i + 1, len(ids)):
|
||
a1, a2 = ids[i], ids[j]
|
||
if _norm_dom(nodes[a1].get("domain")) and \
|
||
_dom_overlap(_norm_dom(nodes[a1].get("domain")),
|
||
_norm_dom(nodes[a2].get("domain"))):
|
||
warns.append("%s 与 %s 的域重叠 ⇒ 永不同批" % (a1, a2))
|
||
|
||
_out({"ok": not errs, "errors": errs, "warnings": warns,
|
||
"nodes": len(nodes)})
|
||
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(_st(nodes, c) in SETTLED for c in kids):
|
||
continue
|
||
arts = []
|
||
for c in kids:
|
||
arts.extend(list((nodes.get(c) or {}).get("artifacts") or []))
|
||
if arts:
|
||
n["artifacts"] = arts
|
||
if all(_is_done(nodes, c) for c in kids):
|
||
n["status"] = "done"
|
||
n["completed_at"] = _now()
|
||
n["evidence"] = n.get("evidence") or ("展开 %d 项全部完成" % len(kids))
|
||
acts.append({"node": nid, "event": "children_done", "to": "done",
|
||
"count": len(kids)})
|
||
else: # 子节点全定下来了，但不全成功 ⇒ 父节点不算成功
|
||
bad = ["%s(%s)" % (c, _st(nodes, c)) for c in kids if not _is_done(nodes, c)]
|
||
n["status"] = "blocked"
|
||
n["blocked_by"] = "child_failed"
|
||
n["blocked_reason"] = "子节点未全部成功：%s" % ", ".join(bad)
|
||
acts.append({"node": nid, "event": "children_failed", "to": "blocked",
|
||
"count": len(kids), "bad": bad})
|
||
changed += 1
|
||
return changed
|
||
|
||
|
||
# ── 条件分支（when / on_dep_fail）───────────────────────────────────────
|
||
def _apply_conditions(g: dict, acts: list) -> int:
|
||
"""把「条件不成立 ⇒ skipped」「上游跑不成且要求 block ⇒ blocked」落到状态上。"""
|
||
nodes = g.get("nodes") or {}
|
||
changed = 0
|
||
for nid, n in list(nodes.items()):
|
||
if str(n.get("status")) != "pending":
|
||
continue
|
||
if n.get("map") and not n.get("expanded"):
|
||
continue
|
||
if n.get("subgraph") and not n.get("expanded"):
|
||
continue
|
||
v = _dep_verdict(nodes, n)
|
||
if v["verdict"] == "skip":
|
||
n["status"] = "skipped"
|
||
n["skipped_at"] = _now()
|
||
n["skipped_reason"] = v["why"]
|
||
acts.append({"node": nid, "event": "skip", "to": "skipped",
|
||
"why": v["why"]})
|
||
changed += 1
|
||
elif v["verdict"] == "block":
|
||
n["status"] = "blocked"
|
||
n["blocked_by"] = "dep"
|
||
n["blocked_reason"] = v["why"]
|
||
acts.append({"node": nid, "event": "block", "to": "blocked",
|
||
"why": v["why"]})
|
||
changed += 1
|
||
return changed
|
||
|
||
|
||
# ── 子图（节点内部再拆子流程）───────────────────────────────────────────
|
||
def _expand_subgraphs(g: dict, acts: list) -> int:
|
||
nodes = g.get("nodes") or {}
|
||
changed = 0
|
||
for nid, n in list(nodes.items()):
|
||
if str(n.get("status")) != "pending" or n.get("expanded") or not n.get("subgraph"):
|
||
continue
|
||
if not _deps_ok(nodes, n)[0]:
|
||
continue
|
||
acts.append(_expand_subgraph(g, nodes, nid))
|
||
changed += 1
|
||
return changed
|
||
|
||
|
||
def _expand_subgraph(g: dict, nodes: dict, nid: str) -> dict:
|
||
"""内联子图 ⇒ 子节点 `<父>/<子id>`，父节点转 gating，子节点全成功才 done。"""
|
||
n = nodes[nid]
|
||
sub = n.get("subgraph") or {}
|
||
specs = [s for s in (sub.get("nodes") or []) if isinstance(s, dict)]
|
||
prefix = nid + "/"
|
||
local, kids = {}, []
|
||
for i, s in enumerate(specs, 1):
|
||
raw = str(s.get("id") or ("S%d" % i)).strip()
|
||
cid = prefix + raw
|
||
if cid in nodes:
|
||
raise SystemExit("子图节点 id 已被占用：%s" % cid)
|
||
local[raw] = cid
|
||
kids.append((cid, s))
|
||
n["expanded"] = True
|
||
n["expanded_at"] = _now()
|
||
if not kids: # 空子图 ⇒ 无需执行
|
||
n["status"] = "done"
|
||
n["children"] = []
|
||
n["evidence"] = n.get("evidence") or "子图为空，无需执行"
|
||
return {"node": nid, "event": "expand", "kind": "subgraph", "count": 0, "to": "done"}
|
||
base = _norm_dom(n.get("domain")) or _norm_dom(sub.get("child_domain"))
|
||
outer = _deps(n)
|
||
for cid, s in kids:
|
||
inner = []
|
||
for d in (s.get("deps") or []):
|
||
tgt = local.get(str(d)) or str(d)
|
||
if tgt not in inner:
|
||
inner.append(tgt)
|
||
deps = list(outer) + [d for d in inner if d not in outer]
|
||
dom = _norm_dom(s.get("domain"))
|
||
if not dom:
|
||
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 [])],
|
||
"status": "pending", "evidence": "", "artifacts": [],
|
||
"attempts": 0, "parent": nid, "sub": True}
|
||
for k in ("deps_rule", "on_dep_fail", "approval", "lease_sec", "when"):
|
||
if s.get(k) not in (None, "", []):
|
||
kid[k] = s[k]
|
||
if s.get("retry"):
|
||
kid["retry"] = _retry_cfg({"meta": {}}, {"retry": s["retry"]})
|
||
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"}
|
||
|
||
|
||
# ── 按域分池 ────────────────────────────────────────────────────────────
|
||
def _pool_caps(g: dict) -> list:
|
||
p = _meta(g).get("pools")
|
||
out = []
|
||
if isinstance(p, dict):
|
||
for k, v in p.items():
|
||
try:
|
||
cap = int(v)
|
||
except (TypeError, ValueError):
|
||
continue
|
||
dom = _norm_dom(k)
|
||
if cap > 0 and dom:
|
||
out.append((dom[0], cap))
|
||
return out
|
||
|
||
|
||
def _pool_of(g: dict, n: dict) -> list:
|
||
dom = _norm_dom((n or {}).get("domain"))
|
||
if not dom:
|
||
return []
|
||
return [pre for pre, _c in _pool_caps(g) if _dom_overlap(dom, [pre])]
|
||
|
||
|
||
def _pool_full(g: dict, n: dict, occupied: list) -> dict:
|
||
"""节点所在池是否已满；满了返回 {"pool","cap","used"}，否则 {}。"""
|
||
caps = dict(_pool_caps(g))
|
||
for pre in _pool_of(g, n):
|
||
cap = caps.get(pre) or 0
|
||
if cap <= 0:
|
||
continue
|
||
used = sum(1 for x in occupied if pre in _pool_of(g, x))
|
||
if used >= cap:
|
||
return {"pool": pre, "cap": cap, "used": used}
|
||
return {}
|
||
|
||
|
||
def _advance(g: dict) -> list:
|
||
"""把「超时回收／子节点收敛／展开／条件判定」推到不动点；返回动了哪些（供写事件）。"""
|
||
acts: list = []
|
||
for _ in range(6):
|
||
moved = 0
|
||
moved += _reclaim_leases(g, acts)
|
||
moved += _complete_parents(g, acts)
|
||
moved += _expand_maps(g, acts)
|
||
moved += _expand_subgraphs(g, acts)
|
||
moved += _apply_conditions(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(_st(nodes, c) in SETTLED for c in kids):
|
||
return True
|
||
elif st == "pending":
|
||
if n.get("expanded"):
|
||
continue
|
||
if n.get("map"):
|
||
if _deps_ok(nodes, n)[0] and _map_items(g, nodes, nid) is not None:
|
||
return True
|
||
continue
|
||
if n.get("subgraph"):
|
||
if _deps_ok(nodes, n)[0]:
|
||
return True
|
||
continue
|
||
v = _dep_verdict(nodes, n)
|
||
if v["verdict"] in ("skip", "block"):
|
||
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(_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
|
||
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
|
||
if dom and _norm_dom((nodes.get(r) or {}).get("domain"))
|
||
and _dom_overlap(dom, _norm_dom((nodes.get(r) or {}).get("domain")))]
|
||
if clash:
|
||
waiting.append({"node": nid, "why": "域与在跑的冲突：%s" % ", ".join(clash)})
|
||
continue
|
||
pf = _pool_full(g, n, [nodes.get(r) or {} for r in running])
|
||
if pf:
|
||
waiting.append({"node": nid, "why": "域池 %s 已满（%d/%d）"
|
||
% (pf["pool"], pf["used"], pf["cap"])})
|
||
continue
|
||
ready.append(nid)
|
||
|
||
slots = max(0, cap - len(running))
|
||
picked, backlog = [], []
|
||
for nid in ready:
|
||
if len(picked) >= slots:
|
||
backlog.append(nid)
|
||
waiting.append({"node": nid, "why": "并发上限 %d 已占满" % cap})
|
||
continue
|
||
dom = _norm_dom((nodes.get(nid) or {}).get("domain"))
|
||
clash = [p for p in picked
|
||
if dom and _norm_dom((nodes.get(p) or {}).get("domain"))
|
||
and _dom_overlap(dom, _norm_dom((nodes.get(p) or {}).get("domain")))]
|
||
if clash:
|
||
# 同批也不能撞域：check 已警告「永不同批」，ready 必须兑现
|
||
waiting.append({"node": nid, "why": "域与本批已选节点重叠：%s" % ", ".join(clash)})
|
||
continue
|
||
pf = _pool_full(g, nodes.get(nid) or {},
|
||
[nodes.get(r) or {} for r in running] +
|
||
[nodes.get(p) or {} for p in picked])
|
||
if pf:
|
||
waiting.append({"node": nid, "why": "域池 %s 已满（%d/%d）"
|
||
% (pf["pool"], pf["used"], pf["cap"])})
|
||
continue
|
||
picked.append(nid)
|
||
|
||
awaiting = [k for k, n in nodes.items() if str(n.get("status")) == "awaiting"]
|
||
return {"gid": g.get("gid"), "running": running, "ready": picked,
|
||
"backlog": backlog, "waiting": waiting, "awaiting": awaiting,
|
||
"pools": dict(_pool_caps(g)),
|
||
"cap": cap, "slots": slots}
|
||
|
||
|
||
def cmd_ready(a) -> int:
|
||
g = _load(a.ws, a.gid)
|
||
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
|
||
|
||
|
||
# ── 状态流转 ────────────────────────────────────────────────────────────
|
||
def cmd_start(a) -> int:
|
||
def _mut(g):
|
||
n = (g.get("nodes") or {}).get(a.node)
|
||
if not n:
|
||
raise SystemExit("没有这个节点：%s" % a.node)
|
||
st = str(n.get("status"))
|
||
if st == "running":
|
||
raise SystemExit("节点 %s 已在跑，不能重复开工" % a.node)
|
||
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"
|
||
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 ""})
|
||
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
|
||
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.pop("not_before", None)
|
||
n.pop("retry_at", None)
|
||
if n.get("approval") and not n.get("approved_at"):
|
||
n["status"] = "awaiting" # 跑完了，等人审过目
|
||
n["awaiting_at"] = _now()
|
||
else:
|
||
n["status"] = "done"
|
||
n["done_at"] = _now()
|
||
|
||
g = _update(a.ws, a.gid, _mut)
|
||
final = _st(g.get("nodes") or {}, a.node)
|
||
_event(a.ws, a.gid, {"node": a.node,
|
||
"to": final if final == "awaiting"
|
||
else (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": 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
|
||
|
||
|
||
# ── 人审 / 取消 / 派活 / 排期 / 进度 ─────────────────────────────────────
|
||
def cmd_approve(a) -> int:
|
||
"""人审节点：`awaiting` ⇒ done（默认）或 blocked（--reject）。"""
|
||
def _mut(g):
|
||
n = (g.get("nodes") or {}).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:
|
||
n["status"] = "blocked"
|
||
n["blocked_by"] = "rejected"
|
||
n["blocked_reason"] = a.reason or "人审驳回"
|
||
n["rejected_at"] = _now()
|
||
else:
|
||
n["status"] = "done"
|
||
n["done_at"] = _now()
|
||
n["approved_at"] = _now()
|
||
n["approved_by"] = a.by or "user"
|
||
|
||
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 ""})
|
||
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"]})
|
||
return 0
|
||
|
||
|
||
def cmd_cancel(a) -> int:
|
||
"""取消节点（连坐它的子节点）；`--all` 取消整图。"""
|
||
if not a.node and not a.all:
|
||
raise SystemExit("要给 --node <节点> 或 --all")
|
||
hit: list = []
|
||
|
||
def _mut(g):
|
||
nodes = g.get("nodes") or {}
|
||
if a.node and a.node not in nodes:
|
||
raise SystemExit("没有这个节点：%s" % a.node)
|
||
stack = [a.node] if a.node else list(nodes)
|
||
while stack:
|
||
nid = stack.pop()
|
||
n = nodes.get(nid)
|
||
if not n or str(n.get("status")) in ("done", "cancelled"):
|
||
continue
|
||
n["status"] = "cancelled"
|
||
n["cancelled_at"] = _now()
|
||
if a.by:
|
||
n["cancelled_by"] = a.by
|
||
if a.reason:
|
||
n["cancel_reason"] = a.reason
|
||
hit.append(nid)
|
||
stack.extend(n.get("children") or [])
|
||
|
||
g = _update(a.ws, a.gid, _mut)
|
||
for nid in hit:
|
||
_event(a.ws, a.gid, {"node": nid, "to": "cancelled",
|
||
"by": a.by or "", "reason": a.reason or ""})
|
||
g, acts = _advance_graph(a.ws, a.gid, g)
|
||
r = _ready(g)
|
||
_out({"ok": True, "cancelled": hit,
|
||
"skipped_by_cancel": [x["node"] for x in acts if x.get("event") == "skip"],
|
||
"next_ready": r["ready"]})
|
||
return 0
|
||
|
||
|
||
def cmd_tick(a) -> int:
|
||
"""自动派活：推进到不动点 → 算就绪集 → 直接把就绪节点开成 running。"""
|
||
started, missed = [], []
|
||
g = _load(a.ws, a.gid)
|
||
g, acts = _advance_graph(a.ws, a.gid, g)
|
||
r = _ready(g)
|
||
cand = list(r["ready"])
|
||
if a.limit:
|
||
cand = cand[:int(a.limit)]
|
||
if cand and not a.dry_run:
|
||
def _mut(gg):
|
||
nodes = gg.get("nodes") or {}
|
||
for nid in cand:
|
||
n = nodes.get(nid)
|
||
if not n or str(n.get("status")) != "pending":
|
||
missed.append(nid)
|
||
continue
|
||
if _dep_verdict(nodes, n)["verdict"] != "go":
|
||
missed.append(nid)
|
||
continue
|
||
now = time.time()
|
||
n["status"] = "running"
|
||
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)
|
||
n["started_by"] = a.by or "tick"
|
||
started.append(nid)
|
||
|
||
g = _update(a.ws, a.gid, _mut)
|
||
for nid in started:
|
||
_event(a.ws, a.gid, {"node": nid, "to": "running", "by": a.by or "tick"})
|
||
g, more = _advance_graph(a.ws, a.gid, g)
|
||
acts.extend(more)
|
||
r2 = _ready(g)
|
||
_out({"gid": a.gid, "started": started, "missed": missed,
|
||
"dry_run": bool(a.dry_run), "advanced": acts,
|
||
"ready_now": r2["ready"], "running": r2["running"],
|
||
"awaiting": r2["awaiting"], "cap": r2["cap"], "slots": r2["slots"]})
|
||
return 0
|
||
|
||
|
||
def cmd_plan(a) -> int:
|
||
"""出一行排期（四条硬规则已按规范形态写好），⛔ 不写宿主库。"""
|
||
ws = str(Path(a.ws).resolve()).replace("\\", "/")
|
||
g = _load(a.ws, a.gid)
|
||
r = _ready(g)
|
||
after = 240 if a.after_sec is None else int(a.after_sec)
|
||
at_ms = int((time.time() + after) * 1000)
|
||
scope = a.scope or (ws.rstrip("/").rsplit("/", 1)[-1] + "/")
|
||
row = {
|
||
"name": "[图模式派活]-%s-%s" % (a.gid, time.strftime("%Y%m%d-%H%M")),
|
||
"prompt": ("跑一次图模式派活：python %s tick --ws %s --gid %s --by tick；"
|
||
"本轮只做这一件事，做完即停。"
|
||
% (Path(__file__).resolve(), ws, a.gid)),
|
||
"cwds": ws,
|
||
"workspace_scope": scope,
|
||
"next_run_at": at_ms,
|
||
"scheduled_at": time.strftime("%Y-%m-%dT%H:%M:%S", time.localtime(at_ms / 1000.0)),
|
||
}
|
||
_out({"ok": True, "gid": a.gid, "rows": [row],
|
||
"ready": r["ready"], "running": r["running"],
|
||
"rules": {"cwds": "正斜杠绝对路径", "workspace_scope": "线名/",
|
||
"next_run_at": "毫秒 epoch", "scheduled_at": "ISO 本地时间"},
|
||
"note": "只产出行，⛔ 不写宿主库（图模式零侵入）"})
|
||
return 0
|
||
|
||
|
||
def _remaining_chain(nodes: dict) -> list:
|
||
"""未完成节点之间的最长依赖链（关键路径）。"""
|
||
todo = {k: v for k, v in nodes.items()
|
||
if str(v.get("status")) not in ("done", "skipped", "cancelled")}
|
||
memo = {}
|
||
|
||
def best(nid):
|
||
if nid in memo:
|
||
return memo[nid]
|
||
memo[nid] = [nid]
|
||
cand = [nid]
|
||
for d in _deps(todo[nid]):
|
||
if d in todo:
|
||
chain = best(d) + [nid]
|
||
if len(chain) > len(cand):
|
||
cand = chain
|
||
memo[nid] = cand
|
||
return cand
|
||
|
||
out: list = []
|
||
for nid in todo:
|
||
c = best(nid)
|
||
if len(c) > len(out):
|
||
out = c
|
||
return out
|
||
|
||
|
||
def cmd_progress(a) -> int:
|
||
g = _load(a.ws, a.gid)
|
||
g, acts = _advance_graph(a.ws, a.gid, g)
|
||
nodes = g.get("nodes") or {}
|
||
cnt: dict = {}
|
||
for n in nodes.values():
|
||
s = str(n.get("status"))
|
||
cnt[s] = cnt.get(s, 0) + 1
|
||
active = len(nodes) - cnt.get("cancelled", 0) - cnt.get("skipped", 0)
|
||
done = cnt.get("done", 0)
|
||
percent = round(100.0 * done / active, 1) if active else 100.0
|
||
r = _ready(g)
|
||
cp = _remaining_chain(nodes)
|
||
_out({"gid": g.get("gid"), "title": g.get("title"), "total": len(nodes),
|
||
"by_status": cnt, "active": active, "done": done, "percent": percent,
|
||
"running": r["running"], "ready": r["ready"], "awaiting": r["awaiting"],
|
||
"blocked": [k for k, n in nodes.items() if str(n.get("status")) == "blocked"],
|
||
"skipped": [k for k, n in nodes.items() if str(n.get("status")) == "skipped"],
|
||
"cancelled": [k for k, n in nodes.items() if str(n.get("status")) == "cancelled"],
|
||
"critical_path": cp, "critical_len": len(cp),
|
||
"advanced": acts})
|
||
return 0
|
||
|
||
|
||
# ── 建图 ────────────────────────────────────────────────────────────────
|
||
def _gid_of(title: str) -> str:
|
||
safe = "".join(c for c in title if c not in '\\/:*?"<>|\r\n\t')[:24].strip()
|
||
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 \
|
||
Path(a.nodes_json).read_text(encoding="utf-8")
|
||
spec = json.loads(raw)
|
||
elif a.kpi:
|
||
spec = {"nodes": [
|
||
{"id": "N%d" % (i + 1), "title": k.strip(), "deps": [],
|
||
"domain": [], "accepts": [k.strip()]}
|
||
for i, k in enumerate([x for x in a.kpi.replace("；", ";").split(";") if x.strip()])
|
||
]}
|
||
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")
|
||
nid = str(raw).strip() if raw not in (None, "") else ""
|
||
if not nid:
|
||
k = i + 1
|
||
while ("N%d" % k) in nodes: # 自动编号要避开已占用的 id
|
||
k += 1
|
||
nid = "N%d" % k
|
||
if nid in nodes: # 显式重复 ⇒ 硬错，⛔ 不静默覆盖前一个
|
||
raise SystemExit("节点 id 重复：%s" % nid)
|
||
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": "", "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"])
|
||
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 n.get("map"):
|
||
raise SystemExit("%s 同时声明了 map 与 subgraph（只能选一个）" % nid)
|
||
sub = {"nodes": [dict(s) for s in sg["nodes"] if isinstance(s, dict)]}
|
||
if sg.get("child_domain"):
|
||
sub["child_domain"] = str(sg["child_domain"])
|
||
nodes[nid]["subgraph"] = sub
|
||
if n.get("when"):
|
||
if not isinstance(n["when"], list):
|
||
raise SystemExit("%s 的 when 必须是列表" % nid)
|
||
clauses = []
|
||
for c in n["when"]:
|
||
if not isinstance(c, dict) or not str(c.get("node") or "").strip():
|
||
raise SystemExit('%s 的 when 子句要写 {"node": "A", "status": "done"}'
|
||
% nid)
|
||
clauses.append({"node": str(c["node"]).strip(),
|
||
"status": str(c.get("status") or "done")})
|
||
nodes[nid]["when"] = clauses
|
||
if n.get("on_dep_fail"):
|
||
if str(n["on_dep_fail"]) not in ON_DEP_FAIL:
|
||
raise SystemExit("%s 的 on_dep_fail 非法：%r（只认 %s）"
|
||
% (nid, n["on_dep_fail"], " / ".join(ON_DEP_FAIL)))
|
||
nodes[nid]["on_dep_fail"] = str(n["on_dep_fail"])
|
||
if n.get("approval"):
|
||
nodes[nid]["approval"] = True
|
||
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),
|
||
"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)
|
||
pools = {}
|
||
if isinstance(spec.get("pools"), dict):
|
||
pools.update(spec["pools"])
|
||
for item in (a.pool or []):
|
||
s = str(item)
|
||
if "=" not in s:
|
||
raise SystemExit("--pool 要写成 域前缀=上限（如 docs/ui/=1）")
|
||
k, v = s.split("=", 1)
|
||
pools[k.strip()] = v
|
||
if pools:
|
||
g["meta"]["pools"] = {}
|
||
for k, v in pools.items():
|
||
dom = _norm_dom(k)
|
||
if not dom:
|
||
raise SystemExit("域池的域不能为空")
|
||
try:
|
||
cap = int(v)
|
||
except (TypeError, ValueError):
|
||
raise SystemExit("域池上限不是整数：%s=%r" % (k, v))
|
||
if cap <= 0:
|
||
raise SystemExit("域池上限必须 ≥1：%s=%r" % (k, v))
|
||
g["meta"]["pools"][dom[0]] = cap
|
||
d.mkdir(parents=True, exist_ok=True)
|
||
_save(a.ws, gid, g)
|
||
_event(a.ws, gid, {"event": "init", "nodes": list(nodes)})
|
||
_out({"ok": True, "gid": gid, "dir": str(d), "nodes": len(nodes),
|
||
"note": "独立图：⛔ 不参与现有机制（不写 goal.json / tasks.json）"})
|
||
return 0
|
||
|
||
|
||
def cmd_show(a) -> int:
|
||
_out(_load(a.ws, a.gid))
|
||
return 0
|
||
|
||
|
||
def cmd_list(a) -> int:
|
||
root = _root(a.ws)
|
||
rows = []
|
||
if root.is_dir():
|
||
for d in sorted(root.iterdir()):
|
||
p = d / "graph.json"
|
||
if not p.is_file():
|
||
continue
|
||
try:
|
||
g = json.loads(p.read_text(encoding="utf-8"))
|
||
except Exception:
|
||
continue
|
||
nodes = g.get("nodes") or {}
|
||
cnt = {}
|
||
for n in nodes.values():
|
||
cnt[str(n.get("status"))] = cnt.get(str(n.get("status")), 0) + 1
|
||
rows.append({"gid": g.get("gid"), "title": g.get("title"),
|
||
"nodes": len(nodes), "by_status": cnt,
|
||
"created_at": g.get("created_at")})
|
||
_out({"ws": str(Path(a.ws).resolve()), "graphs": rows})
|
||
return 0
|
||
|
||
|
||
# ── 入口 ────────────────────────────────────────────────────────────────
|
||
def main(argv=None) -> int:
|
||
ap = argparse.ArgumentParser(prog="graphctl", description="图模式独立实现（不碰现有机制）")
|
||
sub = ap.add_subparsers(dest="cmd", required=True)
|
||
|
||
p = sub.add_parser("init", help="按判据或节点定义建图")
|
||
p.add_argument("--ws", required=True)
|
||
p.add_argument("--title", required=True)
|
||
p.add_argument("--gid", default="")
|
||
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.add_argument("--pool", action="append", default=[],
|
||
help="按域分池：域前缀=并发上限，可重复（如 docs/ui/=1）")
|
||
p.set_defaults(fn=cmd_init)
|
||
|
||
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.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="节点回报")
|
||
p.add_argument("--ws", required=True)
|
||
p.add_argument("--gid", required=True)
|
||
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("approve", help="人审通过／驳回（awaiting ⇒ done／blocked）")
|
||
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("--reject", action="store_true")
|
||
p.add_argument("--reason", default="")
|
||
p.set_defaults(fn=cmd_approve)
|
||
|
||
p = sub.add_parser("cancel", help="取消节点（连坐子节点）或整图")
|
||
p.add_argument("--ws", required=True)
|
||
p.add_argument("--gid", required=True)
|
||
p.add_argument("--node", default="")
|
||
p.add_argument("--all", action="store_true")
|
||
p.add_argument("--by", default="")
|
||
p.add_argument("--reason", default="")
|
||
p.set_defaults(fn=cmd_cancel)
|
||
|
||
p = sub.add_parser("tick", help="自动派活：算就绪集并直接开工")
|
||
p.add_argument("--ws", required=True)
|
||
p.add_argument("--gid", required=True)
|
||
p.add_argument("--by", default="tick")
|
||
p.add_argument("--limit", type=int, default=0, help="本次最多开几个（默认不额外限）")
|
||
p.add_argument("--dry-run", action="store_true", help="只算不开")
|
||
p.add_argument("--lease-sec", type=float, default=None)
|
||
p.set_defaults(fn=cmd_tick)
|
||
|
||
p = sub.add_parser("plan", help="出一行排期（四条硬规则已规范化）")
|
||
p.add_argument("--ws", required=True)
|
||
p.add_argument("--gid", required=True)
|
||
p.add_argument("--scope", default="", help="workspace_scope，默认取工作区目录名")
|
||
p.add_argument("--after-sec", type=int, default=None, help="多久后触发（默认 240 秒）")
|
||
p.set_defaults(fn=cmd_plan)
|
||
|
||
p = sub.add_parser("progress", help="进度与关键路径")
|
||
p.add_argument("--ws", required=True)
|
||
p.add_argument("--gid", required=True)
|
||
p.set_defaults(fn=cmd_progress)
|
||
|
||
p = sub.add_parser("list")
|
||
p.add_argument("--ws", required=True)
|
||
p.set_defaults(fn=cmd_list)
|
||
|
||
a = ap.parse_args(argv)
|
||
return a.fn(a)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|