codegen-drain

#!/usr/bin/env bash
# codegen-drain — possession-mover between fleet nodes draining codegen/pitches/ready/.
#
# Possession is the claim: dispatch is `mv` (transfer, then delete the source),
# never `cp`. Exactly one copy of a pitch exists across the fleet at any time —
# two nodes can never build the same pitch, not by policy, by physics.
#
# Usage:
#   codegen-drain assign --slug=<slug> --node=<node> [--cwd=<dir>]
#   codegen-drain assign --auto [--cwd=<dir>]
#   codegen-drain status [--json] [--cwd=<dir>]
#   codegen-drain init [--cwd=<dir>]
#
# Subcommands:
#   assign    Move a pitch from THIS box's ready/ into <node>'s ready/. Ack
#             (checksum-verified same-fs mv on the far side) BEFORE deleting
#             the local copy. Refuses (fail-closed) on any doubt.
#             --auto: builds a fail-closed fleet dependency snapshot (every
#             declared node must answer), then partitions THIS box's ready/
#             into N lanes (N = reachable node count) via
#             `mix codegen.pitches.scope --fleet-safe --json`, then calls
#             the same single-slug transfer once per lane->node (lane i ->
#             reachable-node i, inventory order). GLOBAL-HOT and UNROUTED
#             slugs are printed by name and never placed — the
#             partitioner's own contract, unchanged. DEPENDENCY-BOUND slugs
#             (a blocks_on: edge to/from anything on the fleet, including a
#             dep outside this batch) are ALSO printed by name and never
#             placed — kept on the source node so the whole dependency
#             chain builds locally. A single-slug `assign --slug --node`
#             refuses (exit 1, named) when the target lacks a prerequisite
#             the slug needs, or already holds a dependent that would be
#             stranded. A failing transfer stops --auto immediately
#             (fail-closed); already-placed lanes stay placed, remaining
#             lanes are NOT attempted.
#   status    ssh every configured node and print which pitches/build state
#             each holds — restores the queue visibility single-copy costs.
#   init      Interactively write codegen/drain-nodes.yaml (gitignored,
#             machine-local fleet inventory).
#
# Options (assign):
#   --slug=<slug>   required (unless --auto): must exist at
#                    codegen/pitches/ready/<slug>.md locally
#   --node=<node>   required (unless --auto): must be a key in
#                    codegen/drain-nodes.yaml
#   --auto          partition ready/ across all reachable nodes and place
#                    every lane (mutually exclusive with --slug/--node)
#
# Options (status):
#   --json          emit machine-readable JSON instead of table text
#
# Options (all subcommands):
#   --cwd=<dir>     codegen repo root (default: $PWD) — codegen-drain is
#                    installed as a FLAT copy (no sibling codegen/ tree), so
#                    it must never derive the repo root from its own install
#                    location; mirrors codegen-build's --cwd contract.
#
# Env (test seam):
#   CODEGEN_DRAIN_SCOPE_CMD   override for the `mix codegen.pitches.scope`
#                              invocation `assign --auto` shells out to
#                              (default: `cd "$CWD/test_harness" &&
#                              MIX_BUILD_PATH=_build/pitch_scope_parity mix
#                              codegen.pitches.scope --dir=ready --cwd="$CWD"
#                              --lanes=<N> --json`). Hermetic hook tests set
#                              this to a stub emitting canned JSON so no real
#                              mix/BEAM run is required to prove the
#                              JSON->lane->node wiring; the real mix leg is
#                              proven by the ExUnit test for the task itself.
#
# Exit codes: 0 = success, 1 = runtime error, 2 = usage/config error.

set -euo pipefail

CWD="$PWD"

# Health thresholds — genuinely-optional tuning knobs (defaults ARE the
# intended values), overridable per-box via env for a slow node.
# Seconds a role process (claude / beam.smp) may run before it is considered
# wedged rather than a legitimately long turn.
CODEGEN_DRAIN_ROLE_MAX_ETIME_SECS="${CODEGEN_DRAIN_ROLE_MAX_ETIME_SECS:-3600}"
# Seconds since the newest cycle log's mtime before that log is considered
# stale (no forward progress) rather than actively being written.
CODEGEN_DRAIN_LOG_STALE_SECS="${CODEGEN_DRAIN_LOG_STALE_SECS:-900}"

usage() {
    printf 'Usage:\n' >&2
    printf '  codegen-drain assign --slug=<slug> --node=<node> [--cwd=<dir>]\n' >&2
    printf '  codegen-drain assign --auto [--cwd=<dir>]\n' >&2
    printf '  codegen-drain status [--json] [--cwd=<dir>]\n' >&2
    printf '  codegen-drain init [--cwd=<dir>]\n' >&2
    exit 2
}

# strip_cwd_flag <argv...> — scans args for --cwd=<dir>/--cwd <dir> BEFORE
# subcommand-specific parsing runs, so every subcommand shares one resolution
# path. Sets global CWD and the global STRIPPED_ARGS array as side effects.
# MUST run in the current shell (no process substitution / subshell) — CWD
# is a plain assignment, and a subshell would mutate only its own copy,
# leaving the parent's CWD (and therefore INVENTORY) unresolved.
STRIPPED_ARGS=()
strip_cwd_flag() {
    STRIPPED_ARGS=()
    while [[ $# -gt 0 ]]; do
        case "$1" in
        --cwd=*)
            CWD="${1#--cwd=}"
            shift
            ;;
        --cwd)
            CWD="$2"
            shift 2
            ;;
        *)
            STRIPPED_ARGS+=("$1")
            shift
            ;;
        esac
    done
}

require_yq() {
    if ! command -v yq >/dev/null 2>&1; then
        printf 'codegen-drain: yq required but not found — install mikefarah/yq\n' >&2
        exit 2
    fi
    if ! yq --version 2>&1 | grep -qi mikefarah; then
        printf 'codegen-drain: installed yq is not mikefarah/yq (apt python-yq is incompatible)\n' >&2
        exit 2
    fi
}

require_inventory() {
    if [[ ! -f "$INVENTORY" ]]; then
        printf 'codegen-drain: no inventory at %s — run `codegen-drain init` first\n' "$INVENTORY" >&2
        exit 2
    fi
}

# node_field <node> <field> — empty string if absent (never fails, tolerates
# a node record with no optional field).
node_field() {
    local node="$1" field="$2"
    yq -r ".nodes[] | select(.name == \"$node\") | .${field} // \"\"" "$INVENTORY"
}

node_exists() {
    local node="$1"
    local hit
    hit="$(yq -r ".nodes[] | select(.name == \"$node\") | .name" "$INVENTORY" 2>/dev/null || true)"
    [[ -n "$hit" ]]
}

# inventory_nodes — prints every declared node name, one per line, in
# inventory order. Single owner for the "every node in drain-nodes.yaml"
# enumeration — snapshot, status, and reachable-node loops all call this
# instead of re-running `yq -r '.nodes[].name'` inline, so there is exactly
# one place this enumeration can drift or silently drop the final entry.
# Validates non-empty, unique names and exact line-count parity against the
# raw yq list; a malformed inventory (duplicate/blank name) is a loud abort,
# never a silently-shortened fleet view.
inventory_nodes() {
    local raw
    raw="$(yq -r '.nodes[].name' "$INVENTORY")"

    local -a names=()
    local n
    while IFS= read -r n; do
        names+=("$n")
    done <<<"$raw"

    local raw_count="${#names[@]}"
    local n2
    for n2 in "${names[@]}"; do
        [[ -n "$n2" ]] || {
            printf 'codegen-drain: inventory has a blank node name in %s\n' "$INVENTORY" >&2
            exit 1
        }
    done

    local uniq_count
    uniq_count="$(printf '%s\n' "${names[@]}" | sort -u | wc -l | tr -d '[:space:]')"
    if [[ "$uniq_count" != "$raw_count" ]]; then
        printf 'codegen-drain: inventory has duplicate node names in %s\n' "$INVENTORY" >&2
        exit 1
    fi

    printf '%s\n' "${names[@]}"
}

# watcher_probe_cmd <repo> — emits a self-contained shell snippet (string)
# that resolves whether <repo> is under queue supervision, by reading the
# single-flight lock BuildLock already writes at
# codegen/gate-pending/queue.lock — NEVER by scanning argv (a process-list
# grep over commandlines matches any process whose argv happens to quote
# the search pattern,
# including a solo `mix codegen.loop --cwd=<repo>` build, or the very agent
# asking the question). Prints "yes" iff the lock names label "queue" AND
# the recorded pid is alive (`kill -0`); "no" on any other state (absent,
# unreadable, malformed, dead pid, or label "solo" — a single build is not
# a watcher). One predicate, textually shared by both the local and remote
# legs of cmd_status, so there is exactly one place this logic can drift.
watcher_probe_cmd() {
    local repo="$1"
    printf 'lockf=%q; if [ -f "$lockf" ]; then read -r wpid wlabel _rest <"$lockf" 2>/dev/null; if [ "$wlabel" = "queue" ] && kill -0 "$wpid" 2>/dev/null; then echo yes; else echo no; fi; else echo no; fi' \
        "$repo/codegen/gate-pending/queue.lock"
}

# box_probe_cmd <repo> — emits a self-contained shell snippet (string) that
# reports git state (head/behind/ahead/dirty), the newest cycle log's mtime
# + slug, and every live role process (claude / beam.smp) with pid/ppid/
# elapsed. One predicate, textually shared by both the local (`bash -c`) and
# remote (`ssh`) legs of cmd_status, so the two can never drift — same
# discipline as watcher_probe_cmd/1.
#
# Output is a fixed 7-line-plus header block:
#   line 1: "PROBE_OK" or "PROBE_FAIL" — whether `ps` itself is usable on
#           this box. PROBE_FAIL is the caller's explicit signal to render
#           health=unknown rather than guessing idle from empty process data.
#   line 2: head (short sha, or empty if not a git repo)
#   line 3: behind (integer, or "null" if no upstream)
#   line 4: ahead  (integer, or "null" if no upstream)
#   line 5: dirty  (integer count of `git status --porcelain` lines)
#   line 6: "<mtime_epoch_or_0> <slug_or_empty>" for the newest cycle log
#   line 7+: one line per live role process: "<pid> <ppid> <etime_secs>"
#            (comm omitted — caller already knows it matched claude|beam.smp)
#
# Never pipes ps output into grep (that pattern is denied by test case (t),
# which closes the self-match class the watcher probe was fixed for) —
# filters with `awk` instead. Tries `ps -eo pid,ppid,etimes,comm` (raw seconds,
# Linux) first; falls back to `ps -eo pid,ppid,etime,comm` (HH:MM:SS /
# DD-HH:MM:SS, BSD/macOS) parsed by an awk time-to-seconds helper.
box_probe_cmd() {
    local repo="$1"
    printf '%s\n' \
        "cd $(printf '%q' "$repo") 2>/dev/null || true;" \
        'if ps -eo pid,ppid,etimes,comm >/dev/null 2>&1 || ps -eo pid,ppid,etime,comm >/dev/null 2>&1; then echo PROBE_OK; else echo PROBE_FAIL; fi;' \
        'if [ -d .git ] || git rev-parse --git-dir >/dev/null 2>&1; then' \
        '  git rev-parse --short HEAD 2>/dev/null || echo "";' \
        '  git rev-list --count @{u}..HEAD 2>/dev/null || echo null;' \
        '  git rev-list --count HEAD..@{u} 2>/dev/null || echo null;' \
        '  printf "%s\\n" "$(git status --porcelain 2>/dev/null | wc -l | tr -d "[:space:]")";' \
        'else' \
        '  echo ""; echo null; echo null; echo 0;' \
        'fi;' \
        "newest=\$(ls -t codegen/logging/*_cycle.jsonl 2>/dev/null | head -n1 || true);" \
        'if [ -n "$newest" ]; then' \
        '  mt=$(stat -c "%Y" "$newest" 2>/dev/null || stat -f "%m" "$newest" 2>/dev/null || echo 0);' \
        '  slug=$(basename "$newest" | sed -E "s/^[0-9]{8}_[0-9]{6}_//; s/_cycle\\.jsonl\$//");' \
        '  printf "%s %s\n" "$mt" "$slug";' \
        'else' \
        '  printf "0 \n";' \
        'fi;' \
        'raw=$(ps -eo pid,ppid,etimes,comm 2>/dev/null || true);' \
        'if [ -n "$raw" ]; then' \
        '  printf "%s\n" "$raw" | awk '"'"'NR>1 && ($4=="claude" || $4=="beam.smp") {print $1, $2, $3}'"'"';' \
        'else' \
        '  raw=$(ps -eo pid,ppid,etime,comm 2>/dev/null || true);' \
        '  printf "%s\n" "$raw" | awk '"'"'NR>1 && ($4=="claude" || $4=="beam.smp") {' \
        '    t=$3; d=0;' \
        '    if (split(t, dparts, "-") == 2) { d=dparts[1]; t=dparts[2]; }' \
        '    n=split(t, hparts, ":");' \
        '    if (n==3) secs=hparts[1]*3600+hparts[2]*60+hparts[3];' \
        '    else if (n==2) secs=hparts[1]*60+hparts[2];' \
        '    else secs=hparts[1];' \
        '    secs+=d*86400;' \
        '    print $1, $2, secs;' \
        '  }'"'"';' \
        'fi'
}

# classify_health <watcher> <role_lines_count> <max_ppid1_or_stale> — pure,
# no I/O. Precedence: unreachable (caller-handled before this is invoked) ->
# unknown (probe failure, caller-handled) -> wedged (any orphan or over-age
# or stale-log role process) -> working (watcher live + at least one healthy
# role process, or watcher live with none yet) -> idle (no watcher, no role
# process). Returns the verdict string on stdout.
classify_health() {
    local watcher="$1" has_wedged="$2" build_count="$3"

    if [[ "$has_wedged" == "1" ]]; then
        echo "wedged"
        return 0
    fi
    if [[ "$watcher" == "yes" ]] || [[ "$build_count" -gt 0 ]]; then
        echo "working"
        return 0
    fi
    echo "idle"
}

# remote_run <host> <run_as> <shell-command>
# Wraps in `su - <run_as> -c '...'` when run_as is set (root-landing ssh
# hosts); plain ssh exec otherwise (ssh user already owns the repo).
remote_run() {
    local host="$1" run_as="$2" cmd="$3"
    if [[ -n "$run_as" ]]; then
        ssh -n -o ConnectTimeout=10 "$host" "su - $run_as -c $(printf '%q' "$cmd")"
    else
        ssh -n -o ConnectTimeout=10 "$host" "$cmd"
    fi
}

# fleet_probe_cmd <repo> — emits a self-contained shell snippet (string)
# that enumerates the current lifecycle state of every ready/building/
# shipped pitch in <repo>, using ONLY find/base64/shell builtins — no yq
# dependency on the probed node (a remote node may not have mikefarah/yq
# installed; the INITIATING node is the only place frontmatter is ever
# parsed — see fleet_snapshot below). Output, one line per pitch:
#   STATE <state> <slug>
# and, for every READY pitch only, a second line carrying its full file
# bytes base64-encoded on one line (so a slug's frontmatter can be parsed
# back on the initiating node without a second round-trip):
#   PITCHB64 <slug> <base64>
# building/shipped pitches emit only the STATE line — their bytes are
# never needed (fleet edges are read from READY pitches' blocks_on: only;
# building/shipped existence is itself the edge-satisfaction signal).
fleet_probe_cmd() {
    local repo="$1"
    printf '%s\n' \
        "cd $(printf '%q' "$repo") 2>/dev/null || exit 1;" \
        'for f in codegen/pitches/ready/*.md; do' \
        '  [ -f "$f" ] || continue;' \
        '  slug=$(basename "$f" .md);' \
        '  printf "STATE ready %s\n" "$slug";' \
        '  b64=$(base64 <"$f" | tr -d "\n");' \
        '  printf "PITCHB64 %s %s\n" "$slug" "$b64";' \
        'done;' \
        'for f in codegen/pitches/building/*.md; do' \
        '  [ -f "$f" ] || continue;' \
        '  slug=$(basename "$f" .md);' \
        '  printf "STATE building %s\n" "$slug";' \
        'done;' \
        'for f in codegen/pitches/shipped/*.md; do' \
        '  [ -f "$f" ] || continue;' \
        '  slug=$(basename "$f" .md);' \
        '  printf "STATE shipped %s\n" "$slug";' \
        'done;' \
        'printf "PROBE_DONE\n";' \
        'exit 0'
}

# fleet_snapshot — probes EVERY node in the inventory (local-fs and
# remote/ssh alike), decodes and parses each node's ready-pitch
# frontmatter LOCALLY (this node's own mikefarah yq — remote nodes need
# no yq at all, they only ever emit base64 bytes), and writes 3 global
# associative-array-shaped outputs via the SNAPSHOT_* globals below:
#
#   SNAPSHOT_STATE["<node>:<slug>"]  = "ready" | "building" | "shipped"
#   SNAPSHOT_EDGES  (array)          = "<node> <ready-slug> <dep-slug>"
#     one entry per blocks_on: edge found in any node's ready/ batch
#
# FAIL-CLOSED: any unreachable node, ssh/base64 transport failure, malformed
# STATE line, or unparseable pitch frontmatter aborts (exit 1, named)
# BEFORE returning — an incomplete fleet view can never prove co-location,
# so assignment must never proceed on a partial snapshot. Bash 3.2 has no
# associative arrays, so SNAPSHOT_STATE is emulated as a delimited-line
# accumulator (SNAPSHOT_STATE_LINES) queried via snapshot_state_lookup.
SNAPSHOT_STATE_LINES=""
SNAPSHOT_EDGES=()

snapshot_state_lookup() {
    local key="$1"
    printf '%s\n' "$SNAPSHOT_STATE_LINES" | awk -F'\t' -v k="$key" '$1 == k { print $2; found=1 } END { if (!found) exit 1 }'
}

fleet_snapshot() {
    require_yq
    require_inventory

    SNAPSHOT_STATE_LINES=""
    SNAPSHOT_EDGES=()

    local -a nodes=()
    local n
    while IFS= read -r n; do
        [[ -n "$n" ]] && nodes+=("$n")
    done < <(inventory_nodes)

    for n in "${nodes[@]}"; do
        local host repo run_as probe_out
        host="$(node_field "$n" host)"
        repo="$(node_field "$n" repo)"
        run_as="$(node_field "$n" run_as)"

        [[ -n "$repo" ]] || {
            printf 'codegen-drain: node %q has no repo path in inventory — cannot snapshot\n' "$n" >&2
            exit 1
        }

        if [[ -z "$host" ]]; then
            probe_out="$(bash -c "$(fleet_probe_cmd "$repo")" 2>/dev/null || true)"
        else
            probe_out="$(remote_run "$host" "$run_as" "$(fleet_probe_cmd "$repo")" 2>/dev/null || true)"
        fi

        if [[ "$probe_out" != *"PROBE_DONE"* ]]; then
            printf 'codegen-drain: node %q unreachable or produced no fleet snapshot output — refusing (dependency safety requires the full fleet topology)\n' "$n" >&2
            exit 1
        fi

        local line kind rest slug b64
        while IFS= read -r line; do
            [[ -n "$line" ]] || continue
            case "$line" in
            "PROBE_DONE")
                continue
                ;;
            "STATE "*)
                rest="${line#STATE }"
                kind="${rest%% *}"
                slug="${rest#* }"
                case "$kind" in
                ready | building | shipped) ;;
                *)
                    printf 'codegen-drain: node %q emitted malformed fleet state line: %q\n' "$n" "$line" >&2
                    exit 1
                    ;;
                esac
                [[ -n "$slug" ]] || {
                    printf 'codegen-drain: node %q emitted a STATE line with an empty slug: %q\n' "$n" "$line" >&2
                    exit 1
                }
                SNAPSHOT_STATE_LINES="${SNAPSHOT_STATE_LINES}${n}:${slug}	${kind}
"
                ;;
            "PITCHB64 "*)
                rest="${line#PITCHB64 }"
                slug="${rest%% *}"
                b64="${rest#* }"
                [[ -n "$slug" && -n "$b64" ]] || {
                    printf 'codegen-drain: node %q emitted a malformed PITCHB64 line for slug %q\n' "$n" "$slug" >&2
                    exit 1
                }

                local deps dep decoded
                if ! decoded="$(printf '%s' "$b64" | yq -r '@base64d' 2>&1)"; then
                    printf 'codegen-drain: node %q emitted undecodable base64 pitch bytes for slug %q: %s\n' "$n" "$slug" "$decoded" >&2
                    exit 1
                fi
                if ! deps="$(printf '%s' "$decoded" | yq --front-matter=extract -r '.blocks_on[]' 2>&1)"; then
                    printf 'codegen-drain: node %q emitted unparseable pitch frontmatter for slug %q: %s\n' "$n" "$slug" "$deps" >&2
                    exit 1
                fi
                while IFS= read -r dep; do
                    [[ -n "$dep" ]] || continue
                    SNAPSHOT_EDGES+=("$n $slug $dep")
                done <<<"$deps"
                ;;
            *)
                printf 'codegen-drain: node %q emitted unrecognized fleet snapshot output: %q\n' "$n" "$line" >&2
                exit 1
                ;;
            esac
        done <<<"$probe_out"
    done
}

cmd_assign() {
    local slug="" node="" auto=""
    while [[ $# -gt 0 ]]; do
        case "$1" in
        --slug=*)
            slug="${1#--slug=}"
            shift
            ;;
        --slug)
            slug="$2"
            shift 2
            ;;
        --node=*)
            node="${1#--node=}"
            shift
            ;;
        --node)
            node="$2"
            shift 2
            ;;
        --auto)
            auto="1"
            shift
            ;;
        *)
            printf 'codegen-drain assign: unknown option %q\n' "$1" >&2
            usage
            ;;
        esac
    done

    if [[ -n "$auto" ]]; then
        if [[ -n "$slug" || -n "$node" ]]; then
            printf 'codegen-drain assign: --auto is mutually exclusive with --slug/--node\n' >&2
            usage
        fi
        cmd_assign_auto
        return $?
    fi

    [[ -n "$slug" ]] || {
        printf 'codegen-drain assign: --slug required\n' >&2
        usage
    }
    [[ -n "$node" ]] || {
        printf 'codegen-drain assign: --node required\n' >&2
        usage
    }

    require_yq
    require_inventory

    cmd_assign_one "$slug" "$node"
}

# dependency_preflight <slug> <target-node> — bidirectional fleet-dependency
# check run before any REAL (non-no-op) cross-node transfer of <slug> to
# <target-node>. Builds a full fleet_snapshot (fail-closed: aborts the
# entire assignment before this function returns if any declared node is
# unreachable), then checks:
#
#   OUTGOING — every blocks_on: dependency <slug> itself declares must
#     already exist on <target-node> (ready, building, or shipped state
#     all count as "on that node" per the local queue's own
#     ready-union-shipped satisfaction contract; building only means the
#     dependent will wait locally there, which is fine).
#   INCOMING — every fleet-ready pitch (on ANY node) that names <slug> as
#     ITS blocks_on: dependency must ALREADY reside on <target-node> too
#     — moving <slug> away from a node holding one of its dependents
#     would strand that dependent.
#
# Refuses (exit 1) naming the blocking edge, its current owner, and a
# concrete remediation command on ANY violation. Never mutates anything —
# read-only check.
dependency_preflight() {
    local slug="$1" target="$2"

    fleet_snapshot

    # OUTGOING: slug's own blocks_on: deps (read from the LOCAL copy still
    # sitting in $CWD/codegen/pitches/ready/ — this box owns it right now).
    local pitch_path="$CWD/codegen/pitches/ready/${slug}.md"
    local content deps dep
    content="$(cat "$pitch_path")"
    if ! deps="$(printf '%s' "$content" | yq --front-matter=extract -r '.blocks_on[]' 2>&1)"; then
        printf 'codegen-drain assign: unparseable pitch frontmatter for %s: %s\n' "$slug" "$deps" >&2
        exit 1
    fi

    while IFS= read -r dep; do
        [[ -n "$dep" ]] || continue
        local dep_state
        if ! dep_state="$(snapshot_state_lookup "${target}:${dep}")"; then
            printf 'codegen-drain assign: %s requires %s\n' "$slug" "$dep" >&2
            printf 'codegen-drain assign: target %s lacks that prerequisite\n' "$target" >&2
            printf 'remediation: codegen-drain assign --slug=%s --node=<node holding %s>\n' "$dep" "$dep" >&2
            exit 1
        fi
    done <<<"$deps"

    # INCOMING: any fleet-ready pitch (any node) naming slug as ITS dep must
    # already be on target too, else moving slug strands that dependent.
    local edge
    for edge in "${SNAPSHOT_EDGES[@]+"${SNAPSHOT_EDGES[@]}"}"; do
        local edge_node edge_slug edge_dep
        edge_node="$(printf '%s' "$edge" | awk '{print $1}')"
        edge_slug="$(printf '%s' "$edge" | awk '{print $2}')"
        edge_dep="$(printf '%s' "$edge" | awk '{print $3}')"

        if [[ "$edge_dep" == "$slug" ]]; then
            if ! snapshot_state_lookup "${target}:${edge_slug}" >/dev/null; then
                printf 'codegen-drain assign: %s requires %s\n' "$edge_slug" "$slug" >&2
                printf 'codegen-drain assign: moving %s to %s would strand %s (currently on %s)\n' \
                    "$slug" "$target" "$edge_slug" "$edge_node" >&2
                printf 'remediation: codegen-drain assign --slug=%s --node=%s\n' "$edge_slug" "$target" >&2
                exit 1
            fi
        fi
    done
}

# refuse_if_recovery_bound <slug> — a ready pitch with a LOCAL recovery
# dossier (codegen/gate-pending/recoveries/<slug>/*.json) owns machine-local
# state that pitch-file transfer does not carry: the dossier JSON and its
# immutable Git ref (see pitch "restarted builds resume owned work"). Moving
# only the pitch markdown to another node would strand that state — the
# recovered bytes become unreachable from the new node, and a later
# same-slug selection there finds nothing to materialize. Fails closed on an
# unreadable directory (never silently treats "can't tell" as "safe to
# move"). The same-node no-op path (checked by callers BEFORE this) is
# unaffected — this only refuses a genuine CROSS-node transfer.
refuse_if_recovery_bound() {
    local slug="$1"
    local recoveries_dir="$CWD/codegen/gate-pending/recoveries/${slug}"

    [[ -d "$recoveries_dir" ]] || return 0

    local hit
    hit="$(find "$recoveries_dir" -maxdepth 1 -name '*.json' -type f -print -quit 2>/dev/null)" || {
        printf 'codegen-drain: recovery-bound %s: could not enumerate %s — refusing (fail-closed)\n' \
            "$slug" "$recoveries_dir" >&2
        exit 1
    }

    if [[ -n "$hit" ]]; then
        printf 'codegen-drain: recovery-bound %s: dossier/ref transfer unsupported; keep on current node\n' "$slug" >&2
        exit 1
    fi
}

# cmd_assign_one <slug> <node> — the single-slug transfer (checksum-verified
# ack-before-delete on remote nodes; same-fs mv on local-fs nodes). Callers
# (cmd_assign's --slug/--node leg, and cmd_assign_auto's per-lane loop) must
# have already run require_yq/require_inventory. Placing a pitch onto the
# node it already occupies is a no-op (exit 0), never a possession refusal.
cmd_assign_one() {
    local slug="$1" node="$2"

    local local_path="$CWD/codegen/pitches/ready/${slug}.md"
    if [[ ! -f "$local_path" ]]; then
        printf 'codegen-drain: %s not in ready/ — run `codegen-drain status` to find which node holds it\n' "$slug" >&2
        exit 1
    fi

    if ! node_exists "$node"; then
        printf 'codegen-drain: unknown node %q — valid nodes:\n' "$node" >&2
        inventory_nodes >&2
        exit 2
    fi

    local host repo run_as
    host="$(node_field "$node" host)"
    repo="$(node_field "$node" repo)"
    run_as="$(node_field "$node" run_as)"

    [[ -n "$repo" ]] || {
        printf 'codegen-drain: node %q has no repo path in inventory\n' "$node" >&2
        exit 2
    }

    # local node (no host) — same-filesystem mv, no ssh needed
    if [[ -z "$host" ]]; then
        local dest_dir="$repo/codegen/pitches/ready"
        local dest="$dest_dir/${slug}.md"
        # Already at its assigned node: source and destination are the same
        # on-disk file (-ef compares inode+device, so a symlinked --cwd still
        # matches). Placing a pitch where it already lives is a no-op, never a
        # possession conflict — the resurrection guard below must see only
        # genuinely distinct files.
        if [[ "$local_path" -ef "$dest" ]]; then
            printf 'already at %s (no-op)\n' "$node"
            return 0
        fi
        refuse_if_recovery_bound "$slug"
        dependency_preflight "$slug" "$node"
        if [[ -f "$dest" ]]; then
            printf 'codegen-drain: %s already exists at node %q — refusing (possession ambiguous)\n' "$dest" "$node" >&2
            exit 1
        fi
        mkdir -p "$dest_dir"
        mv "$local_path" "$dest"
        printf 'assigned %s -> %s (local copy moved)\n' "$slug" "$node"
        return 0
    fi

    refuse_if_recovery_bound "$slug"
    dependency_preflight "$slug" "$node"

    # Refuse if destination already holds it (resurrection guard).
    local existing
    existing="$(remote_run "$host" "$run_as" "test -f '$repo/codegen/pitches/ready/${slug}.md' && echo yes || echo no" 2>/dev/null || echo no)"
    if [[ "$existing" == "yes" ]]; then
        printf 'codegen-drain: %s already exists in %s:ready/ — refusing (possession ambiguous)\n' "$slug" "$node" >&2
        exit 1
    fi

    # Stage via /tmp (root-safe hop), install with correct ownership into
    # .incoming/, verify checksum, then atomic same-fs mv into ready/.
    local tmp_remote="/tmp/${slug}.md"
    if ! scp -o ConnectTimeout=10 "$local_path" "${host}:${tmp_remote}" >/dev/null; then
        printf 'codegen-drain: scp to %s failed — local copy kept\n' "$node" >&2
        exit 1
    fi

    local incoming_dir="$repo/codegen/pitches/.incoming"
    local incoming_path="$incoming_dir/${slug}.md"

    if [[ -n "$run_as" ]]; then
        if ! ssh -o ConnectTimeout=10 "$host" "install -d -o $run_as -g $run_as -m 755 '$incoming_dir' && install -o $run_as -g $run_as -m 644 '$tmp_remote' '$incoming_path' && rm -f '$tmp_remote'"; then
            printf 'codegen-drain: install(1) on %s failed — local copy kept; stale bytes may remain at %s:%s\n' "$node" "$node" "$tmp_remote" >&2
            exit 1
        fi
    else
        if ! ssh -o ConnectTimeout=10 "$host" "mkdir -p '$incoming_dir' && mv '$tmp_remote' '$incoming_path'"; then
            printf 'codegen-drain: staging on %s failed — local copy kept\n' "$node" >&2
            exit 1
        fi
    fi

    local local_sum remote_sum
    local_sum="$(shasum "$local_path" | awk '{print $1}')"
    remote_sum="$(remote_run "$host" "$run_as" "shasum '$incoming_path' 2>/dev/null | awk '{print \$1}'" 2>/dev/null || true)"

    if [[ -z "$remote_sum" || "$local_sum" != "$remote_sum" ]]; then
        printf 'codegen-drain: transfer verify FAILED — %s kept locally; stale copy left at %s:%s for inspection\n' "$slug" "$node" "$incoming_path" >&2
        exit 1
    fi

    if ! remote_run "$host" "$run_as" "mv '$incoming_path' '$repo/codegen/pitches/ready/${slug}.md'"; then
        printf 'codegen-drain: final mv on %s failed — local copy kept; verified bytes remain at %s:%s\n' "$node" "$node" "$incoming_path" >&2
        exit 1
    fi

    # Re-verify it landed before deleting the local copy.
    local landed
    landed="$(remote_run "$host" "$run_as" "test -f '$repo/codegen/pitches/ready/${slug}.md' && echo yes || echo no" 2>/dev/null || echo no)"
    if [[ "$landed" != "yes" ]]; then
        printf 'codegen-drain: could not confirm landing on %s — local copy kept for safety\n' "$node" >&2
        exit 1
    fi

    rm -f "$local_path"
    local count
    count="$(remote_run "$host" "$run_as" "find '$repo/codegen/pitches/ready' -maxdepth 1 -name '*.md' | wc -l" 2>/dev/null | tr -d '[:space:]' || echo '?')"
    printf 'assigned %s -> %s (local copy removed; %s holds %s pitches)\n' "$slug" "$node" "$node" "$count"
}

# reachable_nodes — prints one reachable node name per line, in inventory
# order. A local-fs node (no host:) is always reachable. A remote node is
# reachable iff the same find-ready probe cmd_status uses returns non-empty
# output. Unreachable nodes are silently excluded from the printed list
# (the caller is responsible for naming what was excluded, if it cares).
reachable_nodes() {
    local names node host repo run_as probe_out
    names="$(inventory_nodes)"
    while IFS= read -r node; do
        [[ -n "$node" ]] || continue
        host="$(node_field "$node" host)"
        if [[ -z "$host" ]]; then
            printf '%s\n' "$node"
            continue
        fi
        repo="$(node_field "$node" repo)"
        run_as="$(node_field "$node" run_as)"
        probe_out="$(remote_run "$host" "$run_as" "find '$repo/codegen/pitches/ready' -maxdepth 1 -name '*.md' 2>/dev/null | wc -l" 2>/dev/null || true)"
        [[ -n "$probe_out" ]] && printf '%s\n' "$node"
    done <<<"$names"
}

# cmd_assign_auto — partitions THIS box's ready/ into N lanes (N = reachable
# node count) via `mix codegen.pitches.scope --json`, then places lane i onto
# reachable-node i (inventory order) via cmd_assign_one. GLOBAL-HOT and
# UNROUTED slugs are printed by name and never placed. A failing transfer
# stops immediately (fail-closed) — already-placed lanes stay placed,
# remaining lanes are not attempted.
cmd_assign_auto() {
    require_yq
    require_inventory

    local -a nodes=()
    local n
    while IFS= read -r n; do
        [[ -n "$n" ]] && nodes+=("$n")
    done < <(reachable_nodes)

    local lane_count="${#nodes[@]}"
    if [[ "$lane_count" -eq 0 ]]; then
        printf 'codegen-drain assign --auto: zero reachable nodes — refusing\n' >&2
        exit 1
    fi

    # Fleet-wide dependency snapshot: fail-closed on any unreachable node.
    # externally_referenced = every dep-slug named by a blocks_on: edge on
    # ANY fleet node (this box's own edges are naturally included too, but
    # the fleet_partition/3 outgoing-edge check already covers those; this
    # set specifically catches "some OTHER node's ready pitch needs X
    # local", the case a purely-local partition can never see).
    fleet_snapshot
    local -a ext_ref_slugs=()
    local edge
    for edge in "${SNAPSHOT_EDGES[@]+"${SNAPSHOT_EDGES[@]}"}"; do
        local edge_dep
        edge_dep="$(printf '%s' "$edge" | awk '{print $3}')"
        ext_ref_slugs+=("$edge_dep")
    done
    local externally_referenced=""
    if [[ "${#ext_ref_slugs[@]}" -gt 0 ]]; then
        externally_referenced="$(printf '%s\n' "${ext_ref_slugs[@]}" | sort -u | tr '\n' ',' | sed 's/,$//')"
    fi

    local scope_cmd="${CODEGEN_DRAIN_SCOPE_CMD:-}"
    local json_out
    if [[ -n "$scope_cmd" ]]; then
        json_out="$(CODEGEN_DRAIN_LANES="$lane_count" CODEGEN_DRAIN_CWD="$CWD" bash -c "$scope_cmd")" || {
            printf 'codegen-drain assign --auto: scope command failed\n' >&2
            exit 1
        }
    else
        local -a scope_extra_args=(--fleet-safe)
        if [[ -n "$externally_referenced" ]]; then
            scope_extra_args+=(--externally-referenced="$externally_referenced")
        fi
        json_out="$(cd "$CWD/test_harness" && MIX_BUILD_PATH=_build/pitch_scope_parity mix codegen.pitches.scope --dir=ready --cwd="$CWD" --lanes="$lane_count" --json "${scope_extra_args[@]}")" || {
            printf 'codegen-drain assign --auto: mix codegen.pitches.scope failed\n' >&2
            exit 1
        }
    fi

    [[ -n "$json_out" ]] || {
        printf 'codegen-drain assign --auto: scope command produced no output\n' >&2
        exit 1
    }

    local global_hot unrouted dependency_bound
    global_hot="$(printf '%s' "$json_out" | yq -p=json -r '.global_hot[]' 2>/dev/null || true)"
    unrouted="$(printf '%s' "$json_out" | yq -p=json -r '.unrouted[]' 2>/dev/null || true)"
    dependency_bound="$(printf '%s' "$json_out" | yq -p=json -r '.dependency_bound[]' 2>/dev/null || true)"

    if [[ -n "$global_hot" ]]; then
        printf 'GLOBAL-HOT (never placed — build alone): %s\n' "$(printf '%s' "$global_hot" | tr '\n' ' ')"
    fi
    if [[ -n "$unrouted" ]]; then
        printf 'UNROUTED (never placed — a human must place these): %s\n' "$(printf '%s' "$unrouted" | tr '\n' ' ')"
    fi
    if [[ -n "$dependency_bound" ]]; then
        printf 'DEPENDENCY-BOUND (kept on source node — build chain locally): %s\n' "$(printf '%s' "$dependency_bound" | tr '\n' ' ')"
    fi

    local idx=0
    local placed=0
    for n in "${nodes[@]}"; do
        local lane_slugs
        lane_slugs="$(printf '%s' "$json_out" | yq -p=json -r ".lanes[$idx][]" 2>/dev/null || true)"
        idx=$((idx + 1))

        [[ -n "$lane_slugs" ]] || continue

        local lane_slug
        while IFS= read -r lane_slug; do
            [[ -n "$lane_slug" ]] || continue
            cmd_assign_one "$lane_slug" "$n"
            placed=$((placed + 1))
        done <<<"$lane_slugs"
    done

    printf 'assign --auto: placed %s pitch(es) across %s reachable node(s)\n' "$placed" "$lane_count"
}

# json_escape <string> — minimal escaping for embedding a plain string value
# in a JSON string literal (backslash + double-quote only; slugs/reasons here
# never carry control chars in practice, but escape defensively).
json_escape() {
    local s="$1"
    s="${s//\\/\\\\}"
    s="${s//\"/\\\"}"
    printf '%s' "$s"
}

cmd_status() {
    local json=""
    while [[ $# -gt 0 ]]; do
        case "$1" in
        --json)
            json="1"
            shift
            ;;
        *)
            printf 'codegen-drain status: unknown option %q\n' "$1" >&2
            usage
            ;;
        esac
    done

    require_yq
    require_inventory

    local names
    names="$(inventory_nodes)"

    [[ -n "$json" ]] && printf '['
    local first=1
    while IFS= read -r node; do
        [[ -n "$node" ]] || continue
        local host repo run_as
        host="$(node_field "$node" host)"
        repo="$(node_field "$node" repo)"
        run_as="$(node_field "$node" run_as)"

        local count="0" reachable="1" watcher="no" incoming="0" building="0"
        local box_head="" box_behind="null" box_ahead="null" box_dirty="0"
        local box_log_mtime="0" box_log_slug="" box_reason=""
        local -a box_builds=() # each element: "pid ppid etime_secs"
        local box_probe_ok="1" box_out=""

        if [[ -z "$host" ]]; then
            if [[ -d "$repo/codegen/pitches/ready" ]]; then
                count="$(find "$repo/codegen/pitches/ready" -maxdepth 1 -name '*.md' 2>/dev/null | wc -l | tr -d '[:space:]')"
            fi
            if [[ -d "$repo/codegen/pitches/.incoming" ]]; then
                incoming="$(find "$repo/codegen/pitches/.incoming" -maxdepth 1 -name '*.md' 2>/dev/null | wc -l | tr -d '[:space:]')"
            fi
            if [[ -d "$repo/codegen/pitches/building" ]]; then
                building="$(find "$repo/codegen/pitches/building" -maxdepth 1 -name '*.md' 2>/dev/null | wc -l | tr -d '[:space:]')"
            fi
            watcher="$(bash -c "$(watcher_probe_cmd "$repo")" 2>/dev/null || echo no)"

            box_out="$(bash -c "$(box_probe_cmd "$repo")" 2>/dev/null || true)"
            if [[ -z "$box_out" ]]; then
                box_probe_ok="0"
                box_reason="box probe produced no output"
            fi
        else
            local remote_out
            remote_out="$(remote_run "$host" "$run_as" "find '$repo/codegen/pitches/ready' -maxdepth 1 -name '*.md' 2>/dev/null | wc -l; find '$repo/codegen/pitches/.incoming' -maxdepth 1 -name '*.md' 2>/dev/null | wc -l; $(watcher_probe_cmd "$repo"); find '$repo/codegen/pitches/building' -maxdepth 1 -name '*.md' 2>/dev/null | wc -l" 2>/dev/null || true)"
            if [[ -z "$remote_out" ]]; then
                reachable="0"
            else
                count="$(printf '%s\n' "$remote_out" | sed -n '1p' | tr -d '[:space:]')"
                incoming="$(printf '%s\n' "$remote_out" | sed -n '2p' | tr -d '[:space:]')"
                watcher="$(printf '%s\n' "$remote_out" | sed -n '3p' | tr -d '[:space:]')"
                building="$(printf '%s\n' "$remote_out" | sed -n '4p' | tr -d '[:space:]')"
                [[ -z "$count" ]] && count="0"
                [[ -z "$incoming" ]] && incoming="0"
                [[ -z "$watcher" ]] && watcher="no"
                [[ -z "$building" ]] && building="0"

                box_out="$(remote_run "$host" "$run_as" "$(box_probe_cmd "$repo")" 2>/dev/null || true)"
                if [[ -z "$box_out" ]]; then
                    box_probe_ok="0"
                    box_reason="box probe produced no output"
                fi
            fi
        fi

        if [[ "$reachable" == "1" && -n "$box_out" ]]; then
            local probe_status
            probe_status="$(printf '%s\n' "$box_out" | sed -n '1p' | tr -d '[:space:]')"
            if [[ "$probe_status" != "PROBE_OK" ]]; then
                box_probe_ok="0"
                box_reason="ps unusable on this box — cannot enumerate role processes"
            fi

            box_head="$(printf '%s\n' "$box_out" | sed -n '2p')"
            box_behind="$(printf '%s\n' "$box_out" | sed -n '3p' | tr -d '[:space:]')"
            box_ahead="$(printf '%s\n' "$box_out" | sed -n '4p' | tr -d '[:space:]')"
            box_dirty="$(printf '%s\n' "$box_out" | sed -n '5p' | tr -d '[:space:]')"
            local log_line
            log_line="$(printf '%s\n' "$box_out" | sed -n '6p')"
            box_log_mtime="${log_line%% *}"
            box_log_slug="${log_line#* }"
            [[ "$box_log_slug" == "$log_line" ]] && box_log_slug=""
            [[ -z "$box_behind" ]] && box_behind="null"
            [[ -z "$box_ahead" ]] && box_ahead="null"
            [[ -z "$box_dirty" ]] && box_dirty="0"
            [[ -z "$box_log_mtime" ]] && box_log_mtime="0"

            local proc_line
            while IFS= read -r proc_line; do
                [[ -n "$proc_line" ]] && box_builds+=("$proc_line")
            done < <(printf '%s\n' "$box_out" | sed -n '7,$p')
        fi

        # ── Compute health from the collected fields ────────────────────
        local health="unknown"
        if [[ "$reachable" == "0" ]]; then
            health="unreachable"
        elif [[ "$box_probe_ok" == "0" ]]; then
            health="unknown"
        else
            local now has_wedged="0"
            now="$(date +%s)"
            local log_stale="0"
            if [[ "$box_log_mtime" != "0" ]]; then
                local age=$((now - box_log_mtime))
                [[ "$age" -gt "$CODEGEN_DRAIN_LOG_STALE_SECS" ]] && log_stale="1"
            fi
            local b pid ppid etime_secs
            for b in "${box_builds[@]+"${box_builds[@]}"}"; do
                read -r pid ppid etime_secs <<<"$b"
                if [[ "$ppid" == "1" ]]; then
                    has_wedged="1"
                elif [[ "$etime_secs" -gt "$CODEGEN_DRAIN_ROLE_MAX_ETIME_SECS" ]]; then
                    has_wedged="1"
                elif [[ "$log_stale" == "1" ]]; then
                    has_wedged="1"
                fi
            done
            health="$(classify_health "$watcher" "$has_wedged" "${#box_builds[@]}")"
        fi

        if [[ -n "$json" ]]; then
            [[ $first -eq 1 ]] || printf ','
            first=0
            local builds_json="" b pid ppid etime_secs sep=""
            for b in "${box_builds[@]+"${box_builds[@]}"}"; do
                read -r pid ppid etime_secs <<<"$b"
                local orphaned="false"
                [[ "$ppid" == "1" ]] && orphaned="true"
                builds_json+="${sep}{\"pid\":${pid},\"ppid\":${ppid},\"etime_secs\":${etime_secs},\"orphaned\":${orphaned},\"slug\":\"$(json_escape "$box_log_slug")\",\"slug_confidence\":\"$([[ "${#box_builds[@]}" -eq 1 ]] && echo high || echo low)\"}"
                sep=","
            done
            local head_json="null"
            [[ -n "$box_head" ]] && head_json="\"$(json_escape "$box_head")\""
            local reason_json="\"$(json_escape "$box_reason")\""
            printf '{"node":"%s","reachable":%s,"ready_count":%s,"incoming_count":%s,"watcher":"%s","building_count":%s,"health":"%s","reason":%s,"head":%s,"behind":%s,"ahead":%s,"dirty":%s,"builds":[%s]}' \
                "$node" "$([[ "$reachable" == "1" ]] && echo true || echo false)" "$count" "$incoming" "$watcher" "$building" \
                "$health" "$reason_json" "$head_json" "$box_behind" "$box_ahead" "$box_dirty" "$builds_json"
        else
            if [[ "$reachable" == "0" ]]; then
                printf '%-12s unreachable\n' "$node"
            else
                printf '%-12s ready=%-4s incoming=%-4s watcher=%-4s building=%-4s health=%s\n' "$node" "$count" "$incoming" "$watcher" "$building" "$health"
                if [[ "$health" == "unknown" && -n "$box_reason" ]]; then
                    printf '             health unknown — %s\n' "$box_reason"
                fi
                if [[ "${#box_builds[@]}" -gt 0 ]]; then
                    local b pid ppid etime_secs
                    for b in "${box_builds[@]}"; do
                        read -r pid ppid etime_secs <<<"$b"
                        local marker=""
                        [[ "$ppid" == "1" ]] && marker=" ORPHANED"
                        printf '             build pid=%s ppid=%s elapsed=%ss slug=%s%s\n' "$pid" "$ppid" "$etime_secs" "${box_log_slug:-unknown}" "$marker"
                    done
                fi
                if [[ "$watcher" == "no" ]]; then
                    printf '             not supervised — start with: claude-build --queue --watch\n'
                fi
            fi
        fi
    done <<<"$names"
    [[ -n "$json" ]] && printf ']\n'
    return 0
}

cmd_init() {
    mkdir -p "$(dirname "$INVENTORY")"
    if [[ -f "$INVENTORY" ]]; then
        printf 'codegen-drain init: %s already exists — edit it directly to add/remove nodes\n' "$INVENTORY" >&2
        exit 1
    fi
    cat >"$INVENTORY" <<'YAML'
# codegen/drain-nodes.yaml — fleet inventory for codegen-drain.
# Gitignored: machine-local topology, never committed.
nodes:
  - name: local
    repo: /path/to/local/codegen
YAML
    printf 'wrote %s — edit it to add remote nodes (host, repo, run_as, launch)\n' "$INVENTORY"
}

# ── Subcommand dispatch ──────────────────────────────────────────────────────
if [[ $# -eq 0 ]]; then
    printf 'codegen-drain: subcommand required (assign|status|init)\n' >&2
    usage
fi

SUBCOMMAND="$1"
shift

# Strip --cwd=<dir> / --cwd <dir> from the remaining args wherever it
# appears (mirrors codegen-build's --cwd contract); resolves CWD before any
# subcommand-specific parsing runs, then INVENTORY is derived from CWD, never
# from this script's own install location.
strip_cwd_flag "$@"
set -- "${STRIPPED_ARGS[@]+"${STRIPPED_ARGS[@]}"}"

INVENTORY="${CODEGEN_DRAIN_INVENTORY:-$CWD/codegen/drain-nodes.yaml}"

case "$SUBCOMMAND" in
assign)
    cmd_assign "$@"
    ;;
status)
    cmd_status "$@"
    ;;
init)
    cmd_init "$@"
    ;;
--*)
    printf 'codegen-drain: subcommand required (assign|status|init)\n' >&2
    usage
    ;;
*)
    printf 'codegen-drain: unknown subcommand %q (expected assign|status|init)\n' "$SUBCOMMAND" >&2
    usage
    ;;
esac