Blame for cli/tui.py

#!/usr/bin/env python3
"""
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("'"))

# --------------------------------------------------------------------------- #
# Core binding. In --demo we stub these so the render is provable without a DB.
# --------------------------------------------------------------------------- #
COLUMNS = ["backlog", "ready", "in-flight", "review", "blocked", "done"]

# column -> (fg ansi, glyph) for the interactive head
COLUMN_STYLE = {
    "backlog":   (244, "•"),
    "ready":     (39,  "▸"),
    "in-flight": (220, "◆"),
    "review":    (135, "?"),
    "blocked":   (196, "⛔"),
    "done":      (46,  "✓"),
}


def _now() -> datetime:
    return datetime.now(timezone.utc)


# --------------------------------------------------------------------------- #
# DEMO backend — seeded fake board so `--demo` renders with zero infra.
# Mirrors the exact shape core.db returns (id/title/column/role/worker/attempt).
# --------------------------------------------------------------------------- #
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
        # Each demo board is its own bracelet. Agents share that copy for the
        # life of the process, and moves are checked against the real table.
        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)}

    # -- actions: each maps to exactly one MCP tool ------------------------- #
    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"


# --------------------------------------------------------------------------- #
# FACE 1 — AGENT line protocol. Deterministic JSONL. This is what an agent drives.
# --------------------------------------------------------------------------- #
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:  # never leak a stack trace into the JSONL stream
        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:  # EOF
            return 0
        res = await _dispatch(board, line)
        print(json.dumps(res), flush=True)
        if res.get("bye"):
            return 0


# --------------------------------------------------------------------------- #
# FACE 2 — INTERACTIVE curses board. Same data, for human eyes + keys.
# --------------------------------------------------------------------------- #
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)

        # flat, selectable card list grouped by column
        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)  # refresh every 2s if idle
        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:  # timeout -> refresh
                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


# --------------------------------------------------------------------------- #
# Entry — pick the face. TTY => interactive; pipe/--agent => line protocol.
# --------------------------------------------------------------------------- #
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())