#!/usr/bin/env node
// contributor-relay.sh — ClankeR, the contributor relay: the WebSocket client
// that connects a contributor agent to the Hive hub.
//
// Handles: authentication, task receipt, GitHub token injection, result reporting,
// heartbeat, and reconnection with exponential backoff.
//
// Environment:
//   HIVE_HUB              — WebSocket URL (wss://host:port/contribute);
//                           comma-separated URLs subscribe to multiple hubs
//   HIVE_REGISTRATION_TOKEN — contributor's registration token; for multiple
//                           hubs, provide one comma-separated token per hub in
//                           the same order as HIVE_HUB
//   AGENT_BACKEND          — CLI backend name (claude, copilot, gemini, etc.)
//   AGENT_MODEL            — model override (optional)
//   HIVE_AGENT_ROLE        — optional spoke agent role to claim (scanner,
//                           quality, outreach, etc.; hub-enforced)
//   HIVE_AGENT_SESSION     — tmux session name for the agent (default: contributor)

'use strict';

const WebSocket = require('ws');
const { execSync, execFile } = require('child_process');
const fs = require('fs');
const path = require('path');

const rawHub = process.env.HIVE_HUB || 'wss://hive.kubestellar.io:3001/contribute';
// Multi-hub (kubestellar/hive#multi-hive): HIVE_HUB and HIVE_REGISTRATION_TOKEN
// may each be a comma-separated list, one token per hub in the same order, so
// one relay/CLI session can hold work from more than one hive without running
// duplicate contributor processes. A single value of each (the common case)
// behaves exactly as before — this only ever adds hubs, never changes
// single-hub behaviour.
const rawHubList = rawHub.split(',').map(s => s.trim()).filter(Boolean);
const rawTokenList = (process.env.HIVE_REGISTRATION_TOKEN || '').split(',').map(s => s.trim()).filter(Boolean);
if (rawHubList.length > 1 && rawTokenList.length !== rawHubList.length) {
  console.error(`FATAL: HIVE_HUB lists ${rawHubList.length} hub(s) but HIVE_REGISTRATION_TOKEN lists ${rawTokenList.length} token(s) — need one registration token per hub, in the same order.`);
  process.exit(1);
}
const BACKEND = process.env.AGENT_BACKEND || 'claude';
const MODEL = process.env.AGENT_MODEL || process.env.GOOSE_MODEL || '';
const AGENT_ROLE = (process.env.HIVE_AGENT_ROLE || '').trim();
const TMUX_SESSION = process.env.HIVE_AGENT_SESSION || 'contributor';
const GH_TOKEN_CACHE = process.env.HIVE_GH_TOKEN_CACHE || (fs.existsSync('/var/run/hive-metrics')
  ? '/var/run/hive-metrics/gh-app-token.cache'
  : '/tmp/hive-gh-token.cache');
const TASK_FILE = process.env.HIVE_TASK_FILE || '/tmp/contributor-task.json';

// --- Delivery mode (kubestellar/hive#2538) -------------------------------
// The relay can deliver a task to the backend CLI in one of two ways:
//
//   interactive (default) — the legacy path: type the prompt into a live
//     tmux pane with `tmux send-keys` and scrape the pane for readiness,
//     progress and completion. Requires an attached-or-attachable TTY and is
//     unchanged by this feature.
//
//   headless — the non-interactive path added for #2538: drive the backend
//     CLI in a one-shot / print invocation (`claude -p`, `copilot -p`,
//     `codex exec`, …), capture its stdout/stderr, and report completion or a
//     REAL error back over the same WebSocket channel. No tmux, no pane
//     scraping, no waiting on an invisible prompt — so a K8s Job/Deployment
//     running this mode either runs to completion or fails loudly (the exact
//     "healthy-looking but stalled pod" failure #2538 warns about), and it
//     never needs a human to attach and type `/login`.
//
// This is opt-in and additive: absent/any-other value keeps the interactive
// path exactly as before. K8s manifests (#2549) and the credential boundary
// (#2537) are the explicit follow-ons and are NOT built here.
const MODE_INTERACTIVE = 'interactive';
const MODE_HEADLESS = 'headless';
const CONTRIBUTOR_MODE = process.env.CONTRIBUTOR_MODE === MODE_HEADLESS
  ? MODE_HEADLESS
  : MODE_INTERACTIVE;

// Where the headless runner records its current lifecycle state as JSON, so a
// supervising process (or a future K8s liveness/readiness probe reading the
// file) can distinguish waiting / working / done / failed — instead of a pod
// that merely looks alive. Best-effort: a write failure never aborts a task.
const HEADLESS_STATUS_FILE = process.env.HIVE_HEADLESS_STATUS_FILE || '/tmp/contributor-headless-status.json';

// Coarse lifecycle states reported by the headless runner. Named so probes and
// logs agree on the vocabulary rather than matching free text.
const HEADLESS_STATE_WAITING = 'waiting'; // authenticated, no task in flight
const HEADLESS_STATE_WORKING = 'working'; // one-shot CLI invocation running
const HEADLESS_STATE_DONE = 'done';       // last task completed (exit 0)
const HEADLESS_STATE_FAILED = 'failed';   // last task failed (non-zero/spawn error)

// Interactive pane classifier states. Keep this vocabulary small and explicit:
// "not complete" splits into active work vs. human input needed so the relay
// never reports success for a turn that is actually sitting at a question.
const PANE_STATE_WORKING = 'WORKING';
const PANE_STATE_BLOCKED_ON_HUMAN = 'BLOCKED_ON_HUMAN';
const PANE_STATE_IDLE_COMPLETE = 'IDLE_COMPLETE';

// Cap on captured child output kept in memory / sent to the hub, so a chatty
// CLI cannot grow the buffer without bound. The tail is what matters for an
// audit trail, mirroring TMUX_TAIL_LINES on the interactive path.
const HEADLESS_MAX_OUTPUT_BYTES = 1048576; // 1 MiB

const TMUX_TAIL_LINES = 15;
const HEARTBEAT_INTERVAL_MS = 30000;
const HEARTBEAT_TIMEOUT_MS = 90000;
const PROGRESS_REPORT_INTERVAL_MS = 120000;
const MAX_RECONNECT_DELAY_MS = 60000;
const BASE_RECONNECT_DELAY_MS = 1000;
const TOKEN_REFRESH_MARGIN_MS = 300000;
const MAX_TASK_DURATION_MS = 1800000;
// Hard ceiling on a single headless one-shot invocation (kubestellar/hive#2538).
// The interactive path bounds a task with MAX_TASK_DURATION_MS via a
// tmux-scraping watchdog; the headless child gets the SAME bound enforced
// directly on the process, so a wedged CLI is killed and reported failed rather
// than hanging the pod forever.
const HEADLESS_TASK_TIMEOUT_MS = MAX_TASK_DURATION_MS;
const NETWORK_ERROR_RETRY_DELAY_MS = 5000;
// After the hub sends an explicit task_unavailable negative-ack (no admissible
// work, a disabled tier, a concurrency limit, or a token-mint failure — see
// kubestellar/hive#2436), wait before re-asking so we neither hang forever
// (the old silent-nil behaviour) nor busy-loop the hub.
const TASK_UNAVAILABLE_RETRY_MS = 30000;

// RELAY_PROTOCOL_VERSION is the contributor-protocol version this relay speaks
// (kubestellar/hive#2567). It is DECLARED to the hub in auth_response (additive,
// optional — an older hub simply ignores it) and the hub advertises its own
// version + capability set back on auth_ok. Keep in step with
// contributorProtocolVersion in v2/pkg/dashboard/contribute_protocol.go.
const RELAY_PROTOCOL_VERSION = '1.1';

// Per-task CLI-crash retry budget. Issue #2203: a task whose CLI kept dying was
// reassigned by the hub and failed identically forever (5+ times in ~20min),
// starving that hub task slot. After MAX_TASK_CLI_RESTARTS crash-restarts for
// the SAME repo#number, the relay gives up on that task permanently and tells
// the hub so it can be reassigned elsewhere.
const MAX_TASK_CLI_RESTARTS = 3;
// Backoff before each successive restart of the same task: 5s, 10s, 20s.
const TASK_RESTART_BASE_BACKOFF_MS = 5000;
const TASK_RESTART_MAX_BACKOFF_MS = 60000;
// How long a permanently-given-up task stays on the deny list. Long enough
// that the hub does not immediately hand the same poison task back, short
// enough that a transient environment fault eventually clears.
const GIVE_UP_MEMORY_MS = 3600000;

if (rawTokenList.length === 0) {
  console.error('FATAL: HIVE_REGISTRATION_TOKEN not set. Run `just contribute-register` first.');
  process.exit(1);
}

// One entry per hub, each owning its own connection/reconnect/heartbeat state.
// currentTask, cliReady and everything CLI-facing stay single global values
// below — there is exactly one CLI/tmux session, shared across whichever hub
// currently holds the active task or is being polled for work.
const hubs = rawHubList.map((url, i) => ({
  url: url.replace(/\/contribute\/?$/, '/api/contribute/ws'),
  regToken: rawTokenList[i] || rawTokenList[0],
  ws: null,
  reconnectDelay: BASE_RECONNECT_DELAY_MS,
  heartbeatInterval: null,
  lastPong: Date.now(),
  connectGeneration: 0,
  reconnectTimer: null,
  authenticated: false,
  authFailed: false,
}));
// Index into hubs[] of the hub we are currently soliciting work from (sent it
// the last 'ready'), or that owns currentTask. Round-robins forward only on an
// explicit task_unavailable from the active hub; completing/failing a task
// sticks with the same hub and re-asks it.
let activeHubIndex = 0;

let seq = 0;
let currentTask = null;
let progressInterval = null;
let tokenExpiresAt = null;

function nextSeq() { return ++seq; }

function sendTo(hub, msg) {
  if (hub && hub.ws && hub.ws.readyState === WebSocket.OPEN) {
    hub.ws.send(JSON.stringify(msg));
  }
}

// send() targets the hub that owns the in-flight task, or (idle) the hub
// currently being polled for work. Messages that must go to a specific hub
// regardless of active task use sendTo() directly.
function send(msg) {
  sendTo((currentTask && currentTask._hub) || hubs[activeHubIndex], msg);
}

function currentTaskHub() {
  return (currentTask && currentTask._hub) || hubs[activeHubIndex];
}

function advanceActiveHub(fromHub) {
  const fromIndex = hubs.indexOf(fromHub);
  const start = fromIndex >= 0 ? fromIndex : activeHubIndex;
  for (let offset = 1; offset <= hubs.length; offset++) {
    const idx = (start + offset) % hubs.length;
    if (!hubs[idx].authFailed) {
      activeHubIndex = idx;
      return hubs[idx];
    }
  }
  return null;
}

function injectGhToken(token) {
  const dir = path.dirname(GH_TOKEN_CACHE);
  try { fs.mkdirSync(dir, { recursive: true }); } catch (_) {}
  fs.writeFileSync(GH_TOKEN_CACHE, token, { mode: 0o600 });
}

const CLI_READY_POLL_MS = 2000;
const CLI_READY_TIMEOUT_MS = 600000;
const CONTAINER_NAME = process.env.HIVE_CONTAINER_NAME || 'hive-contributor';

// detectCapabilities builds the OPTIONAL, client-declared capability object the
// relay reports in auth_response (kubestellar/hive#2547, declare half). Every
// entry is a cheap, honest self-report the hub STORES + SURFACES read-only and
// NEVER routes/gates on. It is best-effort: any probe that throws is simply
// omitted, so a constrained environment still authenticates unchanged. Computed
// once at startup and cached.
let cachedCapabilities = null;
function detectCapabilities() {
  if (cachedCapabilities) return cachedCapabilities;
  const caps = {
    os: process.platform,
    arch: process.arch,
    relay_protocol_version: RELAY_PROTOCOL_VERSION,
  };
  // Container runtime: prefer docker, then podman, else none. `command -v` is a
  // cheap presence check; failure just means the runtime is absent.
  let runtime = 'none';
  for (const rt of ['docker', 'podman']) {
    try {
      execSync(`command -v ${rt}`, { stdio: 'ignore' });
      runtime = rt;
      break;
    } catch (_) { /* not installed */ }
  }
  caps.container_runtime = runtime;
  // Credential type: the KIND of GitHub credential the relay authenticates with
  // (never the credential itself). App-token cache present → "app"; an explicit
  // GH_TOKEN/GITHUB_TOKEN in the environment → "pat"; otherwise leave unset.
  try {
    if (fs.existsSync(GH_TOKEN_CACHE)) {
      caps.credential_type = 'app';
    } else if (process.env.GH_TOKEN || process.env.GITHUB_TOKEN) {
      caps.credential_type = 'pat';
    }
  } catch (_) { /* ignore */ }
  cachedCapabilities = caps;
  return caps;
}

// Backends that must NOT be given --model, mirroring contributor-agent.sh.
// amazonq/goose take their model from config/env. bob is excluded because
// --model is actively FATAL for it: bob auto-selects its own model and passing
// one leaves its model config undefined, so every prompt dies with
// "Cannot read properties of undefined (reading 'maxTokens')" (bobshell 1.0.6).
const NO_MODEL_FLAG_BACKENDS = ['amazonq', 'goose', 'bob'];

// Single source of truth for the CLI launch command (issue #2203, bug 1).
// contributor-agent.sh builds "$CMD $PERM_FLAG $MODEL_FLAG" for the FIRST
// launch; every restart path in this file previously rebuilt only "$CMD $PERM"
// inline, silently dropping the resolved model for the rest of the container's
// life. Build it once here and reuse it everywhere so the paths cannot drift.
let cachedLaunchCommand = null;
let cachedBackendResolution = null;

// resolveBackend() returns the { cmd, perm } pair backends.conf maps this
// backend to (binary + permission flags). Shared by the interactive launch
// command and the headless argv builder so the two paths cannot drift on which
// binary/flags a backend uses. Result cached — the resolution is a couple of
// bash sub-shells and never changes for the life of the process.
function resolveBackend() {
  if (cachedBackendResolution) return cachedBackendResolution;
  const confPaths = ['/usr/local/etc/hive/backends.conf', path.join(process.cwd(), 'config/backends.conf')];
  const confPath = confPaths.find(p => fs.existsSync(p)) || confPaths[0];
  let cmd = BACKEND;
  let perm = '';
  try {
    cmd = execSync(`bash -c 'source ${confPath} 2>/dev/null; backend_binary ${BACKEND}'`, { encoding: 'utf8', timeout: 15000 }).trim() || BACKEND;
    perm = execSync(`bash -c 'source ${confPath} 2>/dev/null; backend_perm_flag ${BACKEND}'`, { encoding: 'utf8', timeout: 15000 }).trim();
  } catch (e) {
    console.error(`Could not resolve backend flags from ${confPath}: ${e.message}`);
  }
  cachedBackendResolution = { cmd, perm };
  return cachedBackendResolution;
}

function buildLaunchCommand() {
  if (cachedLaunchCommand) return cachedLaunchCommand;
  const { cmd, perm } = resolveBackend();
  const modelFlag = MODEL && !NO_MODEL_FLAG_BACKENDS.includes(BACKEND) ? `--model ${MODEL}` : '';
  cachedLaunchCommand = [cmd, perm, modelFlag].filter(Boolean).join(' ');
  return cachedLaunchCommand;
}

// --- Headless (non-interactive) one-shot dispatch (kubestellar/hive#2538) ---
//
// Backends whose CLI supports a one-shot / print invocation that takes the
// prompt on the command line, runs to completion, and EXITS with a meaningful
// status — the property the headless mode needs. Each entry says how to turn
// (binary, perm-flags, prompt) into an argv:
//
//   flag — the sub-command/flag(s) that select one-shot mode. Either a single
//          token, where the prompt follows as a bare positional
//          (`claude -p "<prompt>"`, `codex exec "<prompt>"`), or an array of
//          leading tokens when a sub-command AND a flag both precede the
//          prompt (`goose run --no-session -t "<prompt>"`). Either way the
//          prompt is appended as the final, distinct argv element.
//
// Backends NOT listed here have no known non-interactive entry point (bob /
// agy / pi drive an interactive TUI), so headless mode refuses them LOUDLY at
// task time rather than silently stalling. Extending this table is how a
// future PR adds a backend once its headless invocation is verified.
const HEADLESS_BACKENDS = {
  // claude -p "<prompt>" — print mode: runs the prompt non-interactively and
  // exits. Same perm flags as the interactive launch (bypass permissions).
  claude: { flag: '-p' },
  // litellm is the claude binary pointed at a LiteLLM proxy, so the same
  // print-mode invocation applies.
  litellm: { flag: '-p' },
  // copilot -p "<prompt>" — non-interactive programmatic mode.
  copilot: { flag: '-p' },
  // codex exec "<prompt>" — Codex's non-interactive execution sub-command.
  codex: { flag: 'exec' },
  // goose run --no-session -t "<prompt>" — goose's one-shot sub-command. The
  // bare `goose` binary drives the interactive TUI, but `goose run` is a
  // documented non-interactive entry point (#2828): `-t` takes the prompt as
  // its VALUE (not a trailing positional), and --no-session skips creating or
  // resuming a session file, which one-shot dispatch never needs. Verified
  // against goose 1.37.0 — the version v2/Dockerfile pins via GOOSE_VERSION —
  // that `run`, `-t` and `--no-session` all exist and that a failed run exits
  // non-zero, which is the exit-code contract runHeadlessTask() relies on.
  goose: { flag: ['run', '--no-session', '-t'] },
};

// headlessSupportsBackend reports whether the configured backend has a known
// one-shot invocation. Used to fail fast at startup and per task.
function headlessSupportsBackend() {
  return Object.prototype.hasOwnProperty.call(HEADLESS_BACKENDS, BACKEND);
}

// buildHeadlessArgv turns a task prompt into the argv for a one-shot,
// non-interactive backend invocation: [binary, ...permFlags, ...modelFlag,
// ...oneShotFlags, prompt]. Returns null for an unsupported backend. Never
// shell-interpolates the prompt — it is passed as a distinct argv element to
// execFile, so apostrophes/quotes in the prompt (the exact #2203 wedge on the
// interactive path) cannot break anything here.
function buildHeadlessArgv(prompt) {
  const spec = HEADLESS_BACKENDS[BACKEND];
  if (!spec) return null;
  const { cmd, perm } = resolveBackend();
  const permArgs = perm ? perm.split(/\s+/).filter(Boolean) : [];
  const modelArgs = MODEL && !NO_MODEL_FLAG_BACKENDS.includes(BACKEND) ? ['--model', MODEL] : [];
  // spec.flag is a single token for most backends, or an array of leading
  // tokens for backends needing a sub-command plus a flag (goose). Normalize
  // to an array so both shapes spread the same way ahead of the prompt.
  const oneShotArgs = Array.isArray(spec.flag) ? spec.flag : [spec.flag];
  const args = [...permArgs, ...modelArgs, ...oneShotArgs, prompt];
  return { bin: cmd, args };
}

// writeHeadlessStatus records the runner's coarse lifecycle state so a
// supervising process / K8s probe can read it. Best-effort: a failed write is
// logged-by-omission and never aborts the task.
function writeHeadlessStatus(state, extra) {
  const payload = Object.assign({
    mode: MODE_HEADLESS,
    backend: BACKEND,
    state,
    updated_at: new Date().toISOString(),
  }, extra || {});
  try {
    fs.writeFileSync(HEADLESS_STATUS_FILE, JSON.stringify(payload, null, 2));
  } catch (_) { /* probe file is advisory; never fail a task on it */ }
  return payload;
}

// Reference to the in-flight headless child, so a revoke/shutdown can kill it.
let headlessChild = null;

// runHeadlessTask drives a single task to completion WITHOUT tmux: it spawns the
// one-shot CLI, captures (bounded) output, and on exit reports task_complete
// (exit 0) or task_failed (non-zero / spawn error / timeout) over the existing
// WebSocket channel — then announces `ready` for the next task. This is the
// headless analogue of the interactive progressTick() completion path.
function runHeadlessTask(task) {
  const prompt = task.prompt || `Work on ${task.kind} ${task.repo}#${task.number}: ${task.title}`;
  if (!headlessSupportsBackend()) {
    // No non-interactive entry point for this backend: fail LOUDLY rather than
    // stall. This is the #2538 guarantee — a headless run never waits silently.
    const reason = `backend '${BACKEND}' has no headless (non-interactive) mode; supported: ${Object.keys(HEADLESS_BACKENDS).join(', ')}`;
    console.error(`Headless dispatch refused: ${reason}`);
    writeHeadlessStatus(HEADLESS_STATE_FAILED, { task_id: task.task_id, reason });
    failCurrentTask(reason, { permanent: true });
    return;
  }

  const { bin, args } = buildHeadlessArgv(prompt);
  console.log(`Headless: running ${bin} (one-shot) for ${task.repo}#${task.number}`);
  writeHeadlessStatus(HEADLESS_STATE_WORKING, { task_id: task.task_id, repo: task.repo, number: task.number });
  send({ type: 'task_progress', seq: nextSeq(), task_id: task.task_id, kind: task.kind, repo: task.repo, number: task.number, title: task.title, status: 'working' });

  let settled = false;
  const finish = (fn) => { if (settled) return; settled = true; fn(); };

  headlessChild = execFile(bin, args, {
    timeout: HEADLESS_TASK_TIMEOUT_MS,
    maxBuffer: HEADLESS_MAX_OUTPUT_BYTES,
    killSignal: 'SIGKILL',
    cwd: process.env.HIVE_WORKSPACE_DIR || process.cwd(),
  }, (err, stdout, stderr) => {
    headlessChild = null;
    // Tokens can appear in agent output; redact before the tail leaves the host.
    const outTail = redactTokens(String(stdout || '') + String(stderr || ''))
      .split('\n').slice(-TMUX_TAIL_LINES);
    if (err) {
      // A non-zero exit, a spawn failure (ENOENT), or the timeout kill all land
      // here. err.killed && err.signal signals the timeout; report a real
      // failure either way so the hub can reassign — never a silent hang.
      const timedOut = err.killed === true;
      const reason = timedOut
        ? `headless task exceeded ${HEADLESS_TASK_TIMEOUT_MS / 60000}min and was killed`
        : `headless CLI exited with error: ${err.code !== undefined ? `code ${err.code}` : err.message}`;
      finish(() => {
        console.error(`Headless task ${task.task_id} failed: ${reason}`);
        writeHeadlessStatus(HEADLESS_STATE_FAILED, { task_id: task.task_id, reason });
        failCurrentTask(reason, { permanent: false });
      });
      return;
    }
    finish(() => {
      console.log(`Headless task ${task.task_id} completed (exit 0)`);
      const prURL = detectPRURL(outTail, task.repo);
      if (prURL) console.log(`Detected PR for ${task.task_id}: ${prURL}`);
      writeHeadlessStatus(HEADLESS_STATE_DONE, { task_id: task.task_id, pr_url: prURL });
      send({ type: 'task_complete', seq: nextSeq(), task_id: task.task_id, result: 'completed', summary: 'Headless one-shot invocation exited 0', tmux_output: outTail, pr_url: prURL });
      currentTask = null;
      taskAssignedAt = 0;
      tasksCompletedCount++;
      writeHeadlessStatus(HEADLESS_STATE_WAITING);
      send({ type: 'ready', seq: nextSeq() });
    });
  });
}

// A tmux pane can be left in bash's PS2 continuation state ("> ") when task
// text is typed into a bare shell — the prompt contains literal apostrophes
// (e.g. 'gh repo fork ... --clone=false'), so bash's readline sees an
// unbalanced quote and swallows everything typed afterwards, including the
// relaunch command (issue #2203). Clear the line before relaunching.
function recoverWedgedShell() {
  try {
    execSync(`tmux send-keys -t ${TMUX_SESSION} C-c`, { timeout: 15000 });
    execSync(`tmux send-keys -t ${TMUX_SESSION} C-u`, { timeout: 15000 });
    execSync(`tmux send-keys -t ${TMUX_SESSION} Enter`, { timeout: 15000 });
  } catch (_) {}
}

function capturePaneText() {
  try {
    return execSync(
      `tmux capture-pane -t ${TMUX_SESSION} -p 2>/dev/null`,
      { encoding: 'utf8', timeout: 15000 }
    ).toString();
  } catch (_) {
    return '';
  }
}

function blockingPromptKey(text) {
  // codex: "Do you trust the contents of this directory?" → 1. Yes, continue
  if (/Do you trust the contents of this directory/.test(text)) return '1';
  // codex: "✨ Update available! x -> y" → 3. Skip until next version.
  // Deliberately not "1. Update now", which shells out to npm install -g.
  if (/Update available!/.test(text) && /Skip until next version/.test(text)) return '3';
  return null;
}

function getCLIState() {
  try {
    const text = capturePaneText();
    if (BACKEND === 'claude') {
      if (/Not logged in|Please run \/login/.test(text)) return 'needs-login';
      if (/bypass permissions|Welcome back|Try "how does|medium.*effort|@gmail\.com|@.*\.com.*Organization/.test(text)) return 'ready';
      if (/Choose the text style|trust this folder/.test(text)) return 'onboarding';
    } else if (BACKEND === 'copilot') {
      if (/copilot login|gh auth login/.test(text)) return 'needs-login';
      if (/Confirm folder trust|trust the files|Do you trust/.test(text)) return 'onboarding';
      if (/\/ commands.*help/.test(text)) return 'ready';
    } else if (BACKEND === 'gemini') {
      if (/not authenticated|login required/i.test(text)) return 'needs-login';
      if (/>\s*$|❯/.test(text)) return 'ready';
    } else if (BACKEND === 'goose') {
      if (/goose is ready|> Enter to send|>\s*$|goose>|G\s*>/.test(text)) return 'ready';
    } else if (BACKEND === 'bob') {
      // Order matters: the blocked states are checked FIRST because bob's
      // auth prompt contains the literal string "Bob-Shell" ("Enter Bob-Shell
      // API Key"). The former /Bob-Shell/ 'ready' test therefore reported a
      // bob stuck at the API-key prompt as READY, and the relay would dispatch
      // tasks into a pane that could never run them.
      if (/Enter Bob-Shell API Key|enter your Bob-Shell API key|Paste your API key here/i.test(text)) return 'needs-login';
      if (/Do you trust this folder|not trusted/i.test(text)) return 'onboarding';
      // Real prompt chrome of an authenticated, ready bob TUI. Matching the
      // status line ("Auto-approve:", "Tokens left:") or the boxed input hint
      // is far tighter than the old />\s*$/, which matched almost any pane
      // that happened to end in a '>' — including partially drawn frames.
      if (/Enter your prompt, \/ for commands|Auto-approve:|Tokens left:/.test(text)) return 'ready';
    } else if (BACKEND === 'codex') {
      // Order matters: codex can leave ready-looking banner chrome behind
      // modal prompts, so classify those prompts before matching readiness.
      if (/Do you trust the contents of this directory/.test(text)) return 'onboarding';
      if (/Update available!/.test(text) && /Skip until next version/.test(text)) return 'onboarding';
      // codex renders its input marker as '›' (U+203A), not '>', and its
      // banner reads "OpenAI Codex (vX.Y.Z)" — not the literal "Codex CLI".
      if (/codex>|›|OpenAI Codex|Codex CLI|>\s*$/.test(text)) return 'ready';
    } else if (BACKEND === 'pi') {
      if (/pi v\d|0\.0%|auto\)|\d+\.\d+%/.test(text)) return 'ready';
    } else if (BACKEND === 'agy') {
      // agy shows "? for shortcuts" at the bottom when its interactive prompt
      // is ready. The generic />\s*$/ fires too early (during the splash).
      if (/\? for shortcuts/.test(text)) return 'ready';
    } else {
      if (/>\s*$|❯|\$\s*$/.test(text)) return 'ready';
    }
    return 'starting';
  } catch (_) {
    return 'starting';
  }
}

function waitForCLI() {
  let loginMessageShown = false;
  return new Promise((resolve, reject) => {
    const start = Date.now();
    const check = () => {
      const state = getCLIState();
      if (state === 'ready') {
        console.log('CLI ready — accepting tasks');
        resolve();
      } else if (state === 'onboarding') {
        const key = blockingPromptKey(capturePaneText());
        console.log(key ? `Auto-dismissing trust/onboarding dialog (selecting "${key}")...` : 'Auto-dismissing trust/onboarding dialog...');
        try {
          if (key) execSync(`tmux send-keys -t ${TMUX_SESSION} ${key} Enter`, { timeout: 15000 });
          else execSync(`tmux send-keys -t ${TMUX_SESSION} Enter`, { timeout: 15000 });
        } catch (_) {}
        setTimeout(check, CLI_READY_POLL_MS);
      } else if (state === 'needs-login' && !loginMessageShown) {
        loginMessageShown = true;
        console.log('');
        console.log('╔══════════════════════════════════════════════════════════╗');
        console.log('║  Claude Code needs authentication.                      ║');
        console.log('║  In another terminal, run:                              ║');
        console.log(`║  docker exec -it ${CONTAINER_NAME} tmux attach -t ${TMUX_SESSION}`);
        console.log('║  Then type: /login                                      ║');
        console.log('║  Complete the login, then press Ctrl-B D to detach.     ║');
        console.log('║  Waiting for login to complete...                       ║');
        console.log('╚══════════════════════════════════════════════════════════╝');
        console.log('');
        setTimeout(check, CLI_READY_POLL_MS);
      } else if (Date.now() - start > CLI_READY_TIMEOUT_MS) {
        reject(new Error('CLI did not become ready within timeout'));
      } else {
        setTimeout(check, CLI_READY_POLL_MS);
      }
    };
    check();
  });
}

let cliReady = false;
let pendingTask = null;

if (CONTRIBUTOR_MODE === MODE_HEADLESS) {
  // Headless mode has no tmux pane to scrape for readiness. Each task spawns
  // its own one-shot CLI process on demand, so there is nothing to "become
  // ready" — the relay is ready to accept work as soon as it authenticates.
  // Fail fast on an unsupported backend so a K8s pod reports a real error at
  // startup instead of accepting a task it can never run.
  cliReady = true;
  if (!headlessSupportsBackend()) {
    console.error(`FATAL: CONTRIBUTOR_MODE=headless but backend '${BACKEND}' has no non-interactive mode. Supported: ${Object.keys(HEADLESS_BACKENDS).join(', ')}`);
    writeHeadlessStatus(HEADLESS_STATE_FAILED, { reason: `unsupported headless backend: ${BACKEND}` });
    if (process.env.HIVE_RELAY_TEST_MODE !== '1') process.exit(1);
  } else {
    console.log(`Headless mode: backend '${BACKEND}' will run one-shot per task (no tmux).`);
    writeHeadlessStatus(HEADLESS_STATE_WAITING);
  }
} else {
  waitForCLI().then(() => {
    cliReady = true;
    flushPendingTask();
  }).catch(e => console.error(e.message));
}

const ENTER_COUNT = 3;
const ENTER_DELAY_MS = 300;

function sleepMs(ms) {
  // Tests drive the restart/backoff paths synchronously; a real busy-wait
  // would make the suite take minutes of wall clock for no added coverage.
  if (process.env.HIVE_RELAY_TEST_MODE === '1') return;
  const end = Date.now() + ms;
  while (Date.now() < end) {
    try { execSync(`sleep 0.1`, { timeout: 5000 }); } catch (_) {}
  }
}

function tmuxSendEnters() {
  for (let i = 0; i < ENTER_COUNT; i++) {
    execSync(`tmux send-keys -t ${TMUX_SESSION} Enter`, { timeout: 15000 });
    if (i < ENTER_COUNT - 1) sleepMs(ENTER_DELAY_MS);
  }
}

const CLEAR_CONTEXT_THRESHOLD_PCT = 70;

function checkContextUsage() {
  try {
    const output = execSync(
      `tmux capture-pane -t ${TMUX_SESSION} -p -S -3 2>/dev/null`,
      { encoding: 'utf8', timeout: 15000 }
    );
    const match = output.match(/ctx:(\d+)%|(\d+)% context/);
    return match ? parseInt(match[1] || match[2], 10) : 0;
  } catch (_) {
    return 0;
  }
}

function tmuxSendKeys(text) {
  // Hard gate (issue #2203, bug 2): `send-keys -l` types literal keystrokes
  // into whatever owns the pane. If the CLI is not confirmed ready, those
  // keystrokes land on bash, whose readline chokes on the apostrophes in the
  // prompt and drops the pane into PS2 continuation, wedging it permanently.
  // Queue instead; flushPendingTask() delivers it once readiness is confirmed.
  if (!cliReady) {
    console.log('CLI not ready — queuing task prompt instead of typing into the pane');
    pendingTask = text;
    return;
  }
  try {
    try {
      execSync(`find /tmp -maxdepth 1 -type d -user dev -not -name 'tmux-*' -not -name 'claude-*' -not -name 'node-*' -not -name '.' -mmin +60 -exec rm -rf {} + 2>/dev/null; find /tmp -maxdepth 1 -type f -user dev -name '*.out' -o -name '*.html' -mmin +60 -exec rm -f {} + 2>/dev/null`, { timeout: 15000 });
    } catch (_) {}
    const ctxPct = checkContextUsage();
    const RESET_EVERY_N = 3;
    const needsClaudeClear = BACKEND === 'claude' && ctxPct >= CLEAR_CONTEXT_THRESHOLD_PCT;
    // Fire the periodic memory-cleanup restart at most ONCE per threshold
    // crossing (issue #2596). Requiring tasksCompletedCount !== lastResetAtCount
    // stops the #2203 readiness guard from re-triggering the restart when it
    // re-enters tmuxSendKeys() at the same, unchanged count — the re-entry that
    // otherwise loops forever and starves the next task.
    const needsCliRestart = BACKEND !== 'claude' && tasksCompletedCount > 0 &&
      tasksCompletedCount % RESET_EVERY_N === 0 && tasksCompletedCount !== lastResetAtCount;
    if (needsClaudeClear) {
      console.log(`Context at ${ctxPct}% — sending /clear before next task`);
      execSync(`tmux send-keys -t ${TMUX_SESSION} Escape`, { timeout: 15000 });
      sleepMs(200);
      execSync(`tmux send-keys -t ${TMUX_SESSION} C-a`, { timeout: 15000 });
      execSync(`tmux send-keys -t ${TMUX_SESSION} C-k`, { timeout: 15000 });
      sleepMs(200);
      execSync(`tmux send-keys -t ${TMUX_SESSION} -l '/clear'`, { timeout: 15000 });
      sleepMs(200);
      tmuxSendEnters();
      sleepMs(3000);
    } else if (needsCliRestart) {
      // Record that we serviced this count BEFORE relaunching, so when the
      // readiness callback flushes the queued prompt back through here the
      // predicate is already false and we fall through to deliver the next task
      // instead of restarting again (issue #2596).
      lastResetAtCount = tasksCompletedCount;
      console.log(`Restarting ${BACKEND} CLI for memory cleanup (task ${tasksCompletedCount})`);
      execSync(`tmux send-keys -t ${TMUX_SESSION} C-c`, { timeout: 15000 });
      sleepMs(1000);
      execSync(`tmux send-keys -t ${TMUX_SESSION} C-c`, { timeout: 15000 });
      sleepMs(2000);
      // Queue this prompt and hand delivery to the readiness callback.
      // Previously the restart set cliReady=false and then FELL THROUGH to the
      // send loop below, typing the prompt into a pane where the CLI had just
      // been Ctrl-C'd and had not come back — the exact sequence in #2203.
      pendingTask = text;
      cliReady = false;
      try {
        console.log(`CLI restarted: ${relaunchCLI()}`);
      } catch (e) {
        console.error('CLI restart failed:', e.message);
      }
      return;
    }
    const MAX_SEND_RETRIES = 3;
    const RETRY_DELAY_MS = 10000;
    let sent = false;
    for (let attempt = 1; attempt <= MAX_SEND_RETRIES; attempt++) {
      try {
        execSync(`tmux send-keys -t ${TMUX_SESSION} Escape`, { timeout: 15000 });
        sleepMs(200);
        execSync(`tmux send-keys -t ${TMUX_SESSION} C-a`, { timeout: 15000 });
        execSync(`tmux send-keys -t ${TMUX_SESSION} C-k`, { timeout: 15000 });
        sleepMs(200);
        execSync(`tmux send-keys -t ${TMUX_SESSION} -l ${shellQuote(text)}`, { timeout: 30000 });
        sleepMs(300);
        tmuxSendEnters();
        console.log('Task prompt sent to CLI');
        sent = true;
        break;
      } catch (e) {
        console.error(`tmux send-keys attempt ${attempt}/${MAX_SEND_RETRIES} failed: ${e.message}`);
      }
      if (!sent && attempt < MAX_SEND_RETRIES) {
        console.log(`Waiting ${RETRY_DELAY_MS/1000}s before retry...`);
        sleepMs(RETRY_DELAY_MS);
      }
    }
    if (!sent) console.error('All tmux send-keys attempts failed — task prompt lost');
  } catch (e) {
    console.error('tmux send-keys failed:', e.message);
  }
}

function shellQuote(s) {
  return "'" + s.replace(/'/g, "'\\''") + "'";
}

function redactTokens(text) {
  return text.replace(/gho_[A-Za-z0-9]{36}/g, 'gho_***REDACTED***')
    .replace(/ghp_[A-Za-z0-9]{36}/g, 'ghp_***REDACTED***')
    .replace(/ghs_[A-Za-z0-9]{36}/g, 'ghs_***REDACTED***');
}

function captureTmuxLines(n) {
  try {
    const output = execSync(
      `tmux capture-pane -t ${TMUX_SESSION} -p -S -${n} 2>/dev/null`,
      { encoding: 'utf8', timeout: 15000 }
    );
    return output.trim().split('\n').slice(-n).map(l => redactTokens(l));
  } catch (_) {
    return [];
  }
}

// Best-effort scan of the agent's recent output for a GitHub pull-request URL
// it opened for this task. Reported on task_complete as pr_url so the hub can
// tell "work shipped" from "agent merely went idle" and pick the right issue
// cooldown (kubestellar/hive#2393 item 7). This is intentionally best-effort:
// when no PR link is visible we return '' and the hub applies its short no-PR
// cooldown. When `repo` is known (owner/repo) we prefer a URL under that repo
// so an unrelated PR mentioned in passing does not get attributed to the task.
function detectPRURL(lines, repo) {
  if (!Array.isArray(lines) || lines.length === 0) return '';
  // Matches https://github.com/<owner>/<repo>/pull/<number>, capturing owner/repo.
  const PR_URL_RE = /https:\/\/github\.com\/([A-Za-z0-9_.-]+\/[A-Za-z0-9_.-]+)\/pull\/\d+/g;
  let repoMatch = '';
  let anyMatch = '';
  for (const line of lines) {
    let m;
    PR_URL_RE.lastIndex = 0;
    while ((m = PR_URL_RE.exec(line)) !== null) {
      const url = m[0];
      if (!anyMatch) anyMatch = url;
      if (repo && m[1] === repo) { repoMatch = url; break; }
    }
    if (repoMatch) break;
  }
  // Prefer a URL under the task's own repo; otherwise fall back to the first
  // PR URL seen (better an approximate audit trail than none).
  return repoMatch || anyMatch;
}

// True while a bob CLI process is alive. bob exits at the end of every turn,
// so "process gone" means the turn finished — see the bob branch of
// checkTmuxIdle(). Matches the launch command rather than the bare name so a
// stray "bob" substring elsewhere in the process table cannot mask an exit.
const BOB_PROCESS_PATTERN = 'bob --accept-license';

function bobIsRunning() {
  try {
    let procs;
    if (fs.existsSync('/proc')) {
      procs = execSync(
        `for p in /proc/[0-9]*/cmdline; do tr "\\0" " " < "$p" 2>/dev/null; echo; done`,
        { encoding: 'utf8', timeout: 15000 }
      );
    } else {
      procs = execSync('ps -eo command 2>/dev/null', { encoding: 'utf8', timeout: 15000 });
    }
    return procs.includes(BOB_PROCESS_PATTERN);
  } catch (_) {
    // Unknown -> assume still running, so a probe failure cannot fabricate a
    // completion for a task that is actually still in flight.
    return true;
  }
}

function recentPaneLines(text, limit = 12) {
  return text
    .split('\n')
    .map(line => line.trim())
    .filter(Boolean)
    .slice(-limit);
}

function paneLooksBlockedOnHuman(text) {
  const lines = recentPaneLines(text);
  if (lines.length === 0) return false;
  const recent = lines.join('\n');
  const last = lines[lines.length - 1];
  const beforePrompt = [...lines].reverse().find(line =>
    !/^([>$❯]|goose>|G\s*>|> Enter to send|\/ commands.*help)$/i.test(line)
  ) || last;
  const currentMenuLine = /^(?:[❯>]\s*)?(?:\d+[\).]|[A-Za-z][\).])\s+\S+/.test(beforePrompt);

  const hasQuestion = /\?\s*$/.test(beforePrompt);
  const hasNumberedMenu =
    /\b(?:choose|select|option|pick|which|how (?:should|to) proceed|what (?:would|should).+like)\b/i.test(recent) &&
    currentMenuLine &&
    (recent.match(/(?:^|\n)\s*(?:[❯>]\s*)?\d+[\).]\s+\S+/g) || []).length >= 2;
  const blockingPatterns = [
    // Confirmation prompts and TUI continuation screens.
    /\[[Yy]\/[Nn]\]|\([Yy]\/[Nn]\)|\b[Yy]es\/[Nn]o\b/,
    /\b(?:continue|proceed|confirm|approve|allow|deny|accept|reject|choose|select)\b.*\?/i,
    /\bPress Enter to continue\b/i,
    /\bEnter to confirm\b/i,
    // Permission/auth/onboarding prompts seen from Claude/Copilot/Goose/Bob.
    /\b(?:approval|consent|trust this folder|Do you trust|Confirm folder trust)\b/i,
    /\bpermission\b.*\b(?:allow|approve|confirm|continue|proceed)\b/i,
    /\b(?:allow|approve|confirm|continue|proceed)\b.*\bpermission\b/i,
    /\b(?:Allow|Approve|Run|Execute)\b.*\b(?:command|tool|edit|file|operation)\b/i,
    /\b(?:Paste|Enter).*(?:API key|token|code|password)\b/i,
  ];

  return hasQuestion || hasNumberedMenu || blockingPatterns.some(re => re.test(beforePrompt));
}

function classifyTmuxPane(text) {
  let hasIdlePrompt, hasCompletionMarker, isWorking;

  if (BACKEND === 'claude') {
    const lastLines = text.split('\n').slice(-15).join('\n');
    hasIdlePrompt = /bypass permissions|shift\+tab to cycle/.test(text);
    hasCompletionMarker = /[✻✶✽] \S+ed for \d+[ms]|Honking|tokens\)/.test(text);
    isWorking = /─.*Bash\(|Reading|Editing|Writing|Searching/.test(lastLines) || /ing…/.test(lastLines);
  } else if (BACKEND === 'copilot') {
    hasIdlePrompt = /\/ commands.*help/.test(text);
    hasCompletionMarker = true;
    isWorking = /esc cancel/.test(text);
  } else if (BACKEND === 'gemini') {
    hasIdlePrompt = />\s*$|❯\s*$/.test(text);
    hasCompletionMarker = /completed|Done|finished/i.test(text);
    isWorking = /Thinking|Running|Searching/i.test(text);
  } else if (BACKEND === 'goose') {
    hasIdlePrompt = /goose is ready|> Enter to send|>\s*$|goose>|G\s*>/.test(text);
    hasCompletionMarker = true;
    isWorking = /working|running|executing|calling/i.test(text);
  } else if (BACKEND === 'bob') {
    const BOB_IDLE_CHROME = /Enter your prompt, \/ for commands|Auto-approve:|Tokens left:/;
    const BOB_SPINNER = /\(esc to cancel/;
    const bobRunning = bobIsRunning();
    hasIdlePrompt = BOB_IDLE_CHROME.test(text) || !bobRunning;
    hasCompletionMarker = true;
    isWorking = bobRunning && BOB_SPINNER.test(text);
  } else if (BACKEND === 'codex') {
    // Same marker mismatch as getCLIState(): '›' (U+203A), not '>'.
    hasIdlePrompt = /codex>|›|>\s*$/.test(text);
    hasCompletionMarker = /completed|done|finished/i.test(text);
    isWorking = /running|executing|thinking/i.test(text);
  } else if (BACKEND === 'pi') {
    hasIdlePrompt = /pi v\d|0\.0%|auto\)|\d+\.\d+%/.test(text);
    hasCompletionMarker = /completed|done|finished|tokens\)|\d+\.\d+%/i.test(text);
    isWorking = /Reading|Writing|Bash|Editing|thinking|running/i.test(text);
  } else if (BACKEND === 'agy') {
    hasIdlePrompt = /\? for shortcuts/.test(text);
    hasCompletionMarker = true;
    isWorking = /Running|Searching|Reading|Writing|Editing/i.test(text);
  } else {
    hasIdlePrompt = />\s*$|\$\s*$/.test(text);
    hasCompletionMarker = /completed|done|finished/i.test(text);
    isWorking = false;
  }

  if (paneLooksBlockedOnHuman(text)) return PANE_STATE_BLOCKED_ON_HUMAN;
  if (isWorking) return PANE_STATE_WORKING;
  if (hasIdlePrompt && hasCompletionMarker) return PANE_STATE_IDLE_COMPLETE;
  return PANE_STATE_WORKING;
}

function checkTmuxPaneState() {
  try {
    const output = execSync(
      `tmux capture-pane -t ${TMUX_SESSION} -p 2>/dev/null`,
      { encoding: 'utf8', timeout: 15000 }
    );
    const text = output.toString();
    const hasNetworkError = BACKEND === 'goose' && /Network error:|Please resend your message|Could not connect/i.test(text);
    if (hasNetworkError && /> Enter to send/.test(text)) {
      console.log('Goose network error detected — pressing Enter to retry');
      try {
        execSync(`tmux send-keys -t ${TMUX_SESSION} Enter`, { timeout: 15000 });
      } catch (_) {}
      return PANE_STATE_WORKING;
    }
    return classifyTmuxPane(text);
  } catch (_) {
    return PANE_STATE_WORKING;
  }
}

// Relaunch the backend CLI in the tmux session using the flags from
// backends.conf, the same way contributor-agent.sh first launched it.
function relaunchCLI() {
  const launchCmd = buildLaunchCommand();
  // The pane may be wedged in bash PS2 continuation; clear it or the relaunch
  // command is swallowed as more continuation text and never runs.
  recoverWedgedShell();
  execSync(`tmux send-keys -t ${TMUX_SESSION} ${shellQuote(launchCmd)} Enter`, { timeout: 15000 });
  // The CLI is NOT up yet. cliReady must stay false until the readiness
  // classifier positively confirms it, or a task prompt sent in the meantime
  // is typed as literal keystrokes into a bare shell (issue #2203, bug 2).
  cliReady = false;
  waitForCLI().then(() => {
    cliReady = true;
    flushPendingTask();
  }).catch(e => console.error(`CLI did not become ready after relaunch: ${e.message}`));
  return launchCmd;
}

function flushPendingTask() {
  if (!pendingTask) return;
  const t = pendingTask;
  pendingTask = null;
  tmuxSendKeys(t);
}

function checkTmuxIdle() {
  return checkTmuxPaneState() === PANE_STATE_IDLE_COMPLETE;
}

const TASK_GRACE_PERIOD_MS = 180000;
let taskAssignedAt = 0;
let tasksCompletedCount = 0;
// The completed-task count at which the periodic memory-cleanup restart last
// fired (issue #2596). The restart predicate below is re-entered by the #2203
// readiness/pending-task guard: the restart queues the prompt, clears cliReady,
// relaunches, and the readiness callback calls flushPendingTask() ->
// tmuxSendKeys() again. tasksCompletedCount only changes on an actual
// completion, so without latching, "count % RESET_EVERY_N === 0" stays true and
// the CLI restarts forever, never delivering the next task. Latching the count
// makes the reset one-shot per threshold crossing; a value that no real count
// reaches keeps the first crossing (count 0 is excluded anyway) from being
// treated as already-serviced.
let lastResetAtCount = -1;
const PR_REVIEW_EVERY_N = 5;
let taskTimeoutHandle = null;
let lastProgressTick = 0;

// Crash-restart bookkeeping, keyed by the underlying work item (repo#number)
// rather than task_id — the hub mints a NEW task_id on every reassignment of
// the same issue, so a task_id-keyed counter would reset each round and never
// reach the cap (issue #2203, bug 3).
const cliRestartCounts = new Map();
const givenUpTasks = new Map();

function taskKey(task) {
  return task && task.repo ? `${task.repo}#${task.number}` : (task && task.task_id) || 'unknown';
}

function isGivenUp(key) {
  const at = givenUpTasks.get(key);
  if (at === undefined) return false;
  if (Date.now() - at > GIVE_UP_MEMORY_MS) {
    givenUpTasks.delete(key);
    return false;
  }
  return true;
}

function restartBackoffMs(attempt) {
  return Math.min(TASK_RESTART_BASE_BACKOFF_MS * Math.pow(2, attempt - 1), TASK_RESTART_MAX_BACKOFF_MS);
}

function failCurrentTask(reason, opts) {
  if (!currentTask) return;
  const permanent = !!(opts && opts.permanent);
  const taskId = currentTask.task_id;
  const tmuxLines = captureTmuxLines(TMUX_TAIL_LINES);
  console.error(`Task ${taskId} failed${permanent ? ' permanently' : ''}: ${reason}`);
  send({ type: 'task_failed', seq: nextSeq(), task_id: taskId, reason, permanent, tmux_output: tmuxLines });
  currentTask = null;
  taskAssignedAt = 0;
  if (progressInterval) { clearInterval(progressInterval); progressInterval = null; }
  if (taskTimeoutHandle) { clearTimeout(taskTimeoutHandle); taskTimeoutHandle = null; }
  send({ type: 'ready', seq: nextSeq() });
}

function startProgressReporting() {
  if (progressInterval) clearInterval(progressInterval);
  if (taskTimeoutHandle) clearTimeout(taskTimeoutHandle);
  if (!taskAssignedAt) taskAssignedAt = Date.now();
  lastProgressTick = Date.now();

  taskTimeoutHandle = setTimeout(() => {
    if (currentTask) {
      failCurrentTask(`task exceeded max duration (${MAX_TASK_DURATION_MS / 60000}min)`);
    }
  }, MAX_TASK_DURATION_MS);

  progressInterval = setInterval(progressTick, PROGRESS_REPORT_INTERVAL_MS);
}

// One iteration of the progress/completion/crash-detection loop. Extracted from
// the setInterval body so it can be driven deterministically from tests.
function progressTick() {
  lastProgressTick = Date.now();
  if (!currentTask) return;
  if (Date.now() - taskAssignedAt < TASK_GRACE_PERIOD_MS) return;

  try {
    let procs = '';
    try {
      if (fs.existsSync('/proc')) {
        procs = execSync(`for p in /proc/[0-9]*/cmdline; do tr "\\0" " " < "$p" 2>/dev/null; done`, { encoding: 'utf8', timeout: 15000 });
      } else {
        procs = execSync(`ps -eo command 2>/dev/null`, { encoding: 'utf8', timeout: 15000 });
      }
    } catch (_) { procs = BACKEND; }
    const cliAlive = procs.includes(BACKEND) || procs.includes('claude') || procs.includes('copilot') || procs.includes('bob') || procs.includes('codex') || procs.includes('goose') || procs.includes('pi');
    // bob is not a persistent REPL: it exits at the end of every turn ("Bob
    // goes to sleep 💤"). For bob an exited process is the normal completion
    // signal, not a crash, so it must fall through to the checkTmuxIdle()
    // path below and be reported as task_complete. Treating it as a death
    // here reported finished work as task_failed on every single task.
    const cliExitIsNormal = BACKEND === 'bob';
    if (!cliAlive && !cliExitIsNormal) {
      const key = taskKey(currentTask);
      const attempt = (cliRestartCounts.get(key) || 0) + 1;
      cliRestartCounts.set(key, attempt);

      if (attempt > MAX_TASK_CLI_RESTARTS) {
        // Terminal give-up (issue #2203, bug 3). Do NOT relaunch on this
        // task's behalf again; report a permanent failure so the hub can
        // hand the work to a different contributor instead of looping.
        // The relay itself stays healthy and keeps accepting NEW tasks —
        // only this one work item is poisoned — so a single bad task can
        // never wedge the whole contributor.
        givenUpTasks.set(key, Date.now());
        cliRestartCounts.delete(key);
        failCurrentTask(
          `CLI process exited ${MAX_TASK_CLI_RESTARTS} times for ${key} — giving up on this task (relay still accepting other work)`,
          { permanent: true }
        );
        // Bring the CLI back so the next, different task can run.
        try { console.log(`CLI restarted: ${relaunchCLI()}`); } catch (e) { console.error('Failed to restart CLI:', e.message); }
        return;
      }

      const backoff = restartBackoffMs(attempt);
      console.error(`CLI process (${BACKEND}) died — restart ${attempt}/${MAX_TASK_CLI_RESTARTS} for ${key} after ${backoff / 1000}s backoff`);
      sleepMs(backoff);
      try {
        console.log(`CLI restarted: ${relaunchCLI()}`);
      } catch (e) {
        console.error('Failed to restart CLI:', e.message);
      }
      failCurrentTask('CLI process exited — restarted');
      return;
    }
  } catch (_) {}

  const paneState = checkTmuxPaneState();
  const tmuxLines = captureTmuxLines(TMUX_TAIL_LINES);
  if (paneState === PANE_STATE_IDLE_COMPLETE) {
    console.log(`Task ${currentTask.task_id} completed — agent idle`);
    // Successful completion clears this work item's crash-retry budget.
    cliRestartCounts.delete(taskKey(currentTask));
    // Best-effort: report the PR the agent opened, if one is visible in its
    // recent output, so the hub can distinguish "shipped a PR" from "just went
    // idle" and pick the right issue cooldown (kubestellar/hive#2393 item 7).
    // Empty when no PR link is found — the hub then applies the short cooldown.
    const prURL = detectPRURL(tmuxLines, currentTask.repo);
    if (prURL) console.log(`Detected PR for ${currentTask.task_id}: ${prURL}`);
    send({ type: 'task_complete', seq: nextSeq(), task_id: currentTask.task_id, result: 'completed', summary: 'Agent returned to idle', tmux_output: tmuxLines, pr_url: prURL });
    // bob exits after each turn, so the pane is now a bare shell. Bring it
    // back up before the next task, or the prompt would be typed into bash
    // ("-bash: <prompt>: command not found") and silently lost.
    if (BACKEND === 'bob' && !bobIsRunning()) {
      try {
        // relaunchCLI() clears cliReady and re-arms the readiness callback,
        // which flushes any queued prompt once the CLI is confirmed up.
        console.log(`Relaunching bob for the next task: ${relaunchCLI()}`);
      } catch (e) {
        console.error('Failed to relaunch bob:', e.message);
      }
    }
    const completedRepo = currentTask.repo;
    currentTask = null;
    taskAssignedAt = 0;
    clearInterval(progressInterval);
    progressInterval = null;
    if (taskTimeoutHandle) { clearTimeout(taskTimeoutHandle); taskTimeoutHandle = null; }
    tasksCompletedCount++;
    if (tasksCompletedCount % PR_REVIEW_EVERY_N === 0) {
      console.log(`PR review cycle (${tasksCompletedCount} tasks completed) — checking open PRs`);
      currentTask = { task_id: `pr-review-${Date.now()}`, kind: 'review', repo: completedRepo, number: 0, title: 'Review open PRs for comments' };
      taskAssignedAt = Date.now();
      const reviewPrompt = `Check your open PRs on ${completedRepo} for review comments. ` +
        `Run 'GH_TOKEN=$GH_TOKEN gh pr list --repo ${completedRepo} --author @me --state open' to find them. ` +
        `For each PR with review comments, read the comments, address the feedback, push fixes, and respond. ` +
        `If no PRs have comments, just say "No PR comments to address."`;
      tmuxSendKeys(reviewPrompt);
      startProgressReporting();
    } else {
      send({ type: 'ready', seq: nextSeq() });
    }
  } else if (paneState === PANE_STATE_BLOCKED_ON_HUMAN) {
    console.warn(`Task ${currentTask.task_id} is blocked waiting for human input`);
    send({
      type: 'task_progress',
      seq: nextSeq(),
      task_id: currentTask.task_id,
      status: 'blocked_on_human',
      attention: true,
      summary: 'Agent is waiting for human input in the tmux pane',
      tmux_output: tmuxLines,
    });
  } else {
    send({ type: 'task_progress', seq: nextSeq(), task_id: currentTask.task_id, status: 'working', tmux_output: tmuxLines });
  }
}

function handleMessage(data, hub) {
  // hub defaults to hubs[0] so existing single-hub callers (and the test
  // harness, which calls handleMessage(json) directly with no hub arg) keep
  // working unchanged — there is always at least one entry in hubs[].
  hub = hub || hubs[0];
  let msg;
  try { msg = JSON.parse(data); } catch (_) { return; }

  switch (msg.type) {
    case 'auth_challenge':
      // Always reply on the SAME hub that challenged us, regardless of
      // currentTask/activeHubIndex — send() would route elsewhere.
      sendTo(hub, {
        type: 'auth_response',
        seq: nextSeq(),
        registration_token: hub.regToken,
        cli_backend: BACKEND,
        model: MODEL,
        role: AGENT_ROLE,
        // #2547 declare half + #2567: additive, optional self-report of runtime
        // posture and protocol version. An older hub ignores these unknown fields.
        protocol_version: RELAY_PROTOCOL_VERSION,
        capabilities: detectCapabilities(),
      });
      break;

    case 'auth_ok':
      console.log(`Authenticated with ${hub.url} as ${msg.contributor_id} (tier: ${msg.trust_tier})`);
      // #2567: the hub advertises its protocol version + capability set here. We
      // log them (forward-compatible: unknown/absent fields are simply skipped)
      // so a newer relay can adapt to what the deployed server supports instead
      // of probing. No behaviour is gated on them today.
      if (msg.protocol_version || (msg.server_capabilities && msg.server_capabilities.length)) {
        console.log(`Hub protocol ${msg.protocol_version || 'unversioned'}; capabilities: ${(msg.server_capabilities || []).join(', ') || 'none'}`);
      }
      hub.authenticated = true;
      hub.authFailed = false;
      hub.reconnectDelay = BASE_RECONNECT_DELAY_MS;
      if (currentTask && hub === currentTaskHub()) {
        console.log(`Reconnected while working on ${currentTask.repo}#${currentTask.number} — resuming`);
        sendTo(hub, { type: 'task_accepted', seq: nextSeq(), task_id: currentTask.task_id });
        sendTo(hub, { type: 'task_progress', seq: nextSeq(), task_id: currentTask.task_id, kind: currentTask.kind, repo: currentTask.repo, number: currentTask.number, title: currentTask.title, status: 'working' });
        startProgressReporting();
      } else if (!currentTask && hub === hubs[activeHubIndex]) {
        // Only the hub currently in the poll rotation asks for work. A hub
        // that authenticates while it's not its turn just sits connected
        // (heartbeating) until task_unavailable rotates the active slot to it.
        sendTo(hub, { type: 'ready', seq: nextSeq() });
      }
      break;

    case 'auth_failed':
      console.error(`Authentication with ${hub.url} failed: ${msg.reason}`);
      if (msg.accepted_models && msg.accepted_models.length > 0) {
        console.error('\nThis hive accepts the following models:');
        msg.accepted_models.forEach(m => console.error('  - ' + m));
        console.error('\nSet your model: export AGENT_MODEL=<model>');
      }
      // A bad token for ONE hub must not take down a working connection to
      // another — only abort the whole process when every configured hub has
      // failed auth (or there is only one, matching the prior behaviour).
      hub.authFailed = true;
      if (hubs.every(h => h.authFailed)) {
        process.exit(1);
      } else {
        console.error(`Continuing with the remaining ${hubs.filter(h => !h.authFailed).length} hub(s).`);
        if (!currentTask && hub === hubs[activeHubIndex]) {
          const next = advanceActiveHub(hub);
          if (next && next.authenticated) {
            sendTo(next, { type: 'ready', seq: nextSeq() });
          }
        }
      }
      break;

    case 'task_assign':
      if (!currentTask && hub !== hubs[activeHubIndex]) {
        console.log(`Rejecting task ${msg.repo}#${msg.number} from ${hub.url} — hub is not the active polling slot`);
        sendTo(hub, { type: 'task_failed', seq: nextSeq(), task_id: msg.task_id, reason: 'Hub is not the active polling slot' });
        break;
      }
      if (currentTask) {
        console.log(`Rejecting task ${msg.repo}#${msg.number} from ${hub.url} — already working on ${currentTask.repo}#${currentTask.number}`);
        sendTo(hub, { type: 'task_failed', seq: nextSeq(), task_id: msg.task_id, reason: 'Already has active task' });
        break;
      }
      // A task we already gave up on permanently must not restart the loop if
      // the hub reassigns it anyway (issue #2203, bug 3). Reject it up front
      // and stay available for other work.
      if (isGivenUp(taskKey(msg))) {
        console.log(`Rejecting ${taskKey(msg)} — previously given up on after ${MAX_TASK_CLI_RESTARTS} CLI crashes`);
        sendTo(hub, { type: 'task_failed', seq: nextSeq(), task_id: msg.task_id, reason: `previously given up on after ${MAX_TASK_CLI_RESTARTS} CLI crashes`, permanent: true });
        sendTo(hub, { type: 'ready', seq: nextSeq() });
        break;
      }
      currentTask = msg;
      // Non-enumerable: currentTask IS msg, and msg gets JSON.stringify'd
      // wholesale to TASK_FILE a few lines down. hub carries live
      // setInterval/setTimeout handles, which are circular.
      Object.defineProperty(currentTask, '_hub', { value: hub, enumerable: false, writable: true, configurable: true });
      activeHubIndex = hubs.indexOf(hub);
      console.log(`Task assigned: ${msg.kind} ${msg.repo}#${msg.number} — ${msg.title} (from ${hub.url})`);
      if (msg.github_token) {
        injectGhToken(msg.github_token);
        tokenExpiresAt = msg.token_expires_at ? new Date(msg.token_expires_at).getTime() : null;
      }
      fs.writeFileSync(TASK_FILE, JSON.stringify(msg, null, 2));
      send({ type: 'task_accepted', seq: nextSeq(), task_id: msg.task_id });
      if (CONTRIBUTOR_MODE === MODE_HEADLESS) {
        // Non-interactive path (kubestellar/hive#2538): drive a one-shot CLI
        // invocation and report completion/failure from its exit status — no
        // tmux, no pane scraping, no watchdog waiting on an invisible prompt.
        runHeadlessTask(msg);
      } else {
        const taskPrompt = msg.prompt || `Work on ${msg.kind} ${msg.repo}#${msg.number}: ${msg.title}`;
        // tmuxSendKeys() itself queues when the CLI is not confirmed ready, so
        // there is a single gate rather than two that can disagree.
        tmuxSendKeys(taskPrompt);
        startProgressReporting();
      }
      break;

    case 'token_refresh':
      if (!currentTask || currentTaskHub() !== hub) {
        console.log(`Ignoring token_refresh from ${hub.url} — it does not own the active task`);
        break;
      }
      if (msg.github_token) {
        injectGhToken(msg.github_token);
        tokenExpiresAt = msg.token_expires_at ? new Date(msg.token_expires_at).getTime() : null;
        console.log('GitHub token refreshed');
      }
      break;

    case 'task_revoke':
      if (!currentTask) {
        console.log(`Ignoring task_revoked from ${hub.url} for ${msg.task_id} — no active task`);
        break;
      }
      if (currentTaskHub() !== hub || currentTask.task_id !== msg.task_id) {
        console.log(`Ignoring task_revoked from ${hub.url} for ${msg.task_id} — active task belongs to another hub`);
        break;
      }
      console.log(`Task revoked: ${msg.task_id} — ${msg.reason}`);
      currentTask = null;
      taskAssignedAt = 0;
      if (progressInterval) { clearInterval(progressInterval); progressInterval = null; }
      // Headless mode: kill the in-flight one-shot child so the revoked task's
      // process does not keep running (and holding the credential) after the
      // hub took the work back.
      if (CONTRIBUTOR_MODE === MODE_HEADLESS && headlessChild) {
        try { headlessChild.kill('SIGKILL'); } catch (_) {}
        headlessChild = null;
        writeHeadlessStatus(HEADLESS_STATE_WAITING);
      }
      // Stay with the hub that just revoked — it's clearly alive and reachable.
      activeHubIndex = hubs.indexOf(hub);
      sendTo(hub, { type: 'ready', seq: nextSeq() });
      break;

    case 'task_unavailable':
      if (hub !== hubs[activeHubIndex]) {
        console.log(`Ignoring task_unavailable from inactive hub ${hub.url}`);
        break;
      }
      // #2436 finding 1/2/3: the hub explicitly declined to assign work and
      // told us why. Multi-hub round-robin rotates only here, on this reliable
      // negative ack, never on success/failure/completion.
      console.log(`No task assigned on ${hub.url} — reason: ${msg.reason || 'unspecified'}; retrying in ${TASK_UNAVAILABLE_RETRY_MS / 1000}s`);
      setTimeout(() => {
        if (currentTask) return; // picked up work elsewhere in the meantime
        if (hubs.length > 1 && hub === hubs[activeHubIndex]) {
          advanceActiveHub(hub);
        }
        const next = hubs[activeHubIndex];
        // If `next` isn't connected/authenticated yet, its own auth_ok
        // handler sends 'ready' once it comes up and finds itself active.
        sendTo(next, { type: 'ready', seq: nextSeq() });
      }, TASK_UNAVAILABLE_RETRY_MS);
      break;

    case 'notice':
      console.log(msg.message || msg.reason || 'Notice from hub');
      break;

    case 'ping':
      sendTo(hub, { type: 'pong', seq: msg.seq });
      break;

    case 'pong':
      hub.lastPong = Date.now();
      break;

    default:
      console.log('Unknown message type:', msg.type);
  }
}

function connectHub(hub) {
  if (hub.reconnectTimer) { clearTimeout(hub.reconnectTimer); hub.reconnectTimer = null; }
  if (hub.heartbeatInterval) { clearInterval(hub.heartbeatInterval); hub.heartbeatInterval = null; }
  if (hub.ws) { try { hub.ws.removeAllListeners(); hub.ws.terminate(); } catch (_) {} }
  const gen = ++hub.connectGeneration;
  console.log(`Connecting to ${hub.url}...`);
  hub.ws = new WebSocket(hub.url);

  hub.ws.on('open', () => {
    if (gen !== hub.connectGeneration) return;
    console.log(`Connected to ${hub.url}`);
    hub.reconnectDelay = BASE_RECONNECT_DELAY_MS;
    hub.lastPong = Date.now();

    hub.heartbeatInterval = setInterval(() => {
      if (gen !== hub.connectGeneration) { clearInterval(hub.heartbeatInterval); return; }
      if (Date.now() - hub.lastPong > HEARTBEAT_TIMEOUT_MS) {
        console.error(`Heartbeat timeout on ${hub.url} — reconnecting`);
        hub.ws.terminate();
        return;
      }
      sendTo(hub, { type: 'ping', seq: nextSeq() });
    }, HEARTBEAT_INTERVAL_MS);
  });

  hub.ws.on('message', (data) => {
    if (gen !== hub.connectGeneration) return;
    handleMessage(data.toString(), hub);
  });

  hub.ws.on('close', () => {
    if (gen !== hub.connectGeneration) return;
    console.log(`Connection to ${hub.url} closed. Reconnecting in ${hub.reconnectDelay}ms...`);
    if (hub.heartbeatInterval) { clearInterval(hub.heartbeatInterval); hub.heartbeatInterval = null; }
    hub.reconnectTimer = setTimeout(() => connectHub(hub), hub.reconnectDelay);
    hub.reconnectDelay = Math.min(hub.reconnectDelay * 2, MAX_RECONNECT_DELAY_MS);
  });

  hub.ws.on('error', (err) => {
    if (gen !== hub.connectGeneration) return;
    console.error(`WebSocket error on ${hub.url}:`, err.message);
  });
}

function connect() {
  // Kept as the entry point (bottom of file, SIGTERM/SIGINT) so those call
  // sites don't need to know about hubs[] — connects every configured hub.
  hubs.forEach(connectHub);
}

function cleanup() {
  hubs.forEach(hub => {
    if (hub.heartbeatInterval) { clearInterval(hub.heartbeatInterval); hub.heartbeatInterval = null; }
  });
  if (progressInterval) { clearInterval(progressInterval); progressInterval = null; }
}

process.on('SIGTERM', () => { cleanup(); process.exit(0); });
process.on('SIGINT', () => { cleanup(); process.exit(0); });

// Test hook: when HIVE_RELAY_TEST_MODE=1 the relay exposes its internals and
// does NOT open a hub connection, so contributor-relay.test.js can drive the
// restart/queueing/give-up logic directly. Production runs never set this.
if (process.env.HIVE_RELAY_TEST_MODE === '1') {
  module.exports = {
    buildLaunchCommand,
    handleMessage,
    tmuxSendKeys,
    flushPendingTask,
    relaunchCLI,
    failCurrentTask,
    startProgressReporting,
    progressTick,
    classifyTmuxPane,
    paneLooksBlockedOnHuman,
    blockingPromptKey,
    PANE_STATE_WORKING,
    PANE_STATE_BLOCKED_ON_HUMAN,
    PANE_STATE_IDLE_COMPLETE,
    // Run one progress tick with the grace period already elapsed.
    __crashTick: () => { taskAssignedAt = Date.now() - TASK_GRACE_PERIOD_MS - 1; progressTick(); },
    cleanup,
    restartBackoffMs,
    NO_MODEL_FLAG_BACKENDS,
    MAX_TASK_CLI_RESTARTS,
    setCliReady: (v) => { cliReady = v; },
    getCliReady: () => cliReady,
    getPendingTask: () => pendingTask,
    setPendingTask: (v) => { pendingTask = v; },
    setTasksCompletedCount: (v) => { tasksCompletedCount = v; },
    getTasksCompletedCount: () => tasksCompletedCount,
    setLastResetAtCount: (v) => { lastResetAtCount = v; },
    getLastResetAtCount: () => lastResetAtCount,
    getCurrentTask: () => currentTask,
    setCurrentTask: (v) => { currentTask = v; },
    getCLIState,
    setWs: (w) => { hubs[0].ws = w; },
    getHubs: () => hubs,
    // Headless (non-interactive) mode surface (kubestellar/hive#2538).
    CONTRIBUTOR_MODE,
    MODE_INTERACTIVE,
    MODE_HEADLESS,
    HEADLESS_BACKENDS,
    HEADLESS_STATE_WAITING,
    HEADLESS_STATE_WORKING,
    HEADLESS_STATE_DONE,
    HEADLESS_STATE_FAILED,
    headlessSupportsBackend,
    buildHeadlessArgv,
    runHeadlessTask,
    getHeadlessChild: () => headlessChild,
  };
} else {
  connect();
}
