workbuddy_skills/session-mechanism/graph/graphctl.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""graphctl.py —— 「图模式」的独立实现（2026-10-18）。

## 它是什么

把目标拆成节点、按依赖推进的**独立**运行模式。**⛔ 不修改任何现有机制**：
不读也不写 `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},
      "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 <ws> --gid <gid>          环／悬空依赖／空域／同批域冲突
    ready  --ws <ws> --gid <gid>          就绪集（含为什么没被选）
    start  --ws <ws> --gid <gid> --node N1
    report --ws <ws> --gid <gid> --node N1 --state done|blocked [--evidence "..."]
    show   --ws <ws> --gid <gid>
    list   --ws <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())