feat(viz): read-only spec → epic → task graph UI (onecoder viz)

README.md

@@ -15,12 +15,27 @@ python3 -m unittest discover -s tests
 
 ---
 
+## Visual work graph
+
+`onecoder viz` serves a read-only page at `http://127.0.0.1:8765/` that shows the project as
+**specification → epic → task → acceptance criterion**:
+
+* spec: `specs/<name>.md` with frontmatter `id`, `title`, `status`, and `epic: <sprint-id>` (in git)
+* epic: sprint · task: issue · task: governance task (`.onecoder/tasks/<id>`), with its `AC-n` lines
+
+`onecoder viz --print-json` prints the same graph for scripts and agents.
+
+Storage: CSV stays the write path and the fallback. `.onecoder/track.db` is a derived SQLite index,
+rebuilt whenever the CSV, specs, or governance files change. It ignores itself through
+`.onecoder/.gitignore`, so it is never committed. If the index cannot be written, the page is served
+from CSV and shows a warning.
+
 ## Tracking & Agile Governance (CLI + MCP Dual Mode)
 
 OneCoder unifies **macro project/sprint tracking** with **micro task execution**. It operates in two synchronized modes:
 
 1. **Local CLI Mode (Zero-Dependency)**:
-   * 100% Python standard library (built-in SQLite `~/.onecoder/track.db`).
+   * 100% Python standard library. Local data is CSV in `.onecoder/track/`, with a derived SQLite index in `.onecoder/track.db` for `onecoder viz`.
    * No cloud credentials, no external servers, and zero pip packages required.
    * Works completely offline with full sprint, issue, session, and radar capabilities.
 

onecoder/cli.py

@@ -359,6 +359,14 @@ def _add_mode_flags(parser):
     _add_mode_flags(s)
     s.set_defaults(fn=track_cli.cmd_session)
 
+    # Visual work graph (read-only UI over the tracking data)
+    from .viz import cli as viz_cli
+    s = sub.add_parser("viz", help="Serve the spec -> epic -> task graph as a read-only UI")
+    s.add_argument("--host", default="127.0.0.1", help="Bind address (loopback by default)")
+    s.add_argument("--port", type=int, default=8765)
+    s.add_argument("--print-json", action="store_true", help="Print the graph as JSON and exit")
+    s.set_defaults(fn=viz_cli.cmd_viz)
+
     # MCP server command
     s = sub.add_parser("mcp", help="Run MCP stdio server")
     s.add_argument("--backend", choices=["auto", "local", "remote"], default="auto", help="Storage backend to use")

onecoder/track/local.py

@@ -1,6 +1,7 @@
 """LocalTrackStore: Zero-dependency CSV-backed tracking engine."""
 
 import csv
+import uuid
 import json
 import os
 import time
@@ -267,7 +268,7 @@ def validate_workstations(self, workstation: Optional[str] = None) -> Dict[str,
     def create_sprint(self, project: str, name: str, status: str = "planned",
                       starts_at: Optional[str] = None, ends_at: Optional[str] = None,
                       id: Optional[str] = None) -> Dict[str, Any]:
-        sprint_id = id or f"sprint-{slugify(name)}-{int(time.time())}"
+        sprint_id = id or f"sprint-{slugify(name)}-{uuid.uuid4().hex[:8]}"
         now = _now_iso()
         sprint = {
             "id": sprint_id,
@@ -313,7 +314,8 @@ def create_issue(self, project: str, title: str, sprint: Optional[str] = None,
                      status: str = "open", labels: Optional[List[str]] = None,
                      id: Optional[str] = None) -> Dict[str, Any]:
         now = _now_iso()
-        issue_id = id or f"issue-{int(time.time()*1000) % 1000000}"
+        # Random, not time-based: two issues created in the same millisecond used to share an id.
+        issue_id = id or f"issue-{uuid.uuid4().hex[:8]}"
         labels_str = json.dumps(labels or [])
         issue = {
             "id": issue_id,

onecoder/viz/__init__.py

@@ -0,0 +1 @@
+"""Visual work graph: spec -> epic -> task -> acceptance criterion, served as a read-only UI."""

onecoder/viz/cli.py

@@ -0,0 +1,24 @@
+"""`onecoder viz`: serve the project's work graph, or print it as JSON."""
+
+import json
+
+from ..track import cli as track_cli
+from .server import make_server
+from .snapshot import snapshot
+
+
+def cmd_viz(a):
+    root = track_cli._get_root()
+    if a.print_json:
+        print(json.dumps(snapshot(root), indent=2))
+        return 0
+    server = make_server(root, host=a.host, port=a.port)
+    url = f"http://{a.host}:{server.server_address[1]}/"
+    print(f"onecoder viz: {root.name} at {url}  (Ctrl-C to stop)")
+    try:
+        server.serve_forever()
+    except KeyboardInterrupt:
+        pass
+    finally:
+        server.server_close()
+    return 0

onecoder/viz/model.py

@@ -0,0 +1,129 @@
+"""Build the work graph from tracking data, spec files, and governance tasks.
+
+Mapping onto the existing model (no schema change to the CSV store):
+  spec  = specs/<name>.md (frontmatter: id, title, status, epic)   [in git, reviewed]
+  epic  = sprint                                                 [tracking CSV]
+  task  = issue                                                  [tracking CSV]
+  task  = governance task .onecoder/tasks/<id>                   [state.json, review.json]
+  ac    = "- [ ] AC-n: ..." line in a governance task's brief.md
+"""
+
+import json
+import re
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+from .. import policy as pol
+from ..task import list_tasks, tasks_dir
+from ..track.local import LocalTrackStore
+
+SPEC_DIR = "specs"
+AC_RE = re.compile(r"^\s*-\s*\[( |x|X)\]\s*(AC-\d+)\s*:\s*(.+?)\s*$", re.M)
+UNPLANNED = "group:unplanned"
+GOVERNANCE = "group:governance"
+
+
+def parse_frontmatter(text: str) -> Dict[str, str]:
+    """Flat `key: value` frontmatter. A trailing `  # comment` is dropped."""
+    m = re.match(r"^---\n(.*?)\n---", text, re.S)
+    if not m:
+        return {}
+    out: Dict[str, str] = {}
+    for line in m.group(1).splitlines():
+        if ":" not in line or line.lstrip().startswith("#"):
+            continue
+        key, value = line.split(":", 1)
+        out[key.strip()] = re.split(r"\s+#", value, maxsplit=1)[0].strip()
+    return out
+
+
+def _node(id: str, kind: str, title: str, status: str, parent: Optional[str], source: str,
+          data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
+    return {"id": id, "kind": kind, "title": title, "status": status, "parent": parent,
+            "source": source, "data": data or {}}
+
+
+def _load_specs(root: Path, nodes: Dict[str, Dict[str, Any]], epic_parent: Dict[str, str]) -> None:
+    for path in sorted((root / SPEC_DIR).glob("*.md")) if (root / SPEC_DIR).is_dir() else []:
+        text = path.read_text(encoding="utf-8")
+        meta = parse_frontmatter(text)
+        sid = meta.get("id") or path.stem
+        title = meta.get("title") or next((l.lstrip("# ").strip() for l in text.splitlines() if l.startswith("#")), path.stem)
+        nodes[f"spec:{sid}"] = _node(f"spec:{sid}", "spec", title, meta.get("status", "draft"), None, "specs",
+                                    {"path": str(path.relative_to(root))})
+        if meta.get("epic"):
+            epic_parent.setdefault(meta["epic"], f"spec:{sid}")
+
+
+def _load_tracking(store: LocalTrackStore, nodes: Dict[str, Dict[str, Any]], epic_parent: Dict[str, str]) -> None:
+    for s in store.list_sprints().get("sprints", []):
+        eid = f"epic:{s['id']}"
+        parent = epic_parent.get(s["id"])
+        nodes[eid] = _node(eid, "epic", s.get("name") or s["id"], s.get("status") or "planned", parent, "tracking",
+                           {"sprint": s["id"], "project": s.get("project", "")})
+    for i in store.list_issues().get("issues", []):
+        tid = f"task:{i['id']}"
+        parent = f"epic:{i['sprint']}" if i.get("sprint") else UNPLANNED
+        nodes[tid] = _node(tid, "task", i.get("title") or i["id"], i.get("status") or "open", parent, "tracking",
+                           {"issue": i["id"], "labels": i.get("labels", [])})
+
+
+def _load_governance(root: Path, nodes: Dict[str, Dict[str, Any]]) -> None:
+    for state in list_tasks(root):
+        gid = f"gov:{state['id']}"
+        verdict_path = tasks_dir(root) / state["id"] / "review.json"
+        status = "open"
+        if verdict_path.exists():
+            try:
+                status = json.loads(verdict_path.read_text()).get("verdict", "reviewed")
+            except (ValueError, AttributeError):
+                status = "reviewed"
+        issue_parent = f"task:{state['issue']}" if state.get("issue") and f"task:{state['issue']}" in nodes else GOVERNANCE
+        nodes[gid] = _node(gid, "task", state["id"], status, issue_parent, "governance",
+                           {"mode": state.get("mode"), "isolation": state.get("isolation"), "branch": state.get("branch")})
+        brief = tasks_dir(root) / state["id"] / "brief.md"
+        if brief.exists():
+            for m in AC_RE.finditer(brief.read_text(encoding="utf-8")):
+                ac = f"ac:{state['id']}:{m.group(2)}"
+                nodes[ac] = _node(ac, "ac", f"{m.group(2)}: {m.group(3)}", "done" if m.group(1).lower() == "x" else "open",
+                                  gid, "governance")
+
+
+def _ensure_groups(nodes: Dict[str, Dict[str, Any]]) -> None:
+    """Add group nodes only when something lands in them, so the tree has no empty buckets."""
+    for gid, title in ((UNPLANNED, "Unplanned tasks"), (GOVERNANCE, "Governance tasks")):
+        if any(n["parent"] == gid for n in nodes.values()):
+            nodes[gid] = _node(gid, "group", title, "", None, "derived")
+
+
+def build_graph(root: Path, store: Optional[LocalTrackStore] = None) -> Dict[str, Any]:
+    """Pure read of the project. Always from CSV and files, never from SQLite."""
+    store = store or LocalTrackStore(root=root, policy=pol.load(root))
+    nodes: Dict[str, Dict[str, Any]] = {}
+    epic_parent: Dict[str, str] = {}
+    _load_specs(root, nodes, epic_parent)
+    _load_tracking(store, nodes, epic_parent)
+    _load_governance(root, nodes)
+    _ensure_groups(nodes)
+    # A spec-linked epic may point at a spec that does not exist. Detach it rather than dangle.
+    for n in nodes.values():
+        if n["parent"] and n["parent"] not in nodes:
+            n["parent"] = None
+    return {
+        "project": root.name,
+        "generated_at": datetime.now(timezone.utc).isoformat(timespec="seconds"),
+        "source": "csv",
+        "nodes": sorted(nodes.values(), key=lambda n: (n["kind"], n["id"])),
+        "counts": _counts(nodes.values()),
+    }
+
+
+def _counts(nodes) -> Dict[str, Dict[str, int]]:
+    out: Dict[str, Dict[str, int]] = {}
+    for n in nodes:
+        if n["kind"] == "group":
+            continue
+        bucket = out.setdefault(n["kind"], {})
+        bucket[n["status"]] = bucket.get(n["status"], 0) + 1
+    return out