| |
| """ |
| KanBot TUI — the board, in the terminal, driven by humans OR agents. |
| |
| ONE core, TWO faces (same inversion as KanBot itself): |
| |
| • INTERACTIVE (a TTY is attached): a live full-screen curses Kanban. Columns |
| are execution states; cards stream in; keys act on the selected card. Auto- |
| refreshes on an interval (and can be nudged). This is `htop` for the swarm. |
| |
| • AGENT (stdin is a pipe, or `--agent`): a line protocol an agent drives. |
| Read a command on stdin, get one JSON object per line on stdout. No curses, |
| no ANSI, no prompts — deterministic, greppable, scriptable. Every command |
| maps 1:1 onto KanBot's 9 MCP tools; nothing else can fire. |
| |
| Both faces call the SAME functions (core.db + core.mcp_server). There is no |
| second code path to drift — the human view and the agent view are one board. |
| |
| Zero new dependencies: stdlib curses + asyncio only. |
| |
| Run: |
| python -m cli.tui # interactive if TTY, agent-mode if piped |
| python -m cli.tui --agent # force agent line-protocol |
| echo 'board' | python -m cli.tui # one-shot agent query |
| python -m cli.tui --demo # render the board with seeded fake data |
| # (no DB needed — proves the render) |
| |
| Agent protocol (one command per line on stdin; one JSON object per line out): |
| board -> {"ok":true,"columns":{...counts...},"cards":[...]} |
| get CARD-3 -> {"ok":true,"card":{...full detail...}} |
| claim CARD-3 <worker> -> claim_card |
| move CARD-3 <col> [reason] -> move_card (guard-validated server-side) |
| approve CARD-3 -> unblock a dry-run proposal to ready |
| block CARD-3 <reason> -> block_card |
| human CARD-3 <question> -> request_human |
| new "<title>" [spec] -> create_card (lands in backlog) |
| pulse -> db, provider, dry-run (in this terminal) |
| help -> list commands |
| quit -> exit |
| Errors always come back as {"ok":false,"error":"..."} — never a stack trace to |
| stdout, so an agent parsing the stream never chokes. |
| """ |
| from __future__ import annotations |
|
|
| import argparse |
| import asyncio |
| import json |
| import os |
| import shlex |
| import sys |
| from datetime import datetime, timezone |
| from pathlib import Path |
|
|
| def _load_repo_env() -> None: |
| env = Path(__file__).resolve().parents[1] / ".env" |
| if not env.is_file(): |
| return |
| for raw in env.read_text().splitlines(): |
| line = raw.strip() |
| if not line or line.startswith("#") or "=" not in line: |
| continue |
| if line.startswith("export "): |
| line = line[len("export "):] |
| key, value = line.split("=", 1) |
| os.environ.setdefault(key.strip(), value.strip().strip('"').strip("'")) |
|
|
| |
| |
| |
| COLUMNS = ["backlog", "ready", "in-flight", "review", "blocked", "done"] |
|
|
| |
| COLUMN_STYLE = { |
| "backlog": (244, "•"), |
| "ready": (39, "▸"), |
| "in-flight": (220, "◆"), |
| "review": (135, "?"), |
| "blocked": (196, "⛔"), |
| "done": (46, "✓"), |
| } |
|
|
|
|
| def _now() -> datetime: |
| return datetime.now(timezone.utc) |
|
|
|
|
| |
| |
| |
| |
| class DemoCard: |
| def __init__(self, **k): |
| self.thread = [] |
| self.__dict__.update(k) |
| if not hasattr(self, "thread") or self.thread is None: |
| self.thread = [] |
|
|
| def has_artifact(self) -> bool: |
| return "artifact" in self.thread |
|
|
|
|
| DEMO_CARDS = [ |
| DemoCard(id="CARD-1", title="Write hello.txt (parent)", column="done", |
| agent_role_value="planner", assigned_worker=None, attempt=0, |
| lease="—", requires_human=False), |
| DemoCard(id="CARD-2", title="create hello.txt (1 shell step)", column="blocked", |
| agent_role_value="executor", assigned_worker="exec-a1b2c3", attempt=2, |
| lease="expired", requires_human=True), |
| DemoCard(id="CARD-3", title="verify file contains a date", column="ready", |
| agent_role_value="executor", assigned_worker=None, attempt=0, |
| lease="—", requires_human=False), |
| DemoCard(id="CARD-4", title="deploy fleet cockpit surface", column="in-flight", |
| agent_role_value="executor", assigned_worker="exec-9f8e7d", attempt=0, |
| lease="14m", requires_human=False), |
| DemoCard(id="CARD-5", title="wire critic swarm N=3", column="backlog", |
| agent_role_value="planner", assigned_worker=None, attempt=0, |
| lease="—", requires_human=False), |
| DemoCard(id="CARD-6", title="harden sandbox -> Firecracker", column="ready", |
| agent_role_value="executor", assigned_worker=None, attempt=0, |
| lease="—", requires_human=False), |
| ] |
|
|
|
|
| class Board: |
| """Uniform board reader/actor. Wraps the real core in live mode, the seed in demo.""" |
|
|
| def __init__(self, demo: bool): |
| self.demo = demo |
| |
| |
| self._demo = [DemoCard(**c.__dict__) for c in DEMO_CARDS] if demo else [] |
|
|
| async def snapshot(self) -> dict: |
| """Return {'columns': {col: count}, 'cards': [flat dicts]} for both faces.""" |
| cards = [] |
| if self.demo: |
| for c in self._demo: |
| cards.append(dict(id=c.id, title=c.title, column=c.column, |
| role=c.agent_role_value, worker=c.assigned_worker, |
| attempt=c.attempt, lease=c.lease, |
| requires_human=c.requires_human)) |
| else: |
| from core import db |
| from core.contract import Column |
| for col in COLUMNS: |
| for c in await db.list_column(Column(col)): |
| cards.append(dict( |
| id=c.id, title=c.title, column=c.column.value, |
| role=c.agent_role.value, worker=c.assigned_worker, |
| attempt=c.attempt, |
| lease=_lease_str(c.lease_expires_at), |
| requires_human=c.requires_human)) |
| counts = {col: sum(1 for c in cards if c["column"] == col) for col in COLUMNS} |
| return {"columns": counts, "cards": cards} |
|
|
| async def get(self, card_id: str) -> dict: |
| if self.demo: |
| c = self._demo_find(card_id) |
| if c is None: |
| return {"ok": False, "error": f"{card_id} not found"} |
| return {"ok": True, "card": {"id": c.id, "title": c.title, |
| "column": c.column, "attempt": c.attempt, |
| "worker": c.assigned_worker}} |
| from core.mcp_server import get_card |
| raw = json.loads(await get_card(card_id)) |
| return {"ok": "error" not in raw, **({"card": raw} if "error" not in raw else raw)} |
|
|
| |
| def _demo_find(self, cid: str): |
| for c in self._demo: |
| if c.id == cid: |
| return c |
| return None |
|
|
| def _demo_move(self, cid: str, to_col: str, reason: str, worker: str = "", |
| verdict: str = "") -> dict: |
| from core.contract import Column, GuardFailed, IllegalTransition, check_guard, lookup_transition |
| card = self._demo_find(cid) |
| if card is None: |
| return {"ok": False, "error": "not found", "tool": "move_card"} |
| try: |
| frm, to = Column(card.column), Column(to_col) |
| except ValueError: |
| return {"ok": False, "error": f"unknown column {to_col}", "tool": "move_card", "rejected": True} |
| try: |
| t = lookup_transition(frm, to) |
| check_guard(card, to, t.requires, reason=reason, worker=worker, |
| critic_verdict=verdict or None) |
| except (IllegalTransition, GuardFailed) as e: |
| return {"ok": False, "error": str(e), "rejected": True, "tool": "move_card"} |
| card.column = to.value |
| if worker: |
| card.assigned_worker = worker |
| if to != Column.IN_FLIGHT and to != Column.BLOCKED: |
| card.assigned_worker = None |
| return {"ok": True, "demo": True, "tool": "move_card", "from": frm.value, "to": to.value} |
|
|
| def _demo_act(self, name: str, kw: dict) -> dict: |
| if name == "move_card": |
| return self._demo_move(kw["card_id"], kw["to_column"], kw.get("reason") or "", |
| kw.get("worker") or "", kw.get("critic_verdict") or "") |
| if name == "block_card": |
| return self._demo_move(kw["card_id"], "blocked", kw.get("reason") or "") |
| if name == "claim_card": |
| card = self._demo_find(kw["card_id"]) |
| if card is None: |
| return {"ok": False, "error": "not found", "tool": name} |
| if card.column != "ready": |
| return {"ok": False, "rejected": True, "tool": name, |
| "error": f"claim requires ready, {kw['card_id']} is {card.column}"} |
| card.assigned_worker = kw["worker"] |
| return {"ok": True, "demo": True, "tool": name, "worker": kw["worker"]} |
| if name == "request_human": |
| card = self._demo_find(kw["card_id"]) |
| if card is None: |
| return {"ok": False, "error": "not found", "tool": name} |
| card.requires_human = True |
| return {"ok": True, "demo": True, "tool": name} |
| if name == "comment" and kw.get("kind") == "artifact": |
| card = self._demo_find(kw["card_id"]) |
| if card is None: |
| return {"ok": False, "error": "not found", "tool": name} |
| card.thread.append("artifact") |
| return {"ok": True, "demo": True, "tool": name, "kind": "artifact"} |
| if name == "create_card": |
| nums = [int(c.id[5:]) for c in self._demo if c.id.startswith("CARD-") and c.id[5:].isdigit()] |
| card = DemoCard(id=f"CARD-{max(nums or [0]) + 1}", title=kw.get("title") or "untitled", |
| column="backlog", agent_role_value="executor", |
| assigned_worker=None, attempt=0, lease="—", |
| requires_human=False) |
| self._demo.append(card) |
| return {"ok": True, "demo": True, "tool": name, "id": card.id, "to": "backlog"} |
| return {"ok": False, "error": f"demo has no {name}", "tool": name} |
|
|
| async def _tool(self, name: str, **kw) -> dict: |
| if self.demo: |
| return self._demo_act(name, kw) |
| from core import mcp_server |
| fn = getattr(mcp_server, name) |
| raw = json.loads(await fn(**kw)) |
| return {"ok": not raw.get("rejected") and "error" not in raw, "tool": name, **raw} |
|
|
| async def claim(self, cid, worker): return await self._tool("claim_card", card_id=cid, worker=worker) |
| async def move(self, cid, col, why, verdict=""): |
| return await self._tool("move_card", card_id=cid, to_column=col, reason=why, |
| actor="human:remy", critic_verdict=verdict) |
|
|
| async def artifact(self, cid, text): |
| return await self._tool("comment", card_id=cid, text=text, author="human:remy", kind="artifact") |
| async def block(self, cid, why): return await self._tool("block_card", card_id=cid, reason=why, actor="human:remy") |
| async def human(self, cid, q): return await self._tool("request_human", card_id=cid, question=q, actor="human:remy") |
| async def approve(self, cid): return await self._tool("move_card", card_id=cid, to_column="ready", reason="human approved", actor="human:remy") |
| async def new(self, title, spec): return await self._tool("create_card", title=title, spec=spec) |
|
|
| async def pulse(self) -> dict: |
| """Health of this process. No separate doctor command.""" |
| dry = os.environ.get("KANBOT_DRY_RUN", "1") != "0" |
| prov = os.environ.get("KANBOT_PROVIDER", "ollama") |
| if self.demo: |
| return {"ok": True, "tool": "pulse", "demo": True, "postgres": False, |
| "provider": "mock", "dry_run": True} |
| pg = True |
| err = "" |
| try: |
| from core import db |
| async with (await db.pool()).acquire() as conn: |
| await conn.fetchval("SELECT 1") |
| except Exception as e: |
| pg, err = False, type(e).__name__ |
| return {"ok": pg, "tool": "pulse", "demo": False, "postgres": pg, |
| "provider": prov, "dry_run": dry, "error": err} |
|
|
|
|
| def _lease_str(iso: str | None) -> str: |
| if not iso: |
| return "—" |
| try: |
| delta = (datetime.fromisoformat(iso) - _now()).total_seconds() |
| except Exception: |
| return "?" |
| if delta <= 0: |
| return "expired" |
| m = int(delta // 60) |
| return f"{m}m" if m else f"{int(delta)}s" |
|
|
|
|
| |
| |
| |
| AGENT_HELP = { |
| "board": "list all columns + cards (counts + flat card list)", |
| "get": "get CARD-x — full card detail", |
| "claim": "claim CARD-x <worker> — claim a ready card", |
| "move": "move CARD-x <column> [reason] — guarded transition", |
| "artifact": "artifact CARD-x <text> — post the result the review guard requires", |
| "pass": "pass CARD-x — review to done with a critic PASS", |
| "fail": "fail CARD-x <reason> — review back to ready with critic feedback", |
| "approve": "approve CARD-x — unblock a dry-run proposal to ready", |
| "block": "block CARD-x <reason> — halt a card", |
| "human": "human CARD-x <question> — flag for human decision", |
| "new": 'new "<title>" [spec] — create a card in backlog', |
| "pulse": "board, provider, and dry-run — the in-terminal check", |
| "help": "show this", |
| "quit": "exit", |
| } |
|
|
|
|
| async def _dispatch(board: Board, line: str) -> dict: |
| line = line.strip() |
| if not line: |
| return {"ok": True, "noop": True} |
| try: |
| parts = shlex.split(line) |
| except ValueError as e: |
| return {"ok": False, "error": f"parse: {e}"} |
| cmd, args = parts[0].lower(), parts[1:] |
| try: |
| if cmd == "board": |
| snap = await board.snapshot() |
| return {"ok": True, **snap} |
| if cmd == "get" and args: |
| return await board.get(args[0]) |
| if cmd == "claim" and len(args) >= 2: |
| return await board.claim(args[0], args[1]) |
| if cmd == "move" and len(args) >= 2: |
| return await board.move(args[0], args[1], " ".join(args[2:]) or "moved via tui") |
| if cmd == "artifact" and len(args) >= 2: |
| return await board.artifact(args[0], " ".join(args[1:])) |
| if cmd == "pass" and args: |
| return await board.move(args[0], "done", "critic pass", verdict="pass") |
| if cmd == "fail" and len(args) >= 2: |
| return await board.move(args[0], "ready", " ".join(args[1:]), verdict="fail") |
| if cmd == "approve" and args: |
| return await board.approve(args[0]) |
| if cmd == "block" and len(args) >= 2: |
| return await board.block(args[0], " ".join(args[1:])) |
| if cmd == "human" and len(args) >= 2: |
| return await board.human(args[0], " ".join(args[1:])) |
| if cmd == "new" and args: |
| return await board.new(args[0], " ".join(args[1:])) |
| if cmd == "pulse": |
| return await board.pulse() |
| if cmd == "help": |
| return {"ok": True, "commands": AGENT_HELP} |
| if cmd in ("quit", "exit"): |
| return {"ok": True, "bye": True} |
| return {"ok": False, "error": f"unknown or malformed command: {line!r}", |
| "hint": "send 'help'"} |
| except Exception as e: |
| return {"ok": False, "error": f"{type(e).__name__}: {e}"} |
|
|
|
|
| async def run_agent(board: Board) -> int: |
| """Read commands from stdin, emit one JSON object per line. Agent-native.""" |
| loop = asyncio.get_event_loop() |
| print(json.dumps({"ok": True, "kanbot": "agent-tui", "protocol": 1, |
| "columns": COLUMNS}), flush=True) |
| while True: |
| line = await loop.run_in_executor(None, sys.stdin.readline) |
| if not line: |
| return 0 |
| res = await _dispatch(board, line) |
| print(json.dumps(res), flush=True) |
| if res.get("bye"): |
| return 0 |
|
|
|
|
| |
| |
| |
| KEYS = " [↑/↓] select [a]pprove [c]laim [m]ove→ready [b]lock [h]uman [r]efresh [q]uit" |
|
|
|
|
| def run_interactive(board: Board) -> int: |
| import curses |
|
|
| def _draw(stdscr, snap, sel, msg): |
| stdscr.erase() |
| h, w = stdscr.getmaxyx() |
| title = " KANBOT — the board IS the runtime " |
| stdscr.attron(curses.A_REVERSE | curses.A_BOLD) |
| stdscr.addstr(0, 0, title.ljust(w)[:w]) |
| stdscr.attroff(curses.A_REVERSE | curses.A_BOLD) |
|
|
| counts = snap["columns"] |
| summary = " ".join(f"{COLUMN_STYLE[c][1]} {c}:{counts[c]}" for c in COLUMNS) |
| stdscr.addstr(1, 1, summary[:w - 2], curses.A_DIM) |
|
|
| |
| row = 3 |
| flat = [] |
| for col in COLUMNS: |
| cards = [c for c in snap["cards"] if c["column"] == col] |
| if not cards: |
| continue |
| if row < h - 2: |
| stdscr.addstr(row, 1, f"{COLUMN_STYLE[col][1]} {col.upper()}"[:w - 2], |
| curses.A_BOLD) |
| row += 1 |
| for c in cards: |
| flat.append(c) |
| if row >= h - 2: |
| continue |
| idx = len(flat) - 1 |
| worker = c["worker"] or "—" |
| att = f"{c['attempt']}" + (f"/3" if c["attempt"] else "") |
| hum = " ⚑" if c.get("requires_human") else "" |
| linetxt = f" {c['id']:<8} {c['title'][:w-42]:<{max(1,w-42)}} " \ |
| f"[{worker:<10}] a:{att} {c.get('lease','—')}{hum}" |
| attr = curses.A_REVERSE if idx == sel else curses.A_NORMAL |
| stdscr.addstr(row, 1, linetxt[:w - 2], attr) |
| row += 1 |
|
|
| stdscr.addstr(h - 2, 1, (msg or "")[:w - 2], curses.A_DIM) |
| stdscr.addstr(h - 1, 0, KEYS[:w - 1], curses.A_REVERSE) |
| stdscr.refresh() |
| return flat |
|
|
| async def _amove(cid, col, why): return await board.move(cid, col, why) |
|
|
| def _loop(stdscr): |
| curses.curs_set(0) |
| stdscr.nodelay(False) |
| stdscr.timeout(2000) |
| sel, msg = 0, "live board — actions re-confirm before mutating" |
| aloop = asyncio.new_event_loop() |
| snap = aloop.run_until_complete(board.snapshot()) |
| while True: |
| flat = _draw(stdscr, snap, sel, msg) |
| n = len(flat) |
| try: |
| ch = stdscr.getch() |
| except KeyboardInterrupt: |
| break |
| if ch == -1: |
| snap = aloop.run_until_complete(board.snapshot()); continue |
| if ch in (ord("q"), 27): |
| break |
| elif ch in (curses.KEY_DOWN, ord("j")) and n: |
| sel = (sel + 1) % n |
| elif ch in (curses.KEY_UP, ord("k")) and n: |
| sel = (sel - 1) % n |
| elif ch in (ord("r"),): |
| snap = aloop.run_until_complete(board.snapshot()); msg = "refreshed" |
| elif n and ch in (ord("a"), ord("c"), ord("m"), ord("b"), ord("h")): |
| card = flat[sel] |
| cid = card["id"] |
| if ch == ord("a"): |
| res = aloop.run_until_complete(board.approve(cid)) |
| elif ch == ord("c"): |
| res = aloop.run_until_complete(board.claim(cid, "human:remy")) |
| elif ch == ord("m"): |
| res = aloop.run_until_complete(board.move(cid, "ready", "moved via tui")) |
| elif ch == ord("b"): |
| res = aloop.run_until_complete(board.block(cid, "halted via tui")) |
| elif ch == ord("h"): |
| res = aloop.run_until_complete(board.human(cid, "operator flagged via tui")) |
| ok = res.get("ok") |
| msg = f"{'✓' if ok else '✗'} {cid}: {res.get('tool','')} " \ |
| f"{res.get('to', res.get('error',''))}" |
| snap = aloop.run_until_complete(board.snapshot()) |
| aloop.close() |
|
|
| return curses.wrapper(_loop) or 0 |
|
|
|
|
| |
| |
| |
| def main() -> int: |
| _load_repo_env() |
| ap = argparse.ArgumentParser(description="KanBot terminal Kanban (human + agent).") |
| ap.add_argument("--agent", action="store_true", help="force agent line-protocol") |
| ap.add_argument("--demo", action="store_true", help="seeded fake board, no DB needed") |
| args = ap.parse_args() |
|
|
| board = Board(demo=args.demo) |
| is_tty = sys.stdin.isatty() and sys.stdout.isatty() |
| agent_mode = args.agent or not is_tty |
|
|
| if agent_mode: |
| return asyncio.run(run_agent(board)) |
| return run_interactive(board) |
|
|
|
|
| if __name__ == "__main__": |
| raise SystemExit(main()) |
|
|
