97 lines
4.2 KiB
Python
97 lines
4.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Deterministic flow report for the NetKingdom security estate.
|
|
|
|
Replaces agent sessions that poll wait tasks. Reads workplan files only (no hub,
|
|
no network) and prints four lists:
|
|
|
|
READY todo/progress tasks in non-residual workplans: what agents should work on
|
|
UNBLOCKED wait tasks whose every depends_on target is done/cancel: requalify to todo
|
|
ATTENDED needs_human wait tasks: the attended-window checklist for the operator/founder
|
|
STALLED wait tasks qualified only by blocked_on message-from (owner reply), oldest first by file
|
|
|
|
Usage: security_flow_report.py [--repos a,b,c] [--json]
|
|
"""
|
|
import argparse, glob, json, os, re
|
|
|
|
DEFAULT_REPOS = ("the-custodian net-kingdom railiance-platform activity-core access-engine secrets-engine ops-mason "
|
|
"ops-warden key-cape user-engine tenant-engine gate-house kings-guard "
|
|
"policy-nexus rapp-policy-nexus").split()
|
|
TERMINAL = {"done", "cancel"}
|
|
|
|
|
|
def field(block, key):
|
|
m = re.search(rf"^{key}:\s*(.*)$", block, re.M)
|
|
return m.group(1).strip().strip('"') if m else ""
|
|
|
|
|
|
def load(root, repos):
|
|
tasks, wps = {}, {}
|
|
for r in repos:
|
|
for f in glob.glob(os.path.join(root, r, "workplans", "**", "*.md"), recursive=True):
|
|
text = open(f, encoding="utf-8", errors="ignore").read()
|
|
fm = text.split("---")[1] if text.startswith("---") else ""
|
|
wp = {"repo": r, "file": os.path.relpath(f, root),
|
|
"status": field(fm, "status"), "flavor": field(fm, "flavor"),
|
|
"title": field(fm, "title")}
|
|
for b in re.findall(r"```task\n(.*?)```", text, re.S):
|
|
tid = field(b, "id")
|
|
if not tid:
|
|
continue
|
|
deps = re.findall(r"[A-Z][A-Z0-9-]+-T\d+|[A-Z][A-Z0-9-]+-WP-\d+", field(b, "depends_on"))
|
|
tasks[tid] = {"id": tid, "status": field(b, "status"), "deps": deps,
|
|
"human": field(b, "needs_human") == "true",
|
|
"blocked_on": field(b, "blocked_on"),
|
|
"reason": field(b, "blocking_reason") or field(b, "intervention_note"),
|
|
"priority": field(b, "priority"), "wp": wp}
|
|
return tasks
|
|
|
|
|
|
def classify(tasks):
|
|
def dep_done(d):
|
|
if d in tasks:
|
|
return tasks[d]["status"] in TERMINAL
|
|
# workplan-level dependency: every known task of that workplan is terminal
|
|
own = [t for t in tasks.values() if t["id"].startswith(d + "-T")]
|
|
return bool(own) and all(t["status"] in TERMINAL for t in own)
|
|
|
|
out = {"READY": [], "UNBLOCKED": [], "ATTENDED": [], "STALLED": []}
|
|
for t in tasks.values():
|
|
wp = t["wp"]
|
|
if wp["status"] in ("finished", "archived") or wp["flavor"] == "residual":
|
|
continue
|
|
if t["status"] in ("todo", "progress"):
|
|
out["READY"].append(t)
|
|
elif t["status"] == "wait":
|
|
if t["human"]:
|
|
out["ATTENDED"].append(t)
|
|
elif t["deps"] and all(dep_done(d) for d in t["deps"]):
|
|
out["UNBLOCKED"].append(t)
|
|
elif not t["deps"] and t["blocked_on"].startswith("message-from"):
|
|
out["STALLED"].append(t)
|
|
prio = {"high": 0, "medium": 1, "low": 2}
|
|
for k in out:
|
|
out[k].sort(key=lambda t: (prio.get(t["priority"], 3), t["id"]))
|
|
return out
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--root", default=os.path.expanduser("~"))
|
|
ap.add_argument("--repos", default=",".join(DEFAULT_REPOS))
|
|
ap.add_argument("--json", action="store_true")
|
|
a = ap.parse_args()
|
|
res = classify(load(a.root, a.repos.split(",")))
|
|
if a.json:
|
|
print(json.dumps({k: [{x: t[x] for x in ("id", "status", "priority", "reason", "blocked_on")}
|
|
| {"repo": t["wp"]["repo"]} for t in v] for k, v in res.items()}, indent=1))
|
|
return
|
|
for k, v in res.items():
|
|
print(f"\n## {k} ({len(v)})")
|
|
for t in v:
|
|
extra = t["blocked_on"] if k == "STALLED" else t["reason"]
|
|
print(f"- {t['id']} [{t['status']}/{t['priority'] or '-'}] {t['wp']['repo']}"
|
|
+ (f" — {extra[:140]}" if extra else ""))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|