squad-heartbeat

Shared context: read ../squad/shared.md for pipeline levels, status transitions, API endpoints, error handling, and agent context flow. Schema: read ../squad/schema.md for full DB schema, column descriptions, and JSON field formats.

`/squad-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 event to each task's activity stream.

Defaults: --days 3, all active projects, writes a Heartbeat activity event per stagnant task. --dry-run: report only, no activity writes.

Procedure

① Auth & Argument Setup

   Load credentials using the standard shared.md pattern:

AUTH_TOKEN="${SQUAD_AUTH_TOKEN:-}" [ -z "$AUTH_TOKEN" ] && [ -f "$HOME/.squad/auth" ] && AUTH_TOKEN=$(grep '^SQUAD_AUTH_TOKEN=' "$HOME/.squad/auth" | cut -d= -f2-) BASE_URL="${SQUAD_BASE_URL:-}" [ -z "$BASE_URL" ] && [ -f "$HOME/.squad/config" ] && BASE_URL=$(grep '^SQUAD_BASE_URL=' "$HOME/.squad/config" | cut -d= -f2-) BASE_URL="${BASE_URL:-https://steloit-squad.vercel.app}" AUTH_HEADER=(); [ -n "$AUTH_TOKEN" ] && AUTH_HEADER=(-H "Authorization: Bearer $AUTH_TOKEN")

Parse CLI arguments:

--project X → scan only project X (default: all active projects)
--days N → stagnation threshold in days (default: 3)
--dry-run → report only, do not write activity events

② Fetch Projects

If --project X specified: Validate project exists: curl -s "${AUTH_HEADER[@]}" "$BASE_URL/api/projects/$X" If 404 → print error "Project '$X' not found." and exit. PROJECTS=("$X")

Else (all projects): ALL=$(curl -s "${AUTH_HEADER[@]}" "$BASE_URL/api/projects") Extract active projects: PROJECTS = jq '.projects[] | select(.status == "active") | .id' from ALL

③ Fetch Board per Project (full view)

For each project P in PROJECTS: BOARD=$(curl -s "${AUTH_HEADER[@]}" "$BASE_URL/api/board?project=$P")

 Collect tasks from columns: todo, plan, plan_review, impl, impl_review, test
 SKIP the done column entirely.

 If project has 0 tasks across all active columns → skip silently, continue.

④ Extract Last Activity Timestamp per Task

The board list does NOT carry the activity stream, so read each task's events from the dedicated activity reader (newest event's created_at), falling back to created_at:

 # Newest event's created_at (events come back chronological ASC → take the last one).
 curl -s "${AUTH_HEADER[@]}" "$BASE_URL/api/task/$ID/activity?project=$P"

 python3 -c "
 import json, sys
 resp = json.loads(sys.stdin.read())
 events = resp.get('activity') or []
 stamps = [e.get('created_at', '') for e in events if isinstance(e, dict)]
 stamps = [t for t in stamps if t]
 if stamps:
     print(max(stamps))
 else:
     # No events → caller falls back to created_at from the board row.
     print('')
 "

 If the activity read fails or returns no events, fall back to the task's `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 activity events written)"

⑦ Write Heartbeat activity events (skip if --dry-run)

For each stagnant task, append ONE activity event — a single atomic POST, no read-modify-write:

 python3 -c "
 import subprocess, json, sys

 task_id = sys.argv[1]
 project = sys.argv[2]
 days = int(sys.argv[3])
 status = sys.argv[4]
 last_ts = sys.argv[5]
 base_url = sys.argv[6]
 auth_token = sys.argv[7]

 auth_header = ['-H', f'Authorization: Bearer {auth_token}'] if auth_token else []
 body = json.dumps({
     'actor': 'Heartbeat',
     'model': 'system',
     'message': f'⚠️ Stagnant {days} days in {status}. Last activity: {last_ts}',
 })
 subprocess.run(
     ['curl', '-s', *auth_header, '-X', 'POST',
      f'{base_url}/api/task/{task_id}/activity?project={project}',
      '-H', 'Content-Type: application/json',
      '-d', body],
     capture_output=True
 )
 print(f'  Heartbeat written to task #{task_id}')
 " "$TASK_ID" "$PROJECT" "$DAYS" "$STATUS" "$LAST_TS" "$BASE_URL" "$AUTH_TOKEN"

Print: "Heartbeat activity events 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 subprocess, json, sys, datetime, re

# ── 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

# ── Auth setup (tool-agnostic: env → ~/.squad files → default) ────
import pathlib, os
base_url = os.environ.get("SQUAD_BASE_URL", "")
auth_token = os.environ.get("SQUAD_AUTH_TOKEN", "")

auth_file = pathlib.Path.home() / ".squad" / "auth"
config_file = pathlib.Path.home() / ".squad" / "config"
for f in (auth_file, config_file):
    if not f.exists():
        continue
    for line in f.read_text().splitlines():
        if not auth_token and line.startswith("SQUAD_AUTH_TOKEN="):
            auth_token = line.split("=", 1)[1].strip()
        elif not base_url and line.startswith("SQUAD_BASE_URL="):
            base_url = line.split("=", 1)[1].strip()
base_url = base_url or "https://steloit-squad.vercel.app"

def curl_get(url):
    cmd = ["curl", "-s", url]
    if auth_token:
        cmd += ["-H", f"Authorization: Bearer {auth_token}"]
    r = subprocess.run(cmd, capture_output=True, text=True)
    return json.loads(r.stdout)

def curl_post(url, payload):
    cmd = ["curl", "-s", "-X", "POST", url, "-H", "Content-Type: application/json", "-d", json.dumps(payload)]
    if auth_token:
        cmd += ["-H", f"Authorization: Bearer {auth_token}"]
    subprocess.run(cmd, capture_output=True)

# ── Fetch projects ───────────────────────────────────────────────
if project_filter:
    try:
        proj_data = curl_get(f"{base_url}/api/projects/{project_filter}")
        if "error" in proj_data:
            print(f"Error: Project '{project_filter}' not found.")
            sys.exit(1)
        projects = [project_filter]
    except Exception:
        print(f"Error: Project '{project_filter}' not found.")
        sys.exit(1)
else:
    all_proj = curl_get(f"{base_url}/api/projects")
    projects = [p["id"] for p in all_proj.get("projects", []) if p.get("status") == "active"]

if not projects:
    print("No active projects found.")
    sys.exit(0)

# ── Scan boards ──────────────────────────────────────────────────
now = datetime.datetime.now(datetime.timezone.utc)
active_columns = ["todo", "plan", "plan_review", "impl", "impl_review", "test"]
stagnant_tasks = []

for proj in projects:
    try:
        board = curl_get(f"{base_url}/api/board?project={proj}")
    except Exception:
        print(f"Warning: failed to fetch board for project '{proj}', skipping.", file=sys.stderr)
        continue

    for col in active_columns:
        tasks = board.get(col, [])
        if not isinstance(tasks, list):
            continue
        for task in tasks:
            task_id = task.get("id")
            title = task.get("title", "(untitled)")
            status = task.get("status", col)
            created_at = task.get("created_at", "")

            # Extract last activity timestamp from the dedicated activity reader
            # (the board list does not carry the activity stream). Newest event's created_at.
            last_ts = None
            read_error = False
            try:
                resp = curl_get(f"{base_url}/api/task/{task_id}/activity?project={proj}")
                events = resp.get("activity") if isinstance(resp, dict) else None
                if isinstance(events, list) and events:
                    stamps = [e.get("created_at", "") for e in events if isinstance(e, dict)]
                    stamps = [t for t in stamps if t]
                    if stamps:
                        last_ts = max(stamps)
            except (json.JSONDecodeError, TypeError, OSError):
                read_error = True

            if last_ts is None:
                last_ts = created_at
                if read_error:
                    print(f"Warning: task #{task_id} activity read failed, falling back to created_at", file=sys.stderr)

            if not last_ts:
                print(f"Warning: task #{task_id} has no timestamp at all, skipping", file=sys.stderr)
                continue

            # Parse timestamp and compute days
            try:
                # Normalize: drop fractional seconds + map 'Z' → '+00:00' (fromisoformat is strict pre-3.11),
                # then coerce to UTC-aware so it compares with `now` (also UTC-aware).
                clean_ts = re.sub(r"(T\d\d:\d\d:\d\d)\.\d+", r"\1", last_ts.strip()).replace("Z", "+00:00")
                ts_dt = datetime.datetime.fromisoformat(clean_ts)
                if ts_dt.tzinfo is None:
                    ts_dt = ts_dt.replace(tzinfo=datetime.timezone.utc)
            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)

# Sort by days descending
stagnant_tasks.sort(key=lambda t: t["days"], reverse=True)

# Markdown table
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 activity events written)"
print(summary)

# ── Write Heartbeat activity events ───────────────────────────────
if dry_run:
    sys.exit(0)

print("")
written = 0
for t in stagnant_tasks:
    try:
        # One atomic POST per task \u2014 no read-modify-write.
        curl_post(
            f"{base_url}/api/task/{t['id']}/activity?project={t['project']}",
            {
                "actor": "Heartbeat",
                "model": "system",
                "message": f"\u26a0\ufe0f Stagnant {t['days']} days in {t['status']}. Last activity: {t['last_ts']}",
            },
        )
        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"\nHeartbeat activity events written for {written} tasks.")
PYEOF

About this skill

/squad-heartbeat [--project X] [--days N] [--dry-run] -- Stagnant Task Detection

Procedure

`/squad-heartbeat [--project X] [--days N] [--dry-run]` -- Stagnant Task Detection