| """ |
| Agent workers — the three roles that make the board move itself. |
| |
| PLANNER : decomposes a spec into child executor cards, then marks itself done. |
| EXECUTOR : does the work. In DRY-RUN (v0 default) it PROPOSES the tool calls as |
| a thread comment and BLOCKS for human approval instead of executing. |
| When free, it runs the plan and posts an artifact. |
| CRITIC : scores the artifact against acceptance_criteria -> pass/fail. |
| |
| Every worker only ever touches the board through the same operations the MCP |
| tools expose (imported directly here for the in-process Conductor; external |
| clients hit them over MCP). One contract, four faces. |
| """ |
| from __future__ import annotations |
|
|
| import asyncio |
| import os |
| import uuid |
| from datetime import datetime, timezone |
| from pathlib import Path |
|
|
| def _load_repo_env() -> None: |
| """Pick up the repo .env before DRY_RUN is read. Existing env wins.""" |
| 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("'")) |
|
|
| _load_repo_env() |
|
|
| from . import db, llm, sandbox |
| from .contract import ( |
| LEASE_SECONDS_DEFAULT, |
| MAX_ATTEMPTS_DEFAULT, |
| Card, |
| Column, |
| Role, |
| ThreadEntry, |
| ) |
| from .mcp_server import ( |
| block_card, |
| comment, |
| move_card, |
| request_human, |
| spawn_subtask, |
| ) |
|
|
| DRY_RUN = os.environ.get("KANBOT_DRY_RUN", "1") == "1" |
| SWARM_N = max(1, int(os.environ.get("KANBOT_SWARM_N", "1"))) |
| MAX_ATTEMPTS = int(os.environ.get("KANBOT_MAX_ATTEMPTS", str(MAX_ATTEMPTS_DEFAULT))) |
| WORKER_ID = lambda role: f"{role}-{uuid.uuid4().hex[:6]}" |
|
|
|
|
| def _now() -> str: |
| return datetime.now(timezone.utc).isoformat() |
|
|
|
|
| |
| |
| |
| PLANNER_SYS = ( |
| "You are KanBot's PLANNER. Decompose a task spec into an ordered list of " |
| "concrete, independently-executable subtasks. Each subtask must have a short " |
| "title, a precise spec, and 1-3 acceptance_criteria that are objectively " |
| "checkable. Prefer 2-6 subtasks. Return JSON: " |
| '{"subtasks":[{"title":"","spec":"","acceptance_criteria":[""],"heavy":false}]}' |
| ) |
|
|
|
|
| async def run_planner(card: Card) -> None: |
| plan = await llm.chat_json(PLANNER_SYS, |
| f"TASK: {card.title}\nSPEC: {card.spec}", tier="daily") |
| subtasks = plan.get("subtasks", []) |
| if not subtasks: |
| await block_card(card.id, "planner produced no subtasks", actor="planner") |
| return |
| for st in subtasks: |
| await spawn_subtask( |
| parent=card.id, title=st["title"], spec=st.get("spec", ""), |
| role="executor", acceptance_criteria=st.get("acceptance_criteria", []), |
| tags=["heavy"] if st.get("heavy") else []) |
| await comment(card.id, f"decomposed into {len(subtasks)} subtasks", |
| author="planner", kind="comment") |
| await move_card(card.id, "done", actor="planner", reason="decomposition complete") |
|
|
|
|
| |
| |
| |
| async def promote_to_ready(card: Card) -> None: |
| await move_card(card.id, "ready", actor="system", reason="queued for execution") |
|
|
|
|
| |
| |
| |
| EXECUTOR_SYS = ( |
| "You are KanBot's EXECUTOR. Given a subtask, produce a concrete PLAN OF TOOL " |
| "CALLS to accomplish it (shell, code edits, deploys, searches). Then produce " |
| "the resulting ARTIFACT (the actual output/result). Return JSON: " |
| '{"tool_calls":[{"tool":"","args":"","why":""}],"artifact":"the concrete result"}' |
| ) |
|
|
|
|
| async def _plan_one(card: Card, worker: str) -> dict: |
| """One executor's candidate: a plan of tool calls + a claimed artifact.""" |
| return await llm.chat_json( |
| EXECUTOR_SYS, |
| f"SUBTASK: {card.title}\nSPEC: {card.spec}\n" |
| f"ACCEPTANCE: {card.acceptance_criteria}", |
| tier="daily", heavy=card.is_heavy()) |
|
|
|
|
| async def claim_and_execute(card: Card) -> None: |
| """Claim a READY card and produce work. With SWARM_N>1, several executors |
| race the same card in parallel; the critic later ranks their candidates. |
| DRY-RUN gates ALL of them: nothing runs until a human approves.""" |
| lead = WORKER_ID("executor") |
| from .mcp_server import claim_card |
| res = await claim_card(card.id, lead, LEASE_SECONDS_DEFAULT) |
| if '"error"' in res: |
| return |
|
|
| |
| workers_ids = [lead] + [WORKER_ID("executor") for _ in range(SWARM_N - 1)] |
| plans = await asyncio.gather( |
| *[_plan_one(card, w) for w in workers_ids], return_exceptions=True) |
| candidates = [(w, p) for w, p in zip(workers_ids, plans) |
| if isinstance(p, dict)] |
| if not candidates: |
| await block_card(card.id, "all executor candidates failed to plan", actor=lead) |
| return |
|
|
| if DRY_RUN: |
| |
| for w, plan in candidates: |
| tcs = plan.get("tool_calls", []) |
| proposal = (f"CANDIDATE from {w} (dry-run — awaiting approval), " |
| f"{len(tcs)} tool call(s):\n" + "\n".join( |
| f" - {tc.get('tool')}({tc.get('args')}) # {tc.get('why','')}" |
| for tc in tcs)) |
| await comment(card.id, proposal, author=w, kind="tool_call") |
| await request_human( |
| card.id, |
| f"Approve execution for '{card.title}'? " |
| f"{len(candidates)} candidate(s) proposed. " |
| f"kb approve {card.id} | kb reject {card.id}", |
| actor=lead) |
| await block_card(card.id, "dry-run: awaiting human approval", actor=lead) |
| return |
|
|
| |
| sandbox.ensure_root() |
| scored: list[tuple[str, str]] = [] |
| for w, plan in candidates: |
| logs = [] |
| for tc in plan.get("tool_calls", []): |
| cmd = str(tc.get("args", "")) if tc.get("tool") in ("shell", "bash", "sh") \ |
| else "" |
| if cmd: |
| result = await sandbox.run(cmd) |
| await comment(card.id, |
| f"[{w}] $ {cmd}\n{result['stdout'] or result['stderr']}", |
| author=w, kind="tool_result") |
| logs.append(f"$ {cmd} -> code {result['code']}") |
| artifact = plan.get("artifact", "") + ("\n\n[exec log]\n" + "\n".join(logs) |
| if logs else "") |
| scored.append((w, artifact)) |
|
|
| |
| |
| |
| if not any((art or "").strip() for _, art in scored): |
| await block_card( |
| card.id, |
| "executor produced no artifact and ran no tool calls — nothing to review", |
| actor=lead) |
| return |
|
|
| |
| if len(scored) == 1: |
| w, artifact = scored[0] |
| await comment(card.id, artifact, author=w, kind="artifact") |
| else: |
| for w, artifact in scored: |
| await comment(card.id, f"[candidate {w}]\n{artifact}", |
| author=w, kind="artifact") |
| await comment(card.id, f"SWARM: {len(scored)} candidates — critic to pick winner", |
| author=lead, kind="comment") |
| await move_card(card.id, "review", actor=lead, worker=lead, |
| reason="execution complete") |
|
|
|
|
| async def approve_and_run(card_id: str) -> None: |
| """Human approved a dry-run proposal (via CLI/Telegram): unblock -> ready. |
| On the next pass the executor re-plans; flip KANBOT_DRY_RUN=0 to let it run |
| free (through the sandbox).""" |
| await move_card(card_id, "ready", actor="human:remy", |
| reason="human approved proposed tool calls") |
|
|
|
|
| |
| |
| |
| CRITIC_SYS = ( |
| "You are KanBot's CRITIC. Judge whether the ARTIFACT satisfies EVERY listed " |
| "acceptance criterion. Be strict and specific. Return JSON: " |
| '{"verdict":"pass"|"fail","reasons":["per-criterion note"]}' |
| ) |
|
|
| SWARM_CRITIC_SYS = ( |
| "You are KanBot's CRITIC judging a SWARM. Multiple candidate artifacts attempt " |
| "the same task. Pick the single best one against the acceptance criteria and " |
| "judge whether it passes. Return JSON: " |
| '{"winner_index":<0-based int>,"verdict":"pass"|"fail","reasons":["..."]}' |
| ) |
|
|
|
|
| async def run_critic(card: Card) -> None: |
| artifacts = [(e.author, e.content) for e in card.thread if e.kind == "artifact"] |
| if not artifacts: |
| await block_card(card.id, "no artifact to review", actor="critic") |
| return |
|
|
| |
| if len(artifacts) > 1: |
| listing = "\n\n".join(f"[CANDIDATE {i}] by {a}\n{c}" |
| for i, (a, c) in enumerate(artifacts)) |
| pick = await llm.chat_json( |
| SWARM_CRITIC_SYS, |
| f"CANDIDATES:\n{listing}\n\nACCEPTANCE CRITERIA:\n{card.acceptance_criteria}", |
| tier="daily") |
| idx = int(pick.get("winner_index", 0)) |
| idx = idx if 0 <= idx < len(artifacts) else 0 |
| winner_author, _ = artifacts[idx] |
| v = pick.get("verdict", "fail") |
| reasons = "; ".join(pick.get("reasons", [])) |
| await comment(card.id, |
| f"SWARM WINNER: candidate {idx} ({winner_author}) — [{v.upper()}] {reasons}", |
| author="critic", kind="eval") |
| else: |
| _, artifact = artifacts[0] |
| verdict = await llm.chat_json( |
| CRITIC_SYS, |
| f"ARTIFACT:\n{artifact}\n\nACCEPTANCE CRITERIA:\n{card.acceptance_criteria}", |
| tier="daily") |
| v = verdict.get("verdict", "fail") |
| reasons = "; ".join(verdict.get("reasons", [])) |
| await comment(card.id, f"[{v.upper()}] {reasons}", author="critic", kind="eval") |
|
|
| if v == "pass": |
| await move_card(card.id, "done", actor="critic", |
| critic_verdict="pass", reason="acceptance criteria met") |
| else: |
| |
| |
| |
| if MAX_ATTEMPTS and card.attempt >= MAX_ATTEMPTS - 1: |
| await request_human( |
| card.id, |
| f"'{card.title}' failed critic {card.attempt + 1}x " |
| f"(cap {MAX_ATTEMPTS}). Last feedback: {reasons}. " |
| f"kb approve {card.id} to force-retry | kb block {card.id} to abandon.", |
| actor="critic") |
| await block_card( |
| card.id, |
| f"retry cap reached ({card.attempt + 1}/{MAX_ATTEMPTS}): {reasons}", |
| actor="critic") |
| return |
| await move_card(card.id, "ready", actor="critic", |
| critic_verdict="fail", reason=f"retry: {reasons}") |
|
|
|
|
| |
| |
| |
| async def check_parent_completion(parent: Card) -> None: |
| children = [await db.get_card(cid) for cid in parent.children] |
| children = [c for c in children if c] |
| if children and all(c.column == Column.DONE for c in children): |
| if parent.column not in (Column.DONE,): |
| await comment(parent.id, "all subtasks done", author="system") |
| |
| if parent.has_artifact() and parent.column == Column.IN_FLIGHT: |
| await move_card(parent.id, "review", actor="system", |
| reason="all children complete") |
|
|
