/drain-queue

name: drain-queue
description: "Autonomous queue drainer — picks the top /triage cluster, applies safety gates, drains via /implement --issues, pushes, deploys."
argument-hint: "[--dry-run] [--cluster N1,N2,...]"
user-invocable: true
user_facing: true
allowed-tools: [Bash, Read, SlashCommand, PushNotification]

Drain Queue: Autonomous /triage → /implement → push → deploy wrapper

`/drain-queue` is a **thin orchestration wrapper around the existing executor** (`/implement --issues`). It does NOT re-implement batch logic. It adds the guardrails that `/implement --issues` does not have because the human picks issues for `/implement`; `/drain-queue` picks them autonomously, so it needs an extra safety layer.

> One invocation = one drain attempt. Recurrence is opt-in via the user-level > `/loop` or `/schedule` skills. This command does NOT self-loop.

Six guardrails

| # | Gate | Threshold (module constant) | |---|---------------------------------------|----------------------------------| | 1 | Daily drain-count + wall-clock budget | `MAX_DRAINS_PER_DAY=10`, `MAX_WALL_SECONDS_PER_DAY=14400` | | 2 | Cluster severity | only `low`, `info`, `medium` (`AUTO_DRAINABLE_SEVERITY`); `high` blocks (ADR-002 Phase D) | | 3 | Hydrated cluster labels (tag gate) | intersection with `HUMAN_GATE_TAGS` blocks | | 4 | Cluster size | `MAX_CLUSTER_SIZE_AUTO_DRAINABLE=5` | | 5 | Circuit breaker | 2 consecutive failures → 4h pause; 3 in 24h → 24h pause | | 6 | Push / deploy gates | clean worktree + non-divergent remote required |

Module constants live in `plugins/autonomous-dev/lib/drain_queue_state.py`. They are the single source of truth — never duplicate the value in the markdown.

Implementation

Execute the 12 STEPs below in order. STEP 6 delegates to `/implement --issues N1,N2,...` (the existing batch executor at `commands/implement-batch.md`) — this command is a wrapper, not a new drainer. All Python helpers live in `plugins/autonomous-dev/lib/drain_runner.py` and `lib/drain_queue_state.py`. The markdown is the orchestration layer; Python helpers do the subprocess and state work.

# Each STEP in this command invokes the Python helpers below. STEP 1 example:
python3 -c "
import sys
sys.path.insert(0, 'plugins/autonomous-dev/lib')
from drain_runner import check_clean_worktree, default_branch
from drain_queue_state import PauseFlag, DrainBudget
from pathlib import Path
import os
repo = Path.cwd()
env = dict(os.environ)
# Pre-flight checks happen here; full 12-STEP playbook below
"

ARGUMENTS

ARGUMENTS: {{ARGUMENTS}}

Recognized flags:

• `--dry-run` — run only the read-only pre-flight checks (worktree clean,

default branch resolvable, budget within cap, pause flag absent) and report. No `gh`/`git` mutating calls; no state writes.

• `--cluster N1,N2,...` — bypass cluster selection and use the specified issue numbers.

Still applies safety gates. Used by workflows to pre-select clusters.

STEP 1: Pre-flight checks

Run all four checks. **Any fail → STOP and emit notification.**

python3 - <<'PY'
import sys, time
from pathlib import Path
sys.path.insert(0, "plugins/autonomous-dev/lib")

from drain_queue_state import PauseFlag, DrainBudget
from drain_runner import check_clean_worktree, append_stop_notification, _build_env
from pipeline_state import get_legacy_sentinel_path

repo = Path.cwd().resolve()
env = _build_env(repo)
log_dir = repo / ".claude" / "local"
log_dir.mkdir(parents=True, exist_ok=True, mode=0o700)

# 1a. Working tree clean?
if not check_clean_worktree(repo, env):
    append_stop_notification("STEP 1: working tree not clean", log_dir)
    print("STOP: working tree not clean", flush=True)
    sys.exit(1)

# 1b. Universal bypass marker absent? (.claude/.bypass)
if (repo / ".claude" / ".bypass").exists():
    append_stop_notification("STEP 1: .claude/.bypass present — autonomous drain disabled", log_dir)
    print("STOP: .claude/.bypass present", flush=True)
    sys.exit(1)

# 1c. Pipeline-state sentinel fresh (within last 1h) → another /implement is in flight.
#     Uses the PER-REPO sentinel path (Issue #1206). DO NOT hardcode /tmp/...
sentinel = get_legacy_sentinel_path(repo)
if sentinel.exists():
    age = time.time() - sentinel.stat().st_mtime
    if age < 3600:
        append_stop_notification(
            f"STEP 1: /implement in-flight (sentinel age={int(age)}s)", log_dir
        )
        print(f"STOP: /implement in-flight (sentinel age={int(age)}s)", flush=True)
        sys.exit(1)

# 1d. PauseFlag active?
paused, reason = PauseFlag.load(repo).is_active()
if paused:
    append_stop_notification(f"STEP 1: pause flag active — {reason}", log_dir)
    print(f"STOP: pause flag active — {reason}", flush=True)
    sys.exit(1)

print("STEP 1: pre-flight passed", flush=True)
PY

If the snippet exited non-zero, emit a `PushNotification:` tool line based on the stop reason just appended to `.claude/local/drain_notifications.jsonl` and exit `/drain-queue`.

STEP 2: Budget check (drain count + wall-clock)

python3 - <<'PY'
import sys
from pathlib import Path
sys.path.insert(0, "plugins/autonomous-dev/lib")
from drain_queue_state import DrainBudget
from drain_runner import append_stop_notification

repo = Path.cwd().resolve()
log_dir = repo / ".claude" / "local"

budget = DrainBudget.load(repo)
blocked, reason = budget.check_or_block()
if blocked:
    append_stop_notification(f"STEP 2: daily budget — {reason}", log_dir)
    print(f"STOP: daily budget — {reason}", flush=True)
    sys.exit(1)
print(f"STEP 2: budget OK — today {budget.today_drains} drains, "
      f"{budget.today_wall_seconds:.0f}s used", flush=True)
PY

On STOP, emit `PushNotification:` and exit.

STEP 3: Cluster selection (via /triage or --cluster)

Parse arguments and either use explicit --cluster or run triage:

python3 - <<'PY'
import