| |
| """ |
| 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 |
|
|
| |
| |
| 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)) |
|
|
|
|
| |
| |
| |
| 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))) |
|
|
|
|
| |
| |
| |
| 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): |
| return v |
|
|
|
|
| |
| |
| |
| 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 |
|
|
| |
| 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 |
|
|
| |
| 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: |
| |
| 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 |
|
|
|
|
| |
| |
| |
| 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: |
| |
| |
| from cli.tui import _load_repo_env |
| _load_repo_env() |
| parser = build_parser() |
| |
| |
| |
| 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()) |
|
|
