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