--- name: kanban-heartbeat description: Scan kanban boards for stagnant tasks and optionally mark them. Detects tasks with no agent activity for N days (default 3), outputs a markdown report table, and appends Heartbeat entries to agent_log unless --dry-run. license: MIT --- > Shared context: read `../kanban/shared.md` for DB path, pipeline levels, status transitions, DB operations, error handling, and agent context flow. > Schema: read `../kanban/schema.md` for full DB schema, column descriptions, and JSON field formats. ## `/kanban-heartbeat [--project X] [--days N] [--dry-run]` -- Stagnant Task Detection Scan all active projects (or a single project) for tasks that have had no agent activity for N days. Output a markdown table of stagnant tasks and optionally append a Heartbeat warning entry to each task's `agent_log`. **Defaults**: `--days 3`, all active projects, writes to `agent_log`. **`--dry-run`**: report only, no `agent_log` modifications. ### Procedure ``` ① DB Setup & Argument Parsing Read project config and DB path: CONFIG=$(cat .claude/kanban.json 2>/dev/null || cat .codex/kanban.json 2>/dev/null) PROJECT=$(echo "$CONFIG" | python3 -c "import sys,json; d=json.load(sys.stdin); print(d['project'])" 2>/dev/null || basename "$(pwd)") DB="$HOME/.claude/kanban-dbs/${PROJECT}.db" Parse CLI arguments: - --project X → scan only project X (override PROJECT) - --days N → stagnation threshold in days (default: 3) - --dry-run → report only, do not write agent_log entries ② Fetch Projects If --project X specified: Validate project exists: sqlite3 "$DB" "SELECT id FROM projects WHERE id='$X'" If empty → print error "Project '$X' not found." and exit. PROJECTS=("$X") Else: PROJECTS = sqlite3 "$DB" "SELECT id FROM projects WHERE status='active'" If no active projects → scan using PROJECT from config ③ Fetch Tasks per Project (active columns only) For each project P in PROJECTS: sqlite3 -json "$DB" "SELECT id, title, status, agent_log, created_at FROM tasks WHERE project='$P' AND status IN ('todo','plan','plan_review','impl','impl_review','test') ORDER BY id" If project has 0 tasks → skip silently, continue. ④ Extract Last Activity Timestamp per Task For each task, use Python for safe JSON parsing of agent_log: python3 -c " import json, sys task = json.loads(sys.stdin.read()) agent_log_raw = task.get('agent_log') or '[]' try: log = json.loads(agent_log_raw) if isinstance(log, list) and len(log) > 0: timestamps = [e.get('timestamp', '') for e in log if isinstance(e, dict)] timestamps = [t for t in timestamps if t] if timestamps: print(max(timestamps)) sys.exit(0) except (json.JSONDecodeError, TypeError): print('PARSE_ERROR', file=sys.stderr) # Fallback to created_at print(task.get('created_at', '')) " Store: task ID, project, status, title, last_activity_ts ⑤ Compute Stagnation NOW = current UTC timestamp THRESHOLD = NOW - N days For each task: Parse last_activity_ts as datetime days_stagnant = (NOW - last_activity_ts).days If days_stagnant >= N → mark as stagnant If no stagnant tasks across all projects: Print "No stagnant tasks found." Exit. ⑥ Output Markdown Table Sort stagnant tasks by days_stagnant descending. Print: | ID | Project | Status | Days | Title | |----|---------|--------|------|-------| | 2100 | cpet.db | impl | 12 | Add export feature | | 2055 | today.bike | plan | 5 | Refactor route module | Print summary line: "**Heartbeat: X stagnant tasks found across Y projects.**" If --dry-run: append " (dry-run, no agent_log entries written)" ⑦ Write agent_log Entries (skip if --dry-run) For each stagnant task, append a Heartbeat entry using sqlite3 json_insert: NEW_ENTRY=$(python3 -c " import json, datetime entry = { 'agent': 'Heartbeat', 'model': 'system', 'message': f'⚠️ Stagnant {DAYS} days in {STATUS}. Last activity: {LAST_TS}', 'timestamp': datetime.datetime.utcnow().isoformat() + 'Z' } print(json.dumps(entry)) ") sqlite3 "$DB" "UPDATE tasks SET agent_log=json_insert(COALESCE(agent_log,'[]'), '\$[#]', json('$NEW_ENTRY')), updated_at=datetime('now') WHERE id=$TASK_ID AND project='$PROJECT'" echo " Heartbeat written to task #$TASK_ID" Print: "agent_log entries written for X tasks." ``` ### Full Implementation (Copy-Paste Script) The executing agent should run this as a single Python script for reliability: ```bash python3 - "$@" <<'PYEOF' import sqlite3 as sq, json, sys, datetime, re, os, pathlib # ── Parse arguments ────────────────────────────────────────────── args = sys.argv[1:] project_filter = None days_threshold = 3 dry_run = False i = 0 while i < len(args): if args[i] == "--project" and i + 1 < len(args): project_filter = args[i + 1]; i += 2 elif args[i] == "--days" and i + 1 < len(args): days_threshold = int(args[i + 1]); i += 2 elif args[i] == "--dry-run": dry_run = True; i += 1 else: i += 1 # ── DB Setup ───────────────────────────────────────────────────── import subprocess config_paths = [".claude/kanban.json", ".codex/kanban.json"] project_name = None for p in config_paths: if os.path.exists(p): try: d = json.loads(open(p).read()) project_name = d.get("project") break except Exception: pass if not project_name: project_name = os.path.basename(os.getcwd()) db_path = str(pathlib.Path.home() / ".claude" / "kanban-dbs" / f"{project_name}.db") if not os.path.exists(db_path): print(f"Error: DB not found at {db_path}. Run /kanban-init first.") sys.exit(1) conn = sq.connect(db_path) conn.row_factory = sq.Row # ── Fetch projects ─────────────────────────────────────────────── if project_filter: row = conn.execute("SELECT id FROM projects WHERE id=?", (project_filter,)).fetchone() if not row: # Fall back: check if project has tasks even without projects table entry count = conn.execute("SELECT count(*) FROM tasks WHERE project=?", (project_filter,)).fetchone()[0] if count == 0: print(f"Error: Project '{project_filter}' not found.") sys.exit(1) projects = [project_filter] else: rows = conn.execute("SELECT id FROM projects WHERE status='active'").fetchall() projects = [r["id"] for r in rows] if not projects: # Fall back to the project from config projects = [project_name] if not projects: print("No active projects found.") sys.exit(0) # ── Scan tasks ─────────────────────────────────────────────────── now = datetime.datetime.utcnow() active_columns = ("todo", "plan", "plan_review", "impl", "impl_review", "test") stagnant_tasks = [] for proj in projects: rows = conn.execute( "SELECT id, title, status, agent_log, created_at FROM tasks WHERE project=? AND status IN ({})".format( ",".join("?" * len(active_columns)) ), (proj, *active_columns) ).fetchall() for task in rows: task_id = task["id"] title = task["title"] or "(untitled)" status = task["status"] created_at = task["created_at"] or "" agent_log_raw = task["agent_log"] # Extract last activity timestamp last_ts = None parse_error = False if agent_log_raw: try: log = json.loads(agent_log_raw) if isinstance(agent_log_raw, str) else agent_log_raw if isinstance(log, list) and len(log) > 0: timestamps = [e.get("timestamp", "") for e in log if isinstance(e, dict)] timestamps = [t for t in timestamps if t] if timestamps: last_ts = max(timestamps) except (json.JSONDecodeError, TypeError): parse_error = True if last_ts is None: last_ts = created_at if parse_error: print(f"Warning: task #{task_id} has malformed agent_log, falling back to created_at", file=sys.stderr) if not last_ts: print(f"Warning: task #{task_id} has no timestamp, skipping", file=sys.stderr) continue # Parse timestamp try: clean_ts = re.sub(r"\.\d+", "", last_ts.replace("Z", "").replace("+00:00", "")) ts_dt = datetime.datetime.fromisoformat(clean_ts) except (ValueError, AttributeError): print(f"Warning: task #{task_id} has unparseable timestamp '{last_ts}', skipping", file=sys.stderr) continue days_stagnant = (now - ts_dt).days if days_stagnant >= days_threshold: stagnant_tasks.append({ "id": task_id, "project": proj, "status": status, "days": days_stagnant, "title": title, "last_ts": last_ts, }) # ── Output ─────────────────────────────────────────────────────── if not stagnant_tasks: print("No stagnant tasks found.") sys.exit(0) stagnant_tasks.sort(key=lambda t: t["days"], reverse=True) print("") print("| ID | Project | Status | Days | Title |") print("|----|---------|--------|------|-------|") for t in stagnant_tasks: print(f"| {t['id']} | {t['project']} | {t['status']} | {t['days']} | {t['title']} |") print("") project_set = set(t["project"] for t in stagnant_tasks) summary = f"**Heartbeat: {len(stagnant_tasks)} stagnant tasks found across {len(project_set)} projects.**" if dry_run: summary += " (dry-run, no agent_log entries written)" print(summary) # ── Write agent_log entries ────────────────────────────────────── if dry_run: sys.exit(0) print("") written = 0 for t in stagnant_tasks: try: row = conn.execute("SELECT agent_log FROM tasks WHERE id=? AND project=?", (t["id"], t["project"])).fetchone() try: log = json.loads(row["agent_log"] or "[]") except (json.JSONDecodeError, TypeError): log = [] log.append({ "agent": "Heartbeat", "model": "system", "message": f"⚠️ Stagnant {t['days']} days in {t['status']}. Last activity: {t['last_ts']}", "timestamp": now.isoformat() + "Z", }) conn.execute( "UPDATE tasks SET agent_log=?, updated_at=datetime('now') WHERE id=? AND project=?", (json.dumps(log), t["id"], t["project"]) ) conn.commit() print(f" Heartbeat written to task #{t['id']}") written += 1 except Exception as e: print(f" Error writing to task #{t['id']}: {e}", file=sys.stderr) print(f"\nagent_log entries written for {written} tasks.") conn.close() PYEOF ```