From 41752fd1f3a07067d095ab18d2180c28b7889084 Mon Sep 17 00:00:00 2001 From: maogeigei Date: Sun, 18 Oct 2026 22:28:28 +0800 Subject: [PATCH] =?UTF-8?q?feat(session-mechanism):=20=E6=96=B0=E5=A2=9E?= =?UTF-8?q?=E7=8B=AC=E7=AB=8B=E7=9A=84=E3=80=8C=E5=9B=BE=E6=A8=A1=E5=BC=8F?= =?UTF-8?q?=E3=80=8D=E6=A8=A1=E5=9D=97=EF=BC=88=E4=B8=8D=E7=A2=B0=E7=8E=B0?= =?UTF-8?q?=E6=9C=89=E6=9C=BA=E5=88=B6=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - graph/graphctl.py：建图 / 校验 / 算就绪集 / 节点状态流转，一条命令一个动作 就绪四条件：依赖全 done、自己 pending、域不与 running 重叠、不超 max_parallel - graph/README.md：用法与边界，明确与 goal.json / tasks.json / 六道闸完全并列 - 实测跑通（4 节点含依赖与域）：init → check（域重叠告警命中）→ ready（A、B） → start A → report A done → next_ready（B、C） - 不带常驻进程、不注册钩子、不改任何现有文件；停用即不调用 --- session-mechanism/graph/README.md | 90 +++++++ session-mechanism/graph/graphctl.py | 391 ++++++++++++++++++++++++++++ 2 files changed, 481 insertions(+) create mode 100644 session-mechanism/graph/README.md create mode 100644 session-mechanism/graph/graphctl.py diff --git a/session-mechanism/graph/README.md b/session-mechanism/graph/README.md new file mode 100644 index 0000000..0c895eb --- /dev/null +++ b/session-mechanism/graph/README.md @@ -0,0 +1,90 @@ +# 图模式（独立功能） + +> 2026-10-18 新增。**这一套与现有「项目机制」完全并列**：不读写 `goal.json`、`tasks.json`、`collabd-state.json`，不参与六道闸，不接管常驻循环，不改变现有任何会话的行为。 +> 停用方式：不调用它就行；它没有常驻进程、没有注册任何钩子、没有改任何现有文件。 + +## 一、它解决什么 + +现有机制是「一棒干一堆、由会话读摘要决定下一棒」。图模式把这层换成： + +- **节点**：一块独立的活，带自己的**域**（允许写哪些目录）和**判据**。 +- **边**：节点之间的依赖；依赖全 done 才放行。 +- **状态**：写在一份图上，跨会话共享，不依赖任何会话的上下文。 + +于是可以做到：**一批就绪节点同时干、干完各自回报、下一批自动放行**。 + +## 二、怎么用 + +工作区参数 `--ws` 指向要被推进的工作区根目录。图落在 `/.workbuddy/graph//`。 + +### 1、建图 + +按判据建（每条判据一个节点，默认无依赖、无域）： + + python graph/graphctl.py init --ws "" --title "目标名" --kpi "判据A;判据B;判据C" + +按节点定义建（要并行、要依赖、要域，就用这个）： + + python graph/graphctl.py init --ws "" --title "目标名" --nodes-json - <<'JSON' + { + "max_parallel": 3, + "nodes": [ + {"id": "A", "title": "改界面", "domain": ["ui/"], "accepts": ["页面能打开"], "deps": []}, + {"id": "B", "title": "改文档", "domain": ["docs/"], "accepts": ["索引已登记"], "deps": []}, + {"id": "C", "title": "改服务", "domain": ["server/"], "accepts": ["本地能起"], "deps": ["A"]}, + {"id": "D", "title": "汇总", "domain": ["docs/index/"], "accepts": ["三处齐"], "deps": ["A","B","C"]} + ] + } + JSON + +### 2、校验 + + python graph/graphctl.py check --ws "" --gid "" + +报三件硬错（环、依赖指向不存在的节点、状态非法）和三类警告（没有域、没有判据、两个节点域重叠）。 + +### 3、算这一批派谁 + + python graph/graphctl.py ready --ws "" --gid "" + +输出 `ready`（这一批可以同时开工的节点）、`backlog`（就绪但被并发上限挡住）、`waiting`（还差什么）。 + +### 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" + +回报 `done` 后会立刻返回下一批的 `ready`，这就是图在往前走的信号。 + +### 5、看状态 + + python graph/graphctl.py list --ws "" + python graph/graphctl.py show --ws "" --gid "" + +## 三、就绪的四个条件 + +1. 依赖全部 `done`。 +2. 节点自己是 `pending`。 +3. 域不与任何 `running` 节点重叠（`docs/` 与 `docs/ui/` 算重叠）。 +4. 加上在跑的节点后，不超过 `max_parallel`（默认 3）。 + +## 四、与现有机制的关系 + +| 事项 | 现有机制 | 图模式 | +|---|---|---| +| 目标定义 | `goal.json` | 图文件里的 `title` / `nodes` | +| 任务台账 | `tasks.json` | 图里每个节点的 `status` + `events.jsonl` | +| 派活决策 | 检查会话读摘要判断 | `ready` 按图算 | +| 触发 | 常驻 + 六道闸 + 静默 2 分钟 | 调用方驱动（人或会话），无后台进程 | +| 会话创建 | 自动化排期 | 由调用方决定（图不管怎么开会话） | +| 域锁 | 开工时声明 | 同左，节点的 `domain` 直接喂给开工命令 | + +**两套可以同时存在**：图模式不读也不改现有那三份文件，现有机制也永远不会看到 `graph/` 目录。 + +## 五、还没做的（下一步） + +- 自动派活：读 `ready` 后自动建排期（要复用现有排期硬规则：cwds 正斜杠、workspace_scope、next_run_at 毫秒、scheduled_at ISO）。 +- 判定节点完成：现在靠回报 `--state done`，还没有「看证据够不够」的自动判定。 +- 回退边：`attempts` 已经在记，还没有「判不过回到哪个节点」的字段与规则。 + +这三件都只在本目录内加，不影响现有机制。 diff --git a/session-mechanism/graph/graphctl.py b/session-mechanism/graph/graphctl.py new file mode 100644 index 0000000..7fe03d8 --- /dev/null +++ b/session-mechanism/graph/graphctl.py @@ -0,0 +1,391 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""graphctl.py —— 「图模式」的独立实现（2026-10-18）。 + +## 它是什么 + +把目标拆成节点、按依赖推进的**独立**运行模式。**⛔ 不修改任何现有机制**： +不读也不写 `goal.json` / `tasks.json` / `collabd-state.json`，不参与六道闸， +不接管常驻循环，不改变现有会话的任何行为。 + +## 数据落点（独立目录，与现有状态完全分开） + + /.workbuddy/graph//graph.json 图的唯一真源 + /.workbuddy/graph//events.jsonl 每次状态变更的流水 + +## 图的结构 + + { + "gid": "20261018-2230-ui改造", + "title": "给三块功能各加一份文档", + "created_at": "2026-10-18T22:30:00", + "meta": {"max_parallel": 3}, + "nodes": { + "N1": {"title": "...", "deps": [], "domain": ["docs/ui/"], + "accepts": ["产物在位"], "status": "pending", + "evidence": "", "attempts": 0} + } + } + +`status` 四态：`pending` → `running` → `done` ／ `blocked`。 + +## 命令 + + 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 "..."] + show --ws --gid + list --ws + +## 就绪的判据（四条全中） + + ① 依赖全部 done ② 自己还是 pending + ③ 域不与任何 running 节点重叠 ④ 不超 max_parallel + +## 域的用法 + +`domain` 是一组路径前缀（相对工作区根）。两节点域重叠 ⇒ 不能同批跑。 +判定：任一前缀互为前缀即视为重叠（`docs/` 与 `docs/ui/` 算重叠）。 +""" +from __future__ import annotations + +import argparse +import io +import json +import os +import sys +import time +from pathlib import Path + +MARK = ".workbuddy/graph" +STATUSES = ("pending", "running", "done", "blocked") + + +# ── 基础设施 ──────────────────────────────────────────────────────────── +def _now() -> str: + return time.strftime("%Y-%m-%dT%H:%M:%S") + + +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: + p = _gdir(ws, gid) / "graph.json" + tmp = p.with_suffix(".json.tmp") + tmp.write_text(json.dumps(g, ensure_ascii=False, indent=2) + "\n", + encoding="utf-8", newline="\n") + os.replace(tmp, p) # 原子替换，避免半截文件 + + +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 _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 not _norm_dom(n.get("domain")): + 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 []): + warns.append("%s 没有验收判据 ⇒ 判不出完成" % nid) + + 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 _ready(g: dict) -> dict: + nodes = g.get("nodes") or {} + cap = int((g.get("meta") or {}).get("max_parallel") or 3) + running = [k for k, n in nodes.items() if str(n.get("status")) == "running"] + ready, waiting = [], [] + + for nid, n in nodes.items(): + st = str(n.get("status")) + 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)}) + 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 + ready.append(nid) + + slots = max(0, cap - len(running)) + picked = ready[:slots] + for nid in ready[slots:]: + waiting.append({"node": nid, "why": "并发上限 %d 已占满" % cap}) + + return {"gid": g.get("gid"), "running": running, "ready": picked, + "backlog": [x for x in ready[slots:]], "waiting": waiting, + "cap": cap, "slots": slots} + + +def cmd_ready(a) -> int: + g = _load(a.ws, a.gid) + _out(_ready(g)) + return 0 + + +# ── 状态流转 ──────────────────────────────────────────────────────────── +def cmd_start(a) -> int: + g = _load(a.ws, a.gid) + n = (g.get("nodes") or {}).get(a.node) + if not n: + raise SystemExit("没有这个节点：%s" % a.node) + if str(n.get("status")) != "pending": + raise SystemExit("节点 %s 当前是 %s，不能开工" % (a.node, n.get("status"))) + n["status"] = "running" + _save(a.ws, a.gid, g) + _event(a.ws, a.gid, {"node": a.node, "to": "running", "by": a.by or ""}) + _out({"ok": True, "node": a.node, "status": "running"}) + return 0 + + +def cmd_report(a) -> int: + g = _load(a.ws, a.gid) + n = (g.get("nodes") or {}).get(a.node) + if not n: + raise SystemExit("没有这个节点：%s" % a.node) + if a.state not in ("done", "blocked"): + raise SystemExit("--state 只认 done / blocked") + if a.evidence: + n["evidence"] = a.evidence + n["status"] = a.state + if a.state == "blocked": + n["attempts"] = int(n.get("attempts") or 0) + 1 + _save(a.ws, a.gid, g) + _event(a.ws, a.gid, {"node": a.node, "to": a.state, + "evidence": a.evidence or "", "by": a.by or ""}) + r = _ready(g) + _out({"ok": True, "node": a.node, "status": a.state, + "next_ready": r["ready"]}) + 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 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") + + nodes = {} + for i, n in enumerate(spec.get("nodes") or []): + nid = str(n.get("id") or "N%d" % (i + 1)) + 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, + } + 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)}, + "nodes": nodes} + d = _gdir(a.ws, gid) + 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.set_defaults(fn=cmd_init) + + for name, fn, extra in (("check", cmd_check, None), ("ready", cmd_ready, None), + ("show", cmd_show, None)): + 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.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("--by", default="") + p.set_defaults(fn=cmd_report) + + 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())