From a9944def65f811c5968fae7e21169e65c4890f15 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 22 Oct 2026 00:21:08 +0000 Subject: [PATCH] feat(viz): add `onecoder viz`, a read-only spec -> epic -> task graph Adds a stdlib-only UI over the tracking data and spec files. - Graph: spec (specs/*.md) -> epic (sprint) -> task (issue) and governance task -> acceptance criterion (AC-n from brief.md). Unplanned work and governance tasks get group nodes. No dangling parents. - Storage: CSV stays the write path and the fallback. `.onecoder/track.db` is a derived SQLite index, rebuilt when CSV, specs, or governance files change. It ignores itself via `.onecoder/.gitignore`. If the index can't be written, the page serves from CSV and shows a warning. - Server: loopback by default, strict CSP, no external resources. The page builds the DOM with textContent only. - `--print-json` prints the graph for scripts and agents. Fix: issue and sprint ids were time-based (ms modulo 1e6). Issues created in the same millisecond shared an id, so updates hit all of them and the graph dropped the duplicates. Ids are now random. Tests: 61 pass (49 existing, 12 new covering hierarchy, index invalidation, CSV fallback, ignore rules, HTTP surface, CLI, and id collisions). Claude-Session: https://claude.ai/code/session_01NGJmcc1EdTFt3e6CJfF3E9 --- README.md | 17 ++- onecoder/cli.py | 8 ++ onecoder/track/local.py | 6 +- onecoder/viz/__init__.py | 1 + onecoder/viz/cli.py | 24 +++++ onecoder/viz/model.py | 129 ++++++++++++++++++++++ onecoder/viz/page.html | 223 +++++++++++++++++++++++++++++++++++++++ onecoder/viz/server.py | 42 ++++++++ onecoder/viz/snapshot.py | 25 +++++ onecoder/viz/store.py | 98 +++++++++++++++++ pyproject.toml | 3 + tests/test_viz.py | 196 ++++++++++++++++++++++++++++++++++ 12 files changed, 769 insertions(+), 3 deletions(-) create mode 100644 onecoder/viz/__init__.py create mode 100644 onecoder/viz/cli.py create mode 100644 onecoder/viz/model.py create mode 100644 onecoder/viz/page.html create mode 100644 onecoder/viz/server.py create mode 100644 onecoder/viz/snapshot.py create mode 100644 onecoder/viz/store.py create mode 100644 tests/test_viz.py diff --git a/README.md b/README.md index 5359a05..0f2fd2a 100644 --- a/README.md +++ b/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/.md` with frontmatter `id`, `title`, `status`, and `epic: ` (in git) +* epic: sprint · task: issue · task: governance task (`.onecoder/tasks/`), 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. diff --git a/onecoder/cli.py b/onecoder/cli.py index 664394c..f8a8dc1 100755 --- a/onecoder/cli.py +++ b/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") diff --git a/onecoder/track/local.py b/onecoder/track/local.py index 8692921..a61b078 100644 --- a/onecoder/track/local.py +++ b/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, diff --git a/onecoder/viz/__init__.py b/onecoder/viz/__init__.py new file mode 100644 index 0000000..b5bc260 --- /dev/null +++ b/onecoder/viz/__init__.py @@ -0,0 +1 @@ +"""Visual work graph: spec -> epic -> task -> acceptance criterion, served as a read-only UI.""" diff --git a/onecoder/viz/cli.py b/onecoder/viz/cli.py new file mode 100644 index 0000000..e6efde1 --- /dev/null +++ b/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 diff --git a/onecoder/viz/model.py b/onecoder/viz/model.py new file mode 100644 index 0000000..8affee8 --- /dev/null +++ b/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/.md (frontmatter: id, title, status, epic) [in git, reviewed] + epic = sprint [tracking CSV] + task = issue [tracking CSV] + task = governance task .onecoder/tasks/ [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 diff --git a/onecoder/viz/page.html b/onecoder/viz/page.html new file mode 100644 index 0000000..db3dbd6 --- /dev/null +++ b/onecoder/viz/page.html @@ -0,0 +1,223 @@ + + + + + +OneCoder: work graph + + + +
+

OneCoder

+ … + + +
+ +
+
+

Specification → epic → task → criterion

+
+
+
+
+

Details

+
Select a node.
+
+
+ + + diff --git a/onecoder/viz/server.py b/onecoder/viz/server.py new file mode 100644 index 0000000..2ffa4e3 --- /dev/null +++ b/onecoder/viz/server.py @@ -0,0 +1,42 @@ +"""Read-only HTTP server for the work graph. Stdlib only. Binds loopback by default.""" + +import json +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path + +from .snapshot import snapshot + +PAGE = Path(__file__).with_name("page.html") +CSP = "default-src 'none'; script-src 'unsafe-inline'; style-src 'unsafe-inline'; connect-src 'self'; base-uri 'none'; frame-ancestors 'none'" + + +class _Handler(BaseHTTPRequestHandler): + root: Path = Path(".") + + def do_GET(self): # noqa: N802 (stdlib naming) + path = self.path.split("?", 1)[0] + if path == "/": + self._send(200, "text/html; charset=utf-8", PAGE.read_bytes()) + elif path == "/api/graph": + body = json.dumps(snapshot(self.root), indent=2).encode() + self._send(200, "application/json", body, cache="no-store") + else: + self._send(404, "text/plain; charset=utf-8", b"not found") + + def _send(self, status, ctype, body, cache="no-cache"): + self.send_response(status) + self.send_header("Content-Type", ctype) + self.send_header("Content-Length", str(len(body))) + self.send_header("Content-Security-Policy", CSP) + self.send_header("X-Content-Type-Options", "nosniff") + self.send_header("Cache-Control", cache) + self.end_headers() + self.wfile.write(body) + + def log_message(self, fmt, *args): # keep the terminal quiet + return + + +def make_server(root: Path, host: str = "127.0.0.1", port: int = 8765) -> ThreadingHTTPServer: + handler = type("VizHandler", (_Handler,), {"root": root}) + return ThreadingHTTPServer((host, port), handler) diff --git a/onecoder/viz/snapshot.py b/onecoder/viz/snapshot.py new file mode 100644 index 0000000..d0995a1 --- /dev/null +++ b/onecoder/viz/snapshot.py @@ -0,0 +1,25 @@ +"""One entry point for the UI: SQLite when it is current, rebuilt when it is not, CSV when it cannot be used.""" + +import sqlite3 +from pathlib import Path +from typing import Any, Dict + +from . import store +from .model import build_graph + + +def snapshot(root: Path) -> Dict[str, Any]: + try: + fp = store.fingerprint(root) + cached = store.read(root, fp) + if cached is not None: + return cached + graph = build_graph(root) + store.write(root, graph, fp) + graph["source"] = "sqlite" + return graph + except (OSError, sqlite3.Error) as e: + graph = build_graph(root) + graph["source"] = "csv" + graph["warning"] = f"sqlite index unavailable, serving from CSV: {e}" + return graph diff --git a/onecoder/viz/store.py b/onecoder/viz/store.py new file mode 100644 index 0000000..f1ce557 --- /dev/null +++ b/onecoder/viz/store.py @@ -0,0 +1,98 @@ +"""Per-project SQLite index for the visual graph: `.onecoder/track.db`. + +CSV (or the spec and governance files) is the source of truth. This database is derived, +gitignored, and rebuilt whenever its inputs change. If it cannot be written or read, the caller +falls back to building the graph straight from CSV, and says so. +""" + +import hashlib +import json +import os +import sqlite3 +from pathlib import Path +from typing import Any, Dict, Optional + +from .. import gitutil as g +from .. import policy as pol +from ..track.local import LocalTrackStore +from .model import SPEC_DIR, _counts + +DB_NAME = "track.db" +GITIGNORE = "# Derived index. Rebuilt from CSV, specs and governance files. Never commit.\ntrack.db\ntrack.db-*\n" +SCHEMA = """ +CREATE TABLE IF NOT EXISTS nodes ( + id TEXT PRIMARY KEY, kind TEXT NOT NULL, title TEXT, status TEXT, + parent TEXT, source TEXT, data TEXT +); +CREATE TABLE IF NOT EXISTS meta (key TEXT PRIMARY KEY, value TEXT); +""" + + +def db_path(root: Path) -> Path: + return g.state_root(root) / ".onecoder" / DB_NAME + + +def fingerprint(root: Path) -> str: + """Hash of every input's path, mtime and size. Any edit to CSV, specs, or governance files changes it.""" + inputs = [] + track_dir = LocalTrackStore(root=root, policy=pol.load(root)).track_dir + inputs += sorted(track_dir.glob("*.csv")) + inputs += sorted((root / SPEC_DIR).glob("*.md")) if (root / SPEC_DIR).is_dir() else [] + tdir = g.state_root(root) / ".onecoder" / "tasks" + if tdir.is_dir(): + inputs += sorted(p for p in tdir.rglob("*") if p.is_file()) + h = hashlib.sha256() + for p in inputs: + st = p.stat() + h.update(f"{p}|{st.st_mtime_ns}|{st.st_size}\n".encode()) + return h.hexdigest() + + +def ensure_gitignore(dbdir: Path) -> None: + """Self-ignoring: the index never gets committed, in this repo or a nested one.""" + gi = dbdir / ".gitignore" + if not gi.exists(): + gi.write_text(GITIGNORE) + + +def write(root: Path, graph: Dict[str, Any], fp: str) -> Path: + path = db_path(root) + path.parent.mkdir(parents=True, exist_ok=True) + ensure_gitignore(path.parent) + tmp = path.with_name(DB_NAME + ".tmp") + if tmp.exists(): + tmp.unlink() + conn = sqlite3.connect(tmp) + try: + conn.executescript(SCHEMA) + conn.executemany( + "INSERT INTO nodes (id, kind, title, status, parent, source, data) VALUES (?, ?, ?, ?, ?, ?, ?)", + [(n["id"], n["kind"], n["title"], n["status"], n["parent"], n["source"], json.dumps(n["data"])) + for n in graph["nodes"]], + ) + conn.executemany("INSERT INTO meta (key, value) VALUES (?, ?)", + [("fingerprint", fp), ("generated_at", graph["generated_at"]), ("project", graph["project"])]) + conn.commit() + finally: + conn.close() + os.replace(tmp, path) + return path + + +def read(root: Path, fp: str) -> Optional[Dict[str, Any]]: + """Return the cached graph only if it was built from exactly the current inputs.""" + path = db_path(root) + if not path.exists(): + return None + conn = sqlite3.connect(f"file:{path}?mode=ro", uri=True) + try: + meta = dict(conn.execute("SELECT key, value FROM meta").fetchall()) + if meta.get("fingerprint") != fp: + return None + rows = conn.execute("SELECT id, kind, title, status, parent, source, data FROM nodes ORDER BY kind, id").fetchall() + finally: + conn.close() + nodes = [{"id": r[0], "kind": r[1], "title": r[2], "status": r[3], "parent": r[4], "source": r[5], + "data": json.loads(r[6] or "{}")} for r in rows] + return {"project": meta.get("project", root.name), "generated_at": meta.get("generated_at", ""), + "source": "sqlite", "nodes": nodes, "counts": _counts(nodes)} diff --git a/pyproject.toml b/pyproject.toml index f33745e..d5cfe20 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -13,5 +13,8 @@ dependencies = [] [project.scripts] onecoder = "onecoder.cli:main" +[tool.setuptools.package-data] +onecoder = ["viz/page.html"] + [tool.setuptools.packages.find] include = ["onecoder*"] diff --git a/tests/test_viz.py b/tests/test_viz.py new file mode 100644 index 0000000..277ea7c --- /dev/null +++ b/tests/test_viz.py @@ -0,0 +1,196 @@ +"""Tests for the visual work graph: model, SQLite index, CSV fallback, HTTP surface, and CLI.""" + +import json +import os +import subprocess +import sys +import tempfile +import threading +import unittest +import urllib.request +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT)) + +from onecoder.track.local import LocalTrackStore +from onecoder.viz import store as viz_store +from onecoder.viz.model import build_graph, parse_frontmatter +from onecoder.viz.server import make_server +from onecoder.viz.snapshot import snapshot + +BRIEF = """--- +id: gen-ui-core +kind: ship # ship = authorized change +--- +# gen-ui-core + +- [x] AC-1: catalog has four components +- [ ] AC-2: verbatim check is property tested +""" + + +def git_repo(path: Path) -> Path: + path.mkdir(parents=True) + subprocess.run(["git", "init", "-q"], cwd=path, check=True) + subprocess.run(["git", "-c", "user.name=t", "-c", "user.email=t@t", "commit", "-q", "--allow-empty", "-m", "init"], cwd=path, check=True) + return path + + +def seed(root: Path) -> None: + """A realistic project: a spec, an epic that implements it, tasks, and one governance task.""" + (root / "specs").mkdir() + (root / "specs" / "generative-ui.md").write_text( + "---\nid: gen-ui\ntitle: Generative UI MVP\nstatus: approved\nepic: sprint-mvp\n---\n# body\n") + store = LocalTrackStore(root=root) + store.create_sprint(project="demo", name="MVP build", status="active", id="sprint-mvp") + store.create_issue(project="demo", title="Normalizer table tests", sprint="sprint-mvp") + store.create_issue(project="demo", title="Renderer page") + issue = store.list_issues(sprint="sprint-mvp")["issues"][0] + gov = root / ".onecoder" / "tasks" / "gen-ui-core" + gov.mkdir(parents=True) + (gov / "state.json").write_text(json.dumps({"id": "gen-ui-core", "mode": "human-driver", + "isolation": "none", "branch": "main", "issue": issue["id"]})) + (gov / "brief.md").write_text(BRIEF) + (gov / "review.json").write_text(json.dumps({"verdict": "pass"})) + + +class TestModel(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.root = git_repo(Path(self.tmp.name) / "proj") + seed(self.root) + + def tearDown(self): + self.tmp.cleanup() + + def test_frontmatter_drops_inline_comments(self): + meta = parse_frontmatter("---\nkind: ship # ship = authorized\nid: x\n---\nbody") + self.assertEqual(meta, {"kind": "ship", "id": "x"}) + + def test_spec_epic_task_ac_hierarchy(self): + g = build_graph(self.root) + by_id = {n["id"]: n for n in g["nodes"]} + self.assertEqual(by_id["spec:gen-ui"]["title"], "Generative UI MVP") + self.assertEqual(by_id["epic:sprint-mvp"]["parent"], "spec:gen-ui") + issue_id = [n for n in g["nodes"] if n["kind"] == "task" and n["title"] == "Normalizer table tests"][0]["id"] + self.assertEqual(by_id[issue_id]["parent"], "epic:sprint-mvp") + self.assertEqual(by_id["gov:gen-ui-core"]["parent"], issue_id) + self.assertEqual(by_id["gov:gen-ui-core"]["status"], "pass") + self.assertEqual(by_id["ac:gen-ui-core:AC-1"]["status"], "done") + self.assertEqual(by_id["ac:gen-ui-core:AC-2"]["status"], "open") + self.assertEqual(by_id["ac:gen-ui-core:AC-2"]["parent"], "gov:gen-ui-core") + + def test_unplanned_tasks_go_to_a_group_and_no_dangling_parents(self): + g = build_graph(self.root) + by_id = {n["id"]: n for n in g["nodes"]} + renderer = [n for n in g["nodes"] if n["title"] == "Renderer page"][0] + self.assertEqual(renderer["parent"], "group:unplanned") + self.assertIn("group:unplanned", by_id) + for n in g["nodes"]: + if n["parent"]: + self.assertIn(n["parent"], by_id) + + def test_counts_exclude_groups(self): + counts = build_graph(self.root)["counts"] + self.assertNotIn("group", counts) + self.assertEqual(counts["spec"], {"approved": 1}) + + +class TestIdsDoNotCollide(unittest.TestCase): + def test_issues_created_in_a_burst_all_keep_distinct_ids(self): + with tempfile.TemporaryDirectory() as tmp: + store = LocalTrackStore(root=git_repo(Path(tmp) / "p")) + ids = [store.create_issue(project="demo", title=f"t{i}")["id"] for i in range(20)] + self.assertEqual(len(set(ids)), 20) + self.assertEqual(len(store.list_issues()["issues"]), 20) + + +class TestIndexAndFallback(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.root = git_repo(Path(self.tmp.name) / "proj") + seed(self.root) + + def tearDown(self): + self.tmp.cleanup() + + def test_sqlite_index_is_built_then_reused_until_an_input_changes(self): + first = snapshot(self.root) + self.assertEqual(first["source"], "sqlite") + self.assertTrue((self.root / ".onecoder" / "track.db").exists()) + self.assertEqual(snapshot(self.root)["generated_at"], first["generated_at"]) # served from the index + LocalTrackStore(root=self.root).create_issue(project="demo", title="New task after the index") + second = snapshot(self.root) + self.assertTrue(any(n["title"] == "New task after the index" for n in second["nodes"])) + + def test_index_is_self_ignoring_and_git_ignores_only_the_db(self): + snapshot(self.root) + gi = (self.root / ".onecoder" / ".gitignore").read_text() + self.assertIn("track.db", gi) + ignored = subprocess.run(["git", "check-ignore", "-q", ".onecoder/track.db"], cwd=self.root).returncode + kept = subprocess.run(["git", "check-ignore", "-q", ".onecoder/track/issues.csv"], cwd=self.root).returncode + self.assertEqual(ignored, 0) + self.assertNotEqual(kept, 0) + + def test_falls_back_to_csv_when_the_index_cannot_be_written(self): + # Put a directory where the database file must go: the write fails, the UI still works. + (self.root / ".onecoder" / "track.db").mkdir(parents=True) + g = snapshot(self.root) + self.assertEqual(g["source"], "csv") + self.assertIn("serving from CSV", g["warning"]) + self.assertTrue(any(n["title"] == "Renderer page" for n in g["nodes"])) + + def test_stale_index_is_never_served(self): + fp = viz_store.fingerprint(self.root) + viz_store.write(self.root, build_graph(self.root), fp) + (self.root / "specs" / "generative-ui.md").write_text("---\nid: gen-ui\ntitle: Renamed\n---\n") + self.assertIsNone(viz_store.read(self.root, viz_store.fingerprint(self.root))) + + +class TestHttpAndCli(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.root = git_repo(Path(self.tmp.name) / "proj") + seed(self.root) + self.server = make_server(self.root, port=0) + self.base = f"http://127.0.0.1:{self.server.server_address[1]}" + threading.Thread(target=self.server.serve_forever, daemon=True).start() + + def tearDown(self): + self.server.shutdown() + self.server.server_close() + self.tmp.cleanup() + + def get(self, path): + try: + with urllib.request.urlopen(self.base + path) as r: + return r.status, r.headers, r.read() + except urllib.error.HTTPError as e: + return e.code, e.headers, e.read() + + def test_page_and_graph_are_served_with_a_strict_csp(self): + status, headers, body = self.get("/") + self.assertEqual(status, 200) + self.assertIn(b"OneCoder: work graph", body) + self.assertIn("default-src 'none'", headers["Content-Security-Policy"]) + status, headers, body = self.get("/api/graph") + graph = json.loads(body) + self.assertEqual(status, 200) + self.assertIn(graph["source"], {"sqlite", "csv"}) + self.assertTrue(any(n["kind"] == "epic" for n in graph["nodes"])) + self.assertEqual(headers["Cache-Control"], "no-store") + + def test_unknown_paths_are_404(self): + self.assertEqual(self.get("/etc/passwd")[0], 404) + + def test_cli_print_json_runs_from_the_project(self): + r = subprocess.run([sys.executable, "-m", "onecoder", "viz", "--print-json"], + cwd=self.root, capture_output=True, text=True, + env={**os.environ, "PYTHONPATH": str(ROOT)}) + self.assertEqual(r.returncode, 0, r.stderr) + self.assertIn("spec:gen-ui", r.stdout) + + +if __name__ == "__main__": + unittest.main()