Blame for cli/kanbot_cli.py

#!/usr/bin/env python3
"""
kanbot — the unified CLI + agent harness for the board-that-is-the-runtime.

One entrypoint, subcommands, provider-agnostic. Humans read the pretty output;
agents pass --json and get one machine object per command. Every mutating verb
maps 1:1 to a KanBot MCP tool; the LLM backend is swappable via the provider
harness (ollama|openai|anthropic|openrouter|vllm|openai_compat|mock).

  kanbot board [--json] [--demo]        show the board
  kanbot tui   [--demo]                 launch the interactive curses board
  kanbot agent [--demo]                 the JSONL line-protocol loop (agent-driven)
  kanbot card <id> [--json]             card detail
  kanbot move <id> <col> [-m reason]    guarded transition   (move_card)
  kanbot claim <id> <worker>            claim a ready card   (claim_card)
  kanbot approve <id>                   unblock a proposal   (-> ready)
  kanbot block <id> <reason>            halt a card          (block_card)
  kanbot human <id> <question>          flag for a human     (request_human)
  kanbot new "<title>" [-s spec] [-c crit ...]   create a card (create_card)
  kanbot run [--dry-run|--live] [--swarm N]      the autonomous conductor loop
  kanbot providers [--json]             list providers + active + resolved models
  kanbot doctor [--json]                preflight: DB, provider, models, dry-run
  kanbot config [--json]                resolved config (env > default), keys masked

Exit codes: 0 ok · 1 operational failure · 2 usage error.
Config precedence: CLI flag > environment > built-in default. Secrets never printed.
"""
from __future__ import annotations

import argparse
import asyncio
import json
import os
import sys

# --- import the KanBot package whether invoked as `kanbot`, `-m cli.kanbot_cli`,
#     or from inside cli/ -------------------------------------------------------
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

RESET, BOLD, DIM = "\033[0m", "\033[1m", "\033[2m"
GREEN, RED, YEL = "\033[32m", "\033[31m", "\033[33m"


def _out(obj: dict, as_json: bool, human: str | None = None) -> None:
    if as_json:
        print(json.dumps(obj))
    else:
        print(human if human is not None else json.dumps(obj, indent=2))


# --------------------------------------------------------------------------- #
# board / tui / agent  — delegate to the existing dual-mode cli/tui.py
# --------------------------------------------------------------------------- #
def _board(demo: bool):
    from cli.tui import Board
    return Board(demo=demo)


async def cmd_board(a) -> int:
    snap = await _board(a.demo).snapshot()
    if a.json:
        _out({"ok": True, **snap}, True)
        return 0
    cols = snap["columns"]
    print(f"{BOLD} KANBOT board {RESET} " + DIM +
          "  ".join(f"{c}:{cols[c]}" for c in cols) + RESET)
    for col in cols:
        cards = [c for c in snap["cards"] if c["column"] == col]
        if not cards:
            continue
        print(f"\n{BOLD}{col.upper()}{RESET}")
        for c in cards:
            att = f"a:{c['attempt']}" + ("/3" if c["attempt"] else "")
            flag = f" {YEL}⚑{RESET}" if c.get("requires_human") else ""
            print(f"  {c['id']:<8} {c['title'][:52]:<52} "
                  f"[{c['worker'] or '—':<11}] {att} {c.get('lease','—')}{flag}")
    return 0


def cmd_tui(a) -> int:
    from cli.tui import Board, run_interactive
    return run_interactive(Board(demo=a.demo))


def cmd_agent(a) -> int:
    from cli.tui import Board, run_agent
    return asyncio.run(run_agent(Board(demo=a.demo)))


# --------------------------------------------------------------------------- #
# card actions — each maps to one MCP tool via the Board wrapper
# --------------------------------------------------------------------------- #
async def cmd_card(a) -> int:
    res = await _board(a.demo).get(a.id)
    _out(res, a.json, None if a.json else json.dumps(res.get("card", res), indent=2))
    return 0 if res.get("ok") else 1


async def _act(a, coro) -> int:
    res = await coro
    ok = res.get("ok", False)
    if a.json:
        _out(res, True)
    else:
        mark = f"{GREEN}✓{RESET}" if ok else f"{RED}✗{RESET}"
        detail = res.get("to") or res.get("error") or res.get("note") or res.get("tool", "")
        print(f"{mark} {res.get('tool','action')}: {detail}")
    return 0 if ok else 1


async def cmd_move(a):    return await _act(a, _board(a.demo).move(a.id, a.column, a.reason or "moved via cli"))
async def cmd_claim(a):   return await _act(a, _board(a.demo).claim(a.id, a.worker))
async def cmd_approve(a): return await _act(a, _board(a.demo).approve(a.id))
async def cmd_block(a):   return await _act(a, _board(a.demo).block(a.id, a.reason))
async def cmd_human(a):   return await _act(a, _board(a.demo).human(a.id, a.question))


async def cmd_new(a) -> int:
    b = _board(a.demo)
    if b.demo:
        return await _act(a, b.new(a.title, a.spec or ""))
    from core import mcp_server
    raw = json.loads(await mcp_server.create_card(
        title=a.title, spec=a.spec or "", acceptance_criteria=a.criteria or []))
    res = {"ok": "error" not in raw, "tool": "create_card", **raw}
    return await _act(a, _coro_ret(res))


async def _coro_ret(v):  # tiny helper so _act can await a plain dict
    return v


# --------------------------------------------------------------------------- #
# providers / doctor / config — the harness surface
# --------------------------------------------------------------------------- #
def cmd_providers(a) -> int:
    from core import providers
    d = providers.describe()
    if a.json:
        _out({"ok": True, **d}, True); return 0
    print(f"{BOLD}Active provider:{RESET} {d['active']}")
    print(f"{DIM}Supported:{RESET} {', '.join(d['supported'])}")
    print(f"{BOLD}Models{RESET}  fast={d['models']['fast']}  "
          f"daily={d['models']['daily']}  heavy={d['models']['heavy']}")
    if d["key_required"]:
        state = f"{GREEN}present{RESET}" if d["key_present"] else f"{RED}MISSING{RESET}"
        print(f"{BOLD}Key{RESET} {d['key_required']}: {state}")
    if d["base_url"]:
        print(f"{BOLD}Base URL{RESET} {d['base_url']}")
    return 0


async def cmd_doctor(a) -> int:
    """No-mock preflight: probe the REAL services the active config needs."""
    from core import providers
    checks, fail = [], 0

    # provider reachability — mock is always green; real ones get a tiny ping
    d = providers.describe()
    prov = d["active"]
    if prov == "mock":
        checks.append(("provider", True, "mock (hermetic, always ready)"))
    elif prov == "ollama":
        ok, msg = await _ping_ollama()
        checks.append(("provider", ok, msg)); fail += not ok
    else:
        need = d["key_required"]
        ok = (not need) or d["key_present"]
        checks.append(("provider", ok,
                       f"{prov}: key {'present' if ok else 'MISSING ('+str(need)+')'}"))
        fail += not ok

    # postgres (skip in demo)
    if not a.demo:
        ok, msg = await _ping_pg()
        checks.append(("postgres", ok, msg)); fail += not ok

    dry = os.environ.get("KANBOT_DRY_RUN", "1") == "1"
    checks.append(("run-mode", True, "dry-run (proposals only)" if dry else "LIVE (executes)"))

    payload = {"ok": fail == 0, "provider": prov,
               "checks": [{"name": n, "ok": o, "detail": m} for n, o, m in checks]}
    if a.json:
        _out(payload, True)
    else:
        print(f"{BOLD} kanbot doctor {RESET}  provider={prov}\n")
        for n, o, m in checks:
            mark = f"{GREEN}✓{RESET}" if o else f"{RED}✗{RESET}"
            print(f"  {mark} {n:<10} {DIM}{m}{RESET}")
        print("\n  " + (f"{GREEN}READY{RESET}" if fail == 0
                        else f"{RED}NOT READY: {fail} check(s) failed{RESET}"))
    return 0 if fail == 0 else 1


async def _ping_ollama():
    host = os.environ.get("OLLAMA_HOST", "http://localhost:11434")
    try:
        import httpx
        async with httpx.AsyncClient(timeout=4) as c:
            r = await c.get(f"{host}/api/tags"); r.raise_for_status()
        return True, f"ollama reachable at {host}"
    except Exception as e:
        return False, f"ollama unreachable at {host}: {e}"


async def _ping_pg():
    dsn = os.environ.get("DATABASE_URL", "postgresql://kanbot:kanbot@localhost:5432/kanbot")
    try:
        import asyncpg
        conn = await asyncpg.connect(dsn, timeout=4)
        has = await conn.fetchval("SELECT to_regclass('public.cards') IS NOT NULL")
        await conn.close()
        return bool(has), "connected, schema present" if has else "connected, schema MISSING"
    except Exception as e:
        return False, f"cannot connect: {e}"


def cmd_config(a) -> int:
    """Resolved config with precedence env>default. Secrets are masked."""
    def mask(v): return (v[:3] + "…" + v[-2:]) if v and len(v) > 6 else ("set" if v else None)
    cfg = {
        "KANBOT_PROVIDER": os.environ.get("KANBOT_PROVIDER", "ollama (default)"),
        "KANBOT_DRY_RUN": os.environ.get("KANBOT_DRY_RUN", "1 (default)"),
        "KANBOT_MAX_ATTEMPTS": os.environ.get("KANBOT_MAX_ATTEMPTS", "3 (default)"),
        "KANBOT_SWARM_N": os.environ.get("KANBOT_SWARM_N", "1 (default)"),
        "KANBOT_MAX_WORKERS": os.environ.get("KANBOT_MAX_WORKERS", "3 (default)"),
        "DATABASE_URL": os.environ.get("DATABASE_URL", "postgresql://kanbot:kanbot@localhost:5432/kanbot (default)"),
        "OLLAMA_HOST": os.environ.get("OLLAMA_HOST", "http://localhost:11434 (default)"),
        "KANBOT_OPENAI_BASE_URL": os.environ.get("KANBOT_OPENAI_BASE_URL"),
        "KANBOT_MODEL_FAST": os.environ.get("KANBOT_MODEL_FAST", "(provider default)"),
        "KANBOT_MODEL_DAILY": os.environ.get("KANBOT_MODEL_DAILY", "(provider default)"),
        "OPENAI_API_KEY": mask(os.environ.get("OPENAI_API_KEY")),
        "ANTHROPIC_API_KEY": mask(os.environ.get("ANTHROPIC_API_KEY")),
        "OPENROUTER_API_KEY": mask(os.environ.get("OPENROUTER_API_KEY")),
    }
    if a.json:
        _out({"ok": True, "config": cfg}, True)
    else:
        print(f"{BOLD} kanbot config {RESET} {DIM}(env > default; secrets masked){RESET}")
        for k, v in cfg.items():
            print(f"  {k:<24} {v}")
    return 0


def cmd_run(a) -> int:
    """Launch the autonomous conductor loop. Sets run-mode from flags, then hands
    off to core.conductor (the real scheduler). AUDIT: --live requires explicit
    intent; without it we stay in dry-run (propose + park)."""
    if a.live:
        # AUDIT MODE: flipping to live execution is privileged — confirm first.
        from cli import gumfx
        if not gumfx.confirm(
                "Flip KanBot to LIVE? Approved cards will EXECUTE shell calls "
                "(sandboxed). Continue?", default_yes=False):
            print(f"{YEL}aborted — staying in dry-run.{RESET}")
            return 0
        os.environ["KANBOT_DRY_RUN"] = "0"
    elif a.dry_run:
        os.environ["KANBOT_DRY_RUN"] = "1"
    if a.swarm:
        os.environ["KANBOT_SWARM_N"] = str(a.swarm)
    mode = "LIVE — approved cards EXECUTE through the sandbox" if a.live else "dry-run (propose + park)"
    print(f"{BOLD}[kanbot run]{RESET} provider={os.environ.get('KANBOT_PROVIDER','ollama')} "
          f"| mode={mode} | swarm={os.environ.get('KANBOT_SWARM_N','1')}")
    if a.demo:
        print(f"{YEL}--demo:{RESET} conductor not started (no DB in demo). "
              "Run without --demo on the M4 to drive the real loop.")
        return 0
    from core.conductor import main as conductor_main
    try:
        asyncio.run(conductor_main())
    except KeyboardInterrupt:
        print("\n[kanbot run] conductor stopped (in-flight leases reaped back to ready).")
    return 0


# --------------------------------------------------------------------------- #
# arg parsing
# --------------------------------------------------------------------------- #
def build_parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(prog="kanbot",
                                description="KanBot — the board IS the runtime. CLI + agent harness.")
    p.add_argument("--demo", action="store_true",
                   help="use a seeded in-memory board (no DB/provider needed)")
    sub = p.add_subparsers(dest="cmd", required=True)

    def add(name, fn, is_async, help):
        sp = sub.add_parser(name, help=help)
        sp.set_defaults(fn=fn, is_async=is_async)
        return sp

    sp = add("board", cmd_board, True, "show the board"); sp.add_argument("--json", action="store_true"); sp.add_argument("--demo", action="store_true")
    sp = add("tui", cmd_tui, False, "interactive curses board"); sp.add_argument("--demo", action="store_true")
    sp = add("agent", cmd_agent, False, "JSONL line-protocol loop (agent-driven)"); sp.add_argument("--demo", action="store_true")
    sp = add("card", cmd_card, True, "card detail"); sp.add_argument("id"); sp.add_argument("--json", action="store_true")
    sp = add("move", cmd_move, True, "guarded transition"); sp.add_argument("id"); sp.add_argument("column"); sp.add_argument("-m", "--reason", default=""); sp.add_argument("--json", action="store_true")
    sp = add("claim", cmd_claim, True, "claim a ready card"); sp.add_argument("id"); sp.add_argument("worker"); sp.add_argument("--json", action="store_true")
    sp = add("approve", cmd_approve, True, "unblock a proposal to ready"); sp.add_argument("id"); sp.add_argument("--json", action="store_true")
    sp = add("block", cmd_block, True, "halt a card"); sp.add_argument("id"); sp.add_argument("reason"); sp.add_argument("--json", action="store_true")
    sp = add("human", cmd_human, True, "flag for human decision"); sp.add_argument("id"); sp.add_argument("question"); sp.add_argument("--json", action="store_true")
    sp = add("new", cmd_new, True, "create a card"); sp.add_argument("title"); sp.add_argument("-s", "--spec", default=""); sp.add_argument("-c", "--criteria", nargs="*"); sp.add_argument("--json", action="store_true")
    sp = add("run", cmd_run, False, "autonomous conductor loop"); g = sp.add_mutually_exclusive_group(); g.add_argument("--dry-run", action="store_true"); g.add_argument("--live", action="store_true"); sp.add_argument("--swarm", type=int)
    sp = add("providers", cmd_providers, False, "list providers + active + models"); sp.add_argument("--json", action="store_true")
    sp = add("doctor", cmd_doctor, True, "preflight the active config"); sp.add_argument("--json", action="store_true")
    sp = add("config", cmd_config, False, "show resolved config (secrets masked)"); sp.add_argument("--json", action="store_true")
    return p


def main(argv=None) -> int:
    # Existing process env wins. Tests import cli.tui without calling this,
    # so the repo .env cannot leak model overrides into pytest.
    from cli.tui import _load_repo_env
    _load_repo_env()
    parser = build_parser()
    # --demo is accepted both globally (before the subcommand) and on
    # board/tui/agent (after it). argparse would let the second occurrence
    # clobber the first with False, so OR the two: if it appears anywhere, honor it.
    raw = list(sys.argv[1:] if argv is None else argv)
    demo_anywhere = "--demo" in raw
    args = parser.parse_args(argv)
    for attr in ("demo", "json"):
        if not hasattr(args, attr):
            setattr(args, attr, False)
    if demo_anywhere:
        args.demo = True
    try:
        if args.is_async:
            return asyncio.run(args.fn(args))
        return args.fn(args)
    except KeyboardInterrupt:
        return 130
    except Exception as e:
        payload = {"ok": False, "error": f"{type(e).__name__}: {e}"}
        if getattr(args, "json", False):
            print(json.dumps(payload))
        else:
            print(f"{RED}✗ {payload['error']}{RESET}", file=sys.stderr)
        return 1


if __name__ == "__main__":
    raise SystemExit(main())