| |
| """ |
| kb — the terminal face of KanBot. |
| |
| One thin client over the same board operations agents and Telegram use. Drive the |
| whole autonomous board from your shell: |
| |
| kb board # the whole board, column by column |
| kb col ready # list one column |
| kb card CARD-42 # full detail incl. inter-agent thread |
| kb new "title" --spec "..." [--planner] [--criteria a b] [--heavy] |
| kb move CARD-42 review [--reason "..."] |
| kb comment CARD-42 "note" |
| kb approve CARD-42 # approve a dry-run tool-call proposal |
| kb reject CARD-42 "reason" |
| kb block CARD-42 "reason" |
| kb reap # run one reaper sweep now |
| |
| Reads DATABASE_URL from env (see .env). Autonomous cards move themselves once the |
| Conductor is running; this is for you to seed, inspect, and gate. |
| """ |
| from __future__ import annotations |
|
|
| import argparse |
| import asyncio |
| import json |
| import sys |
|
|
| |
| sys.path.insert(0, ".") |
|
|
| from core import db |
| from core.contract import Column, Role |
| from core.mcp_server import ( |
| block_card, |
| comment, |
| create_card, |
| get_card, |
| list_column, |
| move_card, |
| ) |
| from core.reaper import sweep_once |
| from core.workers import approve_and_run |
|
|
| COLS = [c.value for c in Column] |
| _ICON = {"backlog": "·", "ready": "◦", "in-flight": "▸", "review": "?", |
| "blocked": "✕", "done": "✓"} |
|
|
|
|
| async def cmd_board(_a) -> None: |
| for col in COLS: |
| cards = await db.list_column(Column(col)) |
| print(f"\n{_ICON[col]} {col.upper()} ({len(cards)})") |
| for c in cards: |
| tag = f" [{','.join(c.tags)}]" if c.tags else "" |
| hum = " ⚑" if c.requires_human else "" |
| print(f" {c.id:<9} {c.title[:56]:<56}{tag}{hum}") |
|
|
|
|
| async def cmd_col(a) -> None: |
| print(await list_column(a.column)) |
|
|
|
|
| async def cmd_card(a) -> None: |
| data = json.loads(await get_card(a.id)) |
| if "error" in data: |
| print(data["error"]); return |
| print(f"{data['id']} [{data['column']}] role={data['role']} " |
| f"attempt={data['attempt']}") |
| print(f" {data['title']}") |
| if data["spec"]: |
| print(f" spec: {data['spec']}") |
| if data["acceptance_criteria"]: |
| print(" acceptance:") |
| for c in data["acceptance_criteria"]: |
| print(f" - {c}") |
| print(" ── thread ──") |
| for e in data["thread"]: |
| print(f" [{e['kind']}] {e['author']}: {e['content'][:200]}") |
|
|
|
|
| async def cmd_new(a) -> None: |
| res = json.loads(await create_card( |
| title=a.title, spec=a.spec or "", |
| role="planner" if a.planner else "executor", |
| acceptance_criteria=a.criteria or [], |
| tags=["heavy"] if a.heavy else [])) |
| print(f"created {res['id']} in {res['column']}" |
| + (" (planner will decompose once Conductor runs)" if a.planner else "")) |
|
|
|
|
| async def cmd_move(a) -> None: |
| print(await move_card(a.id, a.to, reason=a.reason or "")) |
|
|
|
|
| async def cmd_comment(a) -> None: |
| print(await comment(a.id, a.text)) |
|
|
|
|
| async def cmd_approve(a) -> None: |
| await approve_and_run(a.id) |
| print(f"approved {a.id} — requeued to ready") |
|
|
|
|
| async def cmd_reject(a) -> None: |
| print(await block_card(a.id, a.reason, actor="human:remy")) |
|
|
|
|
| async def cmd_block(a) -> None: |
| print(await block_card(a.id, a.reason, actor="human:remy")) |
|
|
|
|
| async def cmd_reap(_a) -> None: |
| n = await sweep_once() |
| print(f"reaper reclaimed {n} card(s)") |
|
|
|
|
| def build_parser() -> argparse.ArgumentParser: |
| p = argparse.ArgumentParser(prog="kb", description="KanBot board CLI") |
| sub = p.add_subparsers(dest="cmd", required=True) |
|
|
| sub.add_parser("board").set_defaults(fn=cmd_board) |
|
|
| c = sub.add_parser("col"); c.add_argument("column", choices=COLS); c.set_defaults(fn=cmd_col) |
| c = sub.add_parser("card"); c.add_argument("id"); c.set_defaults(fn=cmd_card) |
|
|
| c = sub.add_parser("new") |
| c.add_argument("title") |
| c.add_argument("--spec", default="") |
| c.add_argument("--planner", action="store_true", help="route to planner for decomposition") |
| c.add_argument("--criteria", nargs="*", default=[]) |
| c.add_argument("--heavy", action="store_true", help="tag for cloud executor escalation") |
| c.set_defaults(fn=cmd_new) |
|
|
| c = sub.add_parser("move"); c.add_argument("id"); c.add_argument("to", choices=COLS) |
| c.add_argument("--reason", default=""); c.set_defaults(fn=cmd_move) |
|
|
| c = sub.add_parser("comment"); c.add_argument("id"); c.add_argument("text"); c.set_defaults(fn=cmd_comment) |
| c = sub.add_parser("approve"); c.add_argument("id"); c.set_defaults(fn=cmd_approve) |
| c = sub.add_parser("reject"); c.add_argument("id"); c.add_argument("reason"); c.set_defaults(fn=cmd_reject) |
| c = sub.add_parser("block"); c.add_argument("id"); c.add_argument("reason"); c.set_defaults(fn=cmd_block) |
| sub.add_parser("reap").set_defaults(fn=cmd_reap) |
| return p |
|
|
|
|
| def main() -> None: |
| args = build_parser().parse_args() |
| asyncio.run(args.fn(args)) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
