By 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.mdfor DB path, pipeline levels, status transitions, DB operations, error handling, and agent context flow. Schema: read../kanban/schema.mdfor 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:
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