File size: 5,150 Bytes
587a933 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 | #!/usr/bin/env python3
"""
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
# allow running as `python cli/kb.py` from repo root
sys.path.insert(0, ".")
from core import db # noqa: E402
from core.contract import Column, Role # noqa: E402
from core.mcp_server import ( # noqa: E402
block_card,
comment,
create_card,
get_card,
list_column,
move_card,
)
from core.reaper import sweep_once # noqa: E402
from core.workers import approve_and_run # noqa: E402
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()
|