/**
 * Agent Run Dispatcher — ADR-0030 Phase 2 (dry-run mode).
 *
 * In-process worker that ticks every `_workflow_config.poll_interval_ms`
 * (default 30s) and drives the run-state machine on tickets table 1708.
 *
 * Phase 2 scope (dry-run):
 *   - Tick Part A: stall detection — no-op pass while no live runs exist.
 *   - Tick Part B: claim ready tickets and immediately cancel them with
 *     terminal_reason='phase2_dryrun'. Validates state machine + DB
 *     mutations + concurrency guards without running real agent work.
 *
 * Real agent execution (worktree, claude CLI, stream handler, TOTP gate)
 * lands in Phases 3–5.
 *
 * Module Lifecycle (ADR-0025): exports init / shutdown / health.
 * Manual tick trigger for tests:
 *   POST /api/v3/admin/agent-run-dispatcher/tick   (admin-only).
 *
 * Feature flag: AGENT_RUN_DISPATCHER_ENABLED. Defaults to false in
 * Phase 2 — must be flipped to 'true' explicitly per environment.
 *
 * @see ADR-0030 §3.2 (_workflow_config), §3.3 (tick), §3.4 (run_* cols).
 */

import { dbAll, dbGet, dbRun } from '../../database/connection.js';
import { apiLogger } from '../../utils/logger.js';
import { createWorkspace, destroyWorkspace } from './workspace-manager.js';
import { buildRunPrompt } from './build-run-prompt.mjs';
import { runStreamHandler, eventTranslator, warnLegacyOnce } from './run-stream-handler.mjs';
import {
  STATES,
  SIDE_EFFECTS,
  INITIAL,
  transition,
} from './state-machine.js';
import {
  generateApprovalCode,
  persistApprovalRequest,
  awaitApproval,
  APPROVAL_OUTCOMES,
  APPROVAL_CONSTANTS,
} from './approval-gate.js';
import { classify as classifySecondarySignals, VERDICT } from './secondary-signals.js';
import { isCompletionIntent as evtIsCompletionIntent } from './completion-intent.js';

const log = apiLogger.child({ module: 'agent_run_dispatcher' });

// Phase env flag (ADR-0030 Phase 4):
//   'dryrun'         → Phase 2 behavior (default, safe). Cancel immediately
//                      with run_terminal_reason='phase2_dryrun'.
//   'workspace_only' → Phase 3 behavior. Materialize git worktree, store
//                      path on ticket, then cancel with reason
//                      'phase3_workspace_only'. Still no claude CLI.
//   'live'           → Phase 4 behavior. Materialize worktree, build prompt
//                      from ticket+agent, spawn `claude --print` via the
//                      runner script, stream NDJSON, then transition to
//                      run_state='succeeded'|'failed' (NOT terminal ticket
//                      state). Workspace destroyed in finally.
// Read once at module load. Operators flip via env + restart.
const RUN_DISPATCHER_PHASE = (() => {
  const v = (process.env.RUN_DISPATCHER_PHASE || 'dryrun').toLowerCase();
  if (v !== 'dryrun' && v !== 'workspace_only' && v !== 'live') {
    // Unknown phase value → fall back to safest (dryrun) and warn at init.
    return 'dryrun';
  }
  return v;
})();

// Phase 5 env flag: gate live runs behind a TOTP approval step.
//   'true'  → generate 6-digit code, post to ticket chat, block until
//             approve/deny/expire. DEFAULT in 'live' phase (security-critical).
//   'false' → skip the gate entirely (P4 behavior). Used by smoke tests and
//             one-off operator overrides only.
// We intentionally default to 'true' in 'live' phase; in dryrun/workspace_only
// the gate is irrelevant (we never reach processLive).
const RUN_REQUIRE_APPROVAL = (() => {
  const v = (process.env.RUN_REQUIRE_APPROVAL || '').toLowerCase();
  if (v === 'false' || v === '0' || v === 'no') return false;
  if (v === 'true' || v === '1' || v === 'yes') return true;
  // Default: ON when phase is 'live'.
  return (process.env.RUN_DISPATCHER_PHASE || 'dryrun').toLowerCase() === 'live';
})();

// Audit-log rate limit for noisy stream events (per-ticket).
const AUDIT_RATE_LIMIT_MS = 5_000;

// ─── Constants (table IDs are stable per CLAUDE.md) ────────────
const TICKETS_TABLE_ID = 1708;
const WORKFLOW_CONFIG_TABLE_ID = 100000;
// _workflow_config is a singleton table — id is autoincrement, fetch the
// only row by table_id rather than guessing an id.

const FIRST_TICK_DELAY_MS = 5_000;
const FALLBACK_POLL_INTERVAL_MS = 30_000;
const CONFIG_RELOAD_INTERVAL_MS = 5 * 60 * 1000;

// Terminal `state` values from pipeline-config.js Space 11 — tickets in
// these states MUST NOT be claimed by the run loop ("not closed" filter
// per ADR-0030 §3.3).
const TICKET_TERMINAL_STATES = ['24278', '24277', '43438', '43437'];
//                              done    review  rejected control

// Run states eligible for claim (idle is the implicit default for tickets
// that never started a run — `data->>'run_state'` is NULL on those rows;
// we cover both NULL and 'idle' explicitly).
const CLAIMABLE_RUN_STATES = ['idle', 'queued', 'retry_after'];

// ADR-0030 Phase 6 — run states that are subject to stall detection. Note:
// 'awaiting_approval' is INTENTIONALLY excluded — humans are slow on purpose.
// 'streaming' is currently a synonym used by some flows for 'running'; both
// are checked so we don't miss a hung run that landed in either bucket.
const STALL_CHECK_RUN_STATES = ['preparing', 'running', 'streaming'];

// SIGKILL grace after SIGTERM during stall-kill. Mirrors the runner's own
// timeout path — keep them aligned so a hung child gets the same treatment
// regardless of which timer fires first.
const STALL_SIGKILL_GRACE_MS = 5_000;

// Default config (used only if _workflow_config row vanishes — defensive).
//
// ADR-0042 §10 (lines 296–317) is authoritative for the smart-liveness keys
// below. Flat keys at top level so they round-trip through the existing
// shallow merge in `loadConfig()`; nested objects (`tool_timeout_ms`,
// `secondary_signal_thresholds`) are JSONB blobs replaced wholesale on
// override. Operators tune via the existing `_workflow_config` MCP UI.
const DEFAULT_CONFIG = {
  poll_interval_ms: FALLBACK_POLL_INTERVAL_MS,
  max_concurrent_runs: 3,
  max_per_agent: 1,
  max_attempts: 3,
  paused: false,
  // Phase 6 stall detection knobs (legacy single-threshold path).
  stall_timeout_ms: 300_000,        // 5 min between events before a run is considered stalled (legacy alias)
  // ADR-150 P0 — meaningful-event freshness window. Replaces `stall_timeout_ms`
  // semantically; legacy key is read as a fallback during the deprecation
  // window. Default raised to 10 min: agents legitimately go quiet during
  // long Bash/WebFetch tools and the per-state FSM no longer governs this
  // path. Operators can lower via `_workflow_config`.
  meaningful_event_idle_ms: 600_000,
  retry_backoff_min_ms: 10_000,     // attempt 1 → 10s
  retry_backoff_max_ms: 40_000,     // cap so we don't push retries past 40s window

  // ─── ADR-0042 smart-liveness FSM (Task 3) ─────────────────────
  // AC8 rollback: setting `false` reverts to the legacy 30-min cap path
  // driven by `stall_timeout_ms`.
  smart_liveness_enabled: true,

  // Per-state idle thresholds (ADR §10, §State Machine table lines 62–68).
  idle_idle_ms:           300_000,        // 5 min
  thinking_idle_ms:       360_000,        // 6 min
  closing_grace_ms:        90_000,        // 90 s — no kill, only grace before backstop
  stuck_check_window_ms:   60_000,        // 60 s observation in stuck_check
  // Per-tool timeouts (used when state=tool_active). Lookup is
  // `config.tool_timeout_ms[toolName] ?? config.tool_timeout_ms.default`.
  // ADR §Per-Tool Timeout Map (lines 196–215).
  tool_timeout_ms: {
    Bash:        900_000,    // 15 min — long builds
    WebFetch:    300_000,    // 5 min — slow upstreams
    deep_scrape: 600_000,    // 10 min — multi-page crawls
    web_search:  180_000,    // 3 min
    Grep:         60_000,
    Glob:         60_000,
    Read:         60_000,
    Edit:         60_000,
    Write:        60_000,
    MultiEdit:    60_000,
    TodoWrite:    30_000,
    manage_plan:  30_000,
    default:     300_000,    // fallback for unlisted / MCP tools
  },

  // Secondary-signal classifier (Task 2 owns sampling/classify;
  // dispatcher reads these on stuck_check entry).
  //
  // ADR-0042 hybrid (variant C): CPU% delta-vs-baseline is the PRIMARY
  // liveness axis — sockets/children remain snapshot-only.
  //   - cpu_liveness_threshold_pct: default 1% — over threshold → ALIVE.
  //     Conservative: even a barely-alive event loop ticks above this.
  //   - cpu_alive_pct / cpu_dead_pct / require_all_dead are LEGACY keys
  //     from pre-hybrid Task 2; left in place so older config rows that
  //     reference them don't crash. Remove once §3 migration completes.
  secondary_signal_thresholds: {
    cpu_liveness_threshold_pct: 1,
    cpu_alive_pct:    0.5,        // legacy
    cpu_dead_pct:     0.05,       // legacy
    require_all_dead: true,       // legacy
  },

  // Backstop — wall-clock hard guard. Single warn entry at 75% elapsed.
  // ADR §Stream Handler Changes (lines 232–239).
  runner_backstop_ms:  4 * 60 * 60 * 1000,    // 4h
  backstop_warn_ratio: 0.75,                  // warn at 3h

  // Completion-intent predicate inputs. Locked decision (per Task 3 brief):
  // setter-tools (`update_ticket_status`, `update_table_row`,
  // `mcp__godcrm__update_table_row`) participate when their `input.status`
  // is in `completion_intent_terminal_states` — the FSM module's
  // `transition()` enforces the guard; the list here is the universe.
  completion_intent_tools: [
    'send_chat_message',
    'send_widget_message',
    'send_ticket_message',
    'ExitPlanMode',
    'EndTurn',
    'update_ticket_status',
    'update_table_row',
    'mcp__godcrm__update_table_row',
  ],
  completion_intent_terminal_states: ['Done', 'Closed', 'Resolved'],
};

// Map FSM state → the DEFAULT_CONFIG key holding its idle threshold.
// Extracted for `effectiveThresholdMs` so the lookup stays declarative.
const STATE_TIMEOUT_KEY = Object.freeze({
  idle:        'idle_idle_ms',
  thinking:    'thinking_idle_ms',
  tool_active: null,                  // resolved per-tool; falls back via DEFAULT
  closing:     'closing_grace_ms',
  stuck_check: 'stuck_check_window_ms',
});

/**
 * ADR-0042 §10 — resolve the effective idle threshold (ms) for a given
 * (state, toolName) pair. Pure function, no side effects, no I/O.
 *
 * Priority:
 *   1. If `toolName` is provided AND `config.tool_timeout_ms[toolName]` is
 *      set → return that. (Per-tool override always wins, regardless of
 *      state — this matches AC3: a long Bash run survives the 6-min
 *      thinking threshold because Bash is 15 min.)
 *   2. Else if `state` maps to a flat key (e.g. `idle_idle_ms`) and the
 *      config has a numeric value there → return that.
 *   3. Else → `config.tool_timeout_ms.default` (per-tool fallback).
 *   4. Else → DEFAULT_CONFIG values for the same lookup chain.
 *
 * Defensive: nullish/missing config never throws.
 *
 * @param {string|null|undefined} state — one of FSM STATES values
 * @param {string|null|undefined} toolName — current tool name when state=tool_active
 * @param {object|null|undefined} config — usually the merged loadConfig() result
 * @returns {number} threshold in milliseconds
 */
export function effectiveThresholdMs(state, toolName, config) {
  const cfg = config && typeof config === 'object' ? config : null;

  // 1. Per-tool override.
  if (toolName) {
    const cfgTools = cfg && cfg.tool_timeout_ms;
    if (cfgTools && typeof cfgTools === 'object') {
      const v = cfgTools[toolName];
      if (typeof v === 'number' && Number.isFinite(v) && v > 0) return v;
    }
    const defTools = DEFAULT_CONFIG.tool_timeout_ms;
    const dv = defTools[toolName];
    if (typeof dv === 'number' && Number.isFinite(dv) && dv > 0) return dv;
  }

  // 2. Per-state flat key.
  if (state) {
    const key = STATE_TIMEOUT_KEY[state];
    if (key) {
      if (cfg && typeof cfg[key] === 'number' && Number.isFinite(cfg[key]) && cfg[key] > 0) {
        return cfg[key];
      }
      const dv = DEFAULT_CONFIG[key];
      if (typeof dv === 'number' && Number.isFinite(dv) && dv > 0) return dv;
    }
  }

  // 3. Per-tool default fallback.
  const cfgTools = cfg && cfg.tool_timeout_ms;
  if (cfgTools && typeof cfgTools.default === 'number'
      && Number.isFinite(cfgTools.default) && cfgTools.default > 0) {
    return cfgTools.default;
  }

  // 4. Hard-coded ultimate fallback.
  return DEFAULT_CONFIG.tool_timeout_ms.default;
}

// ─── Module state ──────────────────────────────────────────────
let intervalHandle = null;
let firstTickTimer = null;
let isTicking = false;
let cachedConfig = null;
let configLoadedAt = 0;
let currentPollIntervalMs = FALLBACK_POLL_INTERVAL_MS;

// In-memory tracker for "currently running" attempts. In dry-run we
// transition straight to canceled within a single tick, so this stays
// at zero in practice — but the framework is in place for Phase 3+.
const _activeAttempts = new Map(); // ticket_id → { agent_id, claimedAt }

/**
 * Load the singleton _workflow_config row, with caching.
 * Re-fetches from DB at most every CONFIG_RELOAD_INTERVAL_MS.
 */
export async function loadConfig({ force = false } = {}) {
  const now = Date.now();
  if (!force && cachedConfig && now - configLoadedAt < CONFIG_RELOAD_INTERVAL_MS) {
    return cachedConfig;
  }
  try {
    const row = await dbGet(
      `SELECT data FROM table_rows WHERE table_id = $1 ORDER BY id ASC LIMIT 1`,
      [WORKFLOW_CONFIG_TABLE_ID]
    );
    if (!row?.data) {
      log.warn(
        { table_id: WORKFLOW_CONFIG_TABLE_ID },
        '_workflow_config row missing — using DEFAULT_CONFIG'
      );
      cachedConfig = { ...DEFAULT_CONFIG };
    } else {
      const parsed = typeof row.data === 'string' ? JSON.parse(row.data) : row.data;
      cachedConfig = { ...DEFAULT_CONFIG, ...parsed };
    }
    configLoadedAt = now;
    return cachedConfig;
  } catch (err) {
    log.error({ err }, 'failed to load _workflow_config — keeping previous cache');
    return cachedConfig || { ...DEFAULT_CONFIG };
  }
}

/**
 * Append a transition entry to the ticket's data.run_audit_log JSONB array.
 * Read-modify-write within the existing transaction is unnecessary here:
 * jsonb_insert keeps the operation server-side and atomic per row.
 */
async function appendAuditLog(ticketId, entry) {
  try {
    await dbRun(
      `UPDATE table_rows
          SET data = jsonb_set(
                       COALESCE(data, '{}'::jsonb),
                       '{run_audit_log}',
                       COALESCE(data->'run_audit_log', '[]'::jsonb) || $2::jsonb,
                       true
                     ),
              updated_at = NOW()
        WHERE table_id = $1 AND id = $3`,
      [TICKETS_TABLE_ID, JSON.stringify([entry]), ticketId]
    );
  } catch (err) {
    log.warn({ err, ticket_id: ticketId }, 'audit log append failed (non-blocking)');
  }
}

/**
 * ADR-0030 Phase 6 — exponential backoff for retry scheduling.
 *
 * Pure function, exported for unit-test introspection. Attempt is the
 * 1-based index of the run that just stalled (so the FIRST stall is
 * attempt=1 → minWindow). The next attempt's wait grows by powers of 2
 * but is clamped to retry_backoff_max_ms.
 *
 *   attempt=1 → min        (10000ms default)
 *   attempt=2 → min*2      (20000ms)
 *   attempt=3 → min*4      (40000ms — hits cap by default)
 *   attempt=N → min(min*2^(N-1), max)
 */
export function computeBackoff(attempt, config = DEFAULT_CONFIG) {
  const min = Number(config?.retry_backoff_min_ms) || DEFAULT_CONFIG.retry_backoff_min_ms;
  const max = Number(config?.retry_backoff_max_ms) || DEFAULT_CONFIG.retry_backoff_max_ms;
  const a = Math.max(1, Number(attempt) | 0);
  // 2^(a-1) — guard against absurd attempts saturating Math.pow.
  const exp = Math.min(a - 1, 30);
  return Math.min(min * Math.pow(2, exp), max);
}

/**
 * Probe whether a process exists. `process.kill(pid, 0)` does NOT signal —
 * it just throws if the PID is gone (ESRCH) or refused (EPERM, treated as
 * "exists but ours-not-to-touch"). Returns false for ESRCH / invalid pid,
 * true otherwise so we can decide whether to escalate to SIGKILL.
 */
function probeAlive(pid) {
  if (!pid || typeof pid !== 'number' || pid <= 0) return false;
  try {
    process.kill(pid, 0);
    return true;
  } catch (err) {
    // ESRCH = no such process. Anything else (e.g., EPERM) we treat as
    // "still around" since we can't confirm it's gone.
    if (err && err.code === 'ESRCH') return false;
    return true;
  }
}

/**
 * Send SIGTERM, then SIGKILL after grace if still alive. Best-effort —
 * each kill is wrapped so a missing PID never bubbles. Async because we
 * sleep between term and kill.
 */
async function killStalledProcess(pid, ticketId) {
  if (!probeAlive(pid)) {
    log.debug({ ticket_id: ticketId, pid }, 'stall: pid already gone (no signal needed)');
    return { signaled: false, reason: 'already_gone' };
  }
  try {
    process.kill(pid, 'SIGTERM');
    log.info({ ticket_id: ticketId, pid }, 'stall: SIGTERM sent');
  } catch (err) {
    if (err && err.code === 'ESRCH') {
      return { signaled: false, reason: 'esrch_on_term' };
    }
    log.warn({ err: err.message, pid, ticket_id: ticketId }, 'stall: SIGTERM failed');
  }
  await new Promise((r) => setTimeout(r, STALL_SIGKILL_GRACE_MS));
  if (probeAlive(pid)) {
    try {
      process.kill(pid, 'SIGKILL');
      log.warn({ ticket_id: ticketId, pid }, 'stall: SIGKILL sent (SIGTERM grace expired)');
      return { signaled: true, escalated: true };
    } catch (err) {
      if (err && err.code === 'ESRCH') {
        return { signaled: true, escalated: false };
      }
      log.warn({ err: err.message, pid, ticket_id: ticketId }, 'stall: SIGKILL failed');
    }
  }
  return { signaled: true, escalated: false };
}

// ─── ADR-0042 Task 4 — FSM persistence helpers ─────────────────
//
// All helpers are best-effort (warn + swallow on failure). The FSM is
// pure; this layer mediates between FSM `sideEffects[]` outputs and the
// JSONB columns on the ticket row. Heartbeat ordering: bump after the
// state write so a state flip is never visible without its associated
// `run_last_event_at` advance.

/**
 * Persist `data.run_liveness_state` AND bump `run_state_entered_at` to NOW().
 * Called when transition() returns a state different from the previously-
 * persisted one. The `entered_at` stamp is what the stall-detect tick
 * compares against `effectiveThresholdMs(state, tool)`, so it MUST advance
 * on every state flip — otherwise a long thinking → tool_active → thinking
 * cycle would inherit the original entered_at and trip the threshold.
 */
export async function persistStateChange(rowId, newState) {
  if (!rowId || typeof newState !== 'string') return;
  try {
    const ts = new Date().toISOString();
    await dbRun(
      `UPDATE table_rows
          SET data = data
                     || jsonb_build_object(
                          'run_liveness_state', $2::text,
                          'run_state_entered_at', $3::text
                        ),
              updated_at = NOW()
        WHERE table_id = $1 AND id = $4`,
      [TICKETS_TABLE_ID, newState, ts, rowId]
    );
  } catch (err) {
    log.warn({ err: err.message, ticket_id: rowId, newState }, 'persistStateChange failed (non-blocking)');
  }
}

/**
 * Persist `data.run_current_tool` (object | null). Set on tool_started,
 * cleared on tool_finished. Shape: `{name, tool_use_id, attempt_idx}`.
 */
export async function persistCurrentTool(rowId, toolName) {
  if (!rowId) return;
  try {
    if (toolName == null) {
      // Clear key entirely.
      await dbRun(
        `UPDATE table_rows
            SET data = data - 'run_current_tool',
                updated_at = NOW()
          WHERE table_id = $1 AND id = $2`,
        [TICKETS_TABLE_ID, rowId]
      );
    } else {
      // toolName may be the full {name,...} object or a bare string.
      const payload = typeof toolName === 'object' ? toolName : { name: String(toolName) };
      await dbRun(
        `UPDATE table_rows
            SET data = jsonb_set(data, '{run_current_tool}', $2::jsonb, true),
                updated_at = NOW()
          WHERE table_id = $1 AND id = $3`,
        [TICKETS_TABLE_ID, JSON.stringify(payload), rowId]
      );
    }
  } catch (err) {
    log.warn({ err: err.message, ticket_id: rowId }, 'persistCurrentTool failed (non-blocking)');
  }
}

/**
 * Persist a stuck-check baseline snapshot. ADR-0042 Task 5 shape:
 *   {
 *     baseline: <secondary-signals proof blob>,    // raw classify() proof
 *     prev_state: '<idle|thinking|tool_active|closing>',  // for restore
 *   }
 *
 * Called once on entry into `stuck_check` state. The dispatcher's
 * `_runStallDetect` tick re-reads this on the *next* tick — if the second
 * sample says alive, we restore prev_state; if dead, we kill; if
 * inconclusive past a window, we kill `stuck_inconclusive`.
 *
 * `entered_at` is intentionally NOT duplicated here — that lives in
 * `data.run_state_entered_at` and is set by `persistStateChange`.
 */
export async function persistStuckCheckBaseline(rowId, payload) {
  if (!rowId) return;
  try {
    const blob = payload && typeof payload === 'object'
      ? {
          baseline: payload.baseline ?? null,
          prev_state: typeof payload.prev_state === 'string' ? payload.prev_state : null,
        }
      : { baseline: null, prev_state: null };
    await dbRun(
      `UPDATE table_rows
          SET data = jsonb_set(data, '{run_stuck_check_baseline}', $2::jsonb, true),
              updated_at = NOW()
        WHERE table_id = $1 AND id = $3`,
      [TICKETS_TABLE_ID, JSON.stringify(blob), rowId]
    );
  } catch (err) {
    log.warn({ err: err.message, ticket_id: rowId }, 'persistStuckCheckBaseline failed (non-blocking)');
  }
}

/**
 * ADR-0042 Task 5 — clear the stuck-check baseline + restore prior FSM state.
 * Used when a `stuck_check` re-sample classifies the process as 'alive':
 * the ticket goes back to whatever state it came from, the entered_at clock
 * resets to NOW so the next idle-window starts fresh, and the baseline blob
 * is dropped.
 *
 * Best-effort: any failure is warned + swallowed. The next tick will simply
 * re-evaluate on stale data (worst case: another stuck_check entry).
 */
export async function restorePriorState(rowId, prevState) {
  if (!rowId || typeof prevState !== 'string' || prevState.length === 0) return;
  try {
    const ts = new Date().toISOString();
    await dbRun(
      `UPDATE table_rows
          SET data = (data
                      || jsonb_build_object(
                           'run_liveness_state', $2::text,
                           'run_state_entered_at', $3::text
                         )
                     ) - 'run_stuck_check_baseline',
              updated_at = NOW()
        WHERE table_id = $1 AND id = $4`,
      [TICKETS_TABLE_ID, prevState, ts, rowId]
    );
  } catch (err) {
    log.warn({ err: err.message, ticket_id: rowId, prevState }, 'restorePriorState failed (non-blocking)');
  }
}

/**
 * ADR-150 P0 — Bump `data.run_last_heartbeat_at` to NOW(). Distinct from
 * `bumpLastEventAt` which writes `run_last_event_at`. The split lets the
 * stall detector tell apart "process is alive" (heartbeat fresh) from
 * "agent is doing meaningful work" (event fresh). Cheap UPDATE, single
 * jsonb_set, fire-and-forget.
 *
 * BUG-FIX (ADR-150 P0): previously this bumped `run_last_event_at`, which
 * masked real stalls because the 15s timer kept the freshness clock fresh
 * even when the agent was completely silent.
 */
export async function bumpHeartbeatAt(rowId) {
  if (!rowId) return;
  try {
    const ts = new Date().toISOString();
    await dbRun(
      `UPDATE table_rows
          SET data = jsonb_set(data, '{run_last_heartbeat_at}', to_jsonb($2::text), true)
        WHERE table_id = $1 AND id = $3`,
      [TICKETS_TABLE_ID, ts, rowId]
    );
  } catch (err) {
    log.warn({ err: err.message, ticket_id: rowId }, 'bumpHeartbeatAt failed (non-blocking)');
  }
}

/**
 * ADR-0042 Task 6 — record the single backstop-warn edge for a live run.
 *
 * Fired by the runner's `onBackstopWarn` callback when wall-clock elapsed
 * crosses `backstop_ms * backstop_warn_ratio` (≈3h with the 4h default).
 * The runner already enforces per-run idempotency via a single-shot timer;
 * this helper is the DB-level idempotency net for callers that may resume
 * a long-lived run across dispatcher restarts. The conditional UPDATE
 * writes only when `data.run_backstop_warned_at` is missing/null, so a
 * second invocation is a no-op.
 *
 * Returns `true` iff the row was actually updated (i.e. this is the FIRST
 * warn for this run). Caller uses the return value to decide whether to
 * append the audit entry — we don't want a `backstop_warn` line per
 * resumed tick.
 *
 * Best-effort: any failure is warned + swallowed.
 */
export async function persistBackstopWarn(rowId, info) {
  if (!rowId || !info || typeof info !== 'object') return false;
  const at = typeof info.at === 'string' && info.at ? info.at : new Date().toISOString();
  try {
    const result = await dbRun(
      `UPDATE table_rows
          SET data = jsonb_set(
                       COALESCE(data, '{}'::jsonb),
                       '{run_backstop_warned_at}',
                       to_jsonb($2::text),
                       true
                     ),
              updated_at = NOW()
        WHERE table_id = $1 AND id = $3
          AND (data->>'run_backstop_warned_at') IS NULL`,
      [TICKETS_TABLE_ID, at, rowId]
    );
    return (result?.changes ?? 0) > 0;
  } catch (err) {
    log.warn({ err: err.message, ticket_id: rowId }, 'persistBackstopWarn failed (non-blocking)');
    return false;
  }
}

/**
 * Bump `data.run_completion_intent_at`. Fired once on the entry edge
 * into `closing` so observers can compute "time since the agent first
 * tried to wrap up" without scanning the audit log.
 */
export async function bumpCompletionIntentAt(rowId) {
  if (!rowId) return;
  try {
    const ts = new Date().toISOString();
    await dbRun(
      `UPDATE table_rows
          SET data = jsonb_set(data, '{run_completion_intent_at}', to_jsonb($2::text), true),
              updated_at = NOW()
        WHERE table_id = $1 AND id = $3`,
      [TICKETS_TABLE_ID, ts, rowId]
    );
  } catch (err) {
    log.warn({ err: err.message, ticket_id: rowId }, 'bumpCompletionIntentAt failed (non-blocking)');
  }
}

// In-memory FSM state per active row. Reset on dispatcher restart — we
// re-read `data.run_liveness_state` lazily on the first event after
// startup. Keyed by ticketId.
const _fsmState = new Map(); // rowId → { state, currentTool }

/**
 * ADR-0042 Task 4 — single entry point invoked from the streaming
 * attach handler for every NDJSON event. Translates the legacy event,
 * runs the FSM, applies side-effects.
 *
 * Pipeline:
 *   1. eventTranslator() → AnthropicEvent | null   (skip if null)
 *   2. Read prev FSM state (from in-mem cache, falling back to row).
 *   3. transition(prev, evt, config) → {state, currentTool, sideEffects}
 *   4. Apply each side effect via persistence helpers.
 *
 * Best-effort: each persistence step is independently try/caught inside
 * its helper. A bug here MUST NOT take the runner down.
 *
 * @param {number|string} rowId
 * @param {object} event — raw legacy NDJSON event from the runner
 * @param {object} [opts]
 * @param {object} [opts.config] — usually the dispatcher's loaded config
 */
export async function onMeaningfulEvent(rowId, event, opts = {}) {
  if (!rowId) return;

  const translated = eventTranslator(event);
  if (!translated) return;  // info / unknown — heartbeat-only via the runner's own onHeartbeat.

  // Pull prev state from in-mem cache. On the first event for a row we
  // seed from the last persisted value (handles dispatcher restart
  // mid-run); on subsequent events we trust the cache. Reads are
  // best-effort — on failure we fall back to INITIAL.
  let prev = _fsmState.get(rowId);
  if (!prev) {
    let dbState = STATES.IDLE;
    let dbTool = null;
    try {
      const row = await dbGet(
        `SELECT data->>'run_liveness_state' AS s,
                data->'run_current_tool'    AS t
           FROM table_rows
          WHERE table_id = $1 AND id = $2`,
        [TICKETS_TABLE_ID, rowId]
      );
      if (row?.s && typeof row.s === 'string') dbState = row.s;
      if (row?.t && typeof row.t === 'object') dbTool = row.t;
    } catch (err) {
      log.debug({ err: err.message, ticket_id: rowId }, 'onMeaningfulEvent: prev state lookup failed');
    }
    prev = { state: dbState, currentTool: dbTool };
  }

  const config = opts.config || cachedConfig || DEFAULT_CONFIG;
  const completionTools = Array.isArray(config?.completion_intent_tools)
    ? config.completion_intent_tools
    : DEFAULT_CONFIG.completion_intent_tools;
  const terminalStates = Array.isArray(config?.completion_intent_terminal_states)
    ? config.completion_intent_terminal_states
    : DEFAULT_CONFIG.completion_intent_terminal_states;

  let next;
  try {
    next = transition(prev, translated, { completionTools, terminalStates });
  } catch (err) {
    log.warn({ err: err.message, ticket_id: rowId }, 'onMeaningfulEvent: transition threw (non-blocking)');
    return;
  }

  // Update in-mem cache eagerly so successive events see the new state
  // even if a persistence write is still in flight.
  _fsmState.set(rowId, { state: next.state, currentTool: next.currentTool });

  const sideEffects = Array.isArray(next.sideEffects) ? next.sideEffects : [];

  // Persist state change FIRST, then heartbeat, so a tail observer never
  // sees a fresh event timestamp paired with the old state.
  if (next.state !== prev.state) {
    await persistStateChange(rowId, next.state);
  }

  // tool started/finished: keep `run_current_tool` in lockstep with FSM.
  if (sideEffects.includes(SIDE_EFFECTS.TOOL_STARTED) && next.currentTool) {
    await persistCurrentTool(rowId, next.currentTool);
  }
  if (sideEffects.includes(SIDE_EFFECTS.TOOL_FINISHED)) {
    await persistCurrentTool(rowId, null);
  }

  // Completion-intent edge: fire bumpCompletionIntentAt once.
  if (sideEffects.includes(SIDE_EFFECTS.BUMP_COMPLETION_INTENT)) {
    await bumpCompletionIntentAt(rowId);
  }

  // stuck_check entry → record baseline. The FSM never enters stuck_check
  // by itself in the meaningful-event path (it's the dispatcher tick that
  // promotes a stale tool_active/thinking into stuck_check); but if a
  // stuck_check transition does happen here, write a baseline blob with
  // prev_state + an initial /proc CPU sample. The dispatcher's
  // `_runStallDetect` will pass this `baseline.cpu` to `classify(pid, ...)`
  // on the next tick to compute a CPU%-delta verdict (ADR-0042 §9 hybrid).
  if (next.state === STATES.STUCK_CHECK && prev.state !== STATES.STUCK_CHECK) {
    let baseline = null;
    const entry = _activeAttempts.get(rowId);
    const pid = entry && Number.isFinite(entry.pid) ? entry.pid : null;
    if (pid) {
      try {
        // First call: pass baseline=null to capture the initial CPU
        // sample. We only care about `proof.cpu` here — the verdict
        // itself is irrelevant on entry (we're recording, not deciding).
        const r = await classifySecondarySignals(pid, null);
        if (r && r.proof && r.proof.cpu) {
          baseline = { cpu: r.proof.cpu };
        }
      } catch (err) {
        // Non-blocking: persist a null baseline if classify failed; the
        // next tick's classify(pid, null) on retry will re-capture.
        log.debug(
          { err: err.message, ticket_id: rowId },
          'stuck_check entry: classify baseline capture failed'
        );
      }
    }
    await persistStuckCheckBaseline(rowId, { baseline, prev_state: prev.state });
  }

  // ADR-150 P0: BUMP_EVENT is a meaningful-event signal — bump the
  // event-freshness clock (not the heartbeat clock).
  if (sideEffects.includes(SIDE_EFFECTS.BUMP_EVENT)) {
    await bumpLastEventAt(rowId);
  }
}

// Test-only accessor for the in-memory FSM cache.
export function _getFsmStateForTest() {
  return _fsmState;
}

/**
 * ADR-0030 Phase 6 — Tick Part A.
 *
 * Find tickets stuck in (preparing, running, streaming) where
 *   now() - run_last_event_at > config.stall_timeout_ms
 * and either retry-enqueue or terminally fail them.
 *
 * `awaiting_approval` is intentionally NOT in STALL_CHECK_RUN_STATES —
 * humans are slow on purpose; the approval gate has its own TTL.
 *
 * Steps per stalled row:
 *   1. SIGTERM/SIGKILL the tracked PID (if any).
 *   2. Best-effort destroyWorkspace.
 *   3. If run_attempt < max_attempts: → retry_after with exponential backoff.
 *      Else: terminal failed + run_terminal_reason='stall'.
 *   4. Audit log entry.
 *   5. Free the _activeAttempts slot.
 *
 * Updates `stats` in place with: stalled, stall_retried, stall_failed.
 * Returns the count of stalled rows handled.
 */
export async function runStallDetect({ source = 'manual', stats = null, config = null } = {}) {
  // Smoke-test affordance: when called directly, ensure config + stats are
  // populated. The internal tick path always passes both.
  if (!config) config = await loadConfig();
  if (!stats) stats = { stalled: 0, stall_retried: 0, stall_failed: 0, errors: 0 };
  return _runStallDetect({ source, stats, config });
}

/**
 * ADR-0042 Task 5 — kill a ticket because its FSM said it's stuck.
 *
 * Mirrors the legacy retry/terminal decision but with a parametrized reason
 * so the smart path can write `stall`, `stuck`, or `stuck_inconclusive`. Also
 * SIGTERMs the tracked PID and best-effort destroys the workspace.
 *
 * Optional `deps` for tests:
 *   - kill         — replacement for killStalledProcess
 *   - destroy      — replacement for destroyWorkspace
 *   - audit        — replacement for appendAuditLog
 *   - dbRunFn      — replacement for dbRun
 *   - activeMap    — replacement for the _activeAttempts Map
 */
async function killTicket(ticketId, {
  prevState,
  attempt,
  reason,            // 'stall' | 'stuck' | 'stuck_inconclusive'
  mode,              // ADR-150 P0: 'smart' | 'legacy' — recorded in audit
  source,
  config,
  stats,
  deps = {},
} = {}) {
  const _kill         = deps.kill         || killStalledProcess;
  const _destroy      = deps.destroy      || destroyWorkspace;
  const _audit        = deps.audit        || appendAuditLog;
  const _dbRun        = deps.dbRunFn      || dbRun;
  const _activeMap    = deps.activeMap    || _activeAttempts;
  const maxAttempts   = Number(config?.max_attempts) || DEFAULT_CONFIG.max_attempts;
  const ticketIdNum   = Number(ticketId);
  const attemptNum    = Number(attempt) || 0;

  // 1. Kill the live process if we tracked it.
  const tracked = _activeMap.get(ticketIdNum);
  if (tracked && tracked.pid) {
    try {
      await _kill(tracked.pid, ticketIdNum);
    } catch (killErr) {
      log.warn({ err: killErr.message, ticket_id: ticketIdNum }, 'stall: killStalledProcess threw (non-blocking)');
    }
  }

  // 2. Cleanup workspace — best-effort.
  try {
    await _destroy(ticketIdNum);
  } catch (wsErr) {
    log.warn({ err: wsErr.message, ticket_id: ticketIdNum }, 'stall: workspace cleanup failed (non-blocking)');
  }

  // 3. Retry vs terminal.
  const now = new Date();
  const ts = now.toISOString();
  let toState;

  if (attemptNum < maxAttempts) {
    const backoffMs = computeBackoff(attemptNum, config);
    const nextAt = new Date(now.getTime() + backoffMs).toISOString();
    toState = 'retry_after';
    try {
      await _dbRun(
        `UPDATE table_rows
            SET data = (data
                        || jsonb_build_object(
                             'run_state', 'retry_after',
                             'run_next_attempt_after', $2::text,
                             'run_last_event_at', $3::text
                           )
                       ) - 'run_stuck_check_baseline' - 'run_liveness_state' - 'run_state_entered_at' - 'run_current_tool',
                updated_at = NOW()
          WHERE table_id = $1 AND id = $4`,
        [TICKETS_TABLE_ID, nextAt, ts, ticketIdNum]
      );
    } catch (err) {
      if (stats) stats.errors = (stats.errors || 0) + 1;
      log.error({ err, ticket_id: ticketIdNum }, 'stall: failed to flip → retry_after');
      return { outcome: 'error' };
    }
    if (stats) stats.stall_retried = (stats.stall_retried || 0) + 1;
    log.info(
      { ticket_id: ticketIdNum, attempt: attemptNum, backoffMs, nextAt, source, reason },
      'stall: enqueued retry'
    );
  } else {
    toState = 'failed';
    try {
      await _dbRun(
        `UPDATE table_rows
            SET data = (data
                        || jsonb_build_object(
                             'run_state', 'failed',
                             'run_terminal_reason', $2::text,
                             'run_finished_at', $3::text,
                             'run_last_event_at', $3::text
                           )
                       ) - 'run_next_attempt_after' - 'run_stuck_check_baseline' - 'run_liveness_state' - 'run_state_entered_at' - 'run_current_tool',
                updated_at = NOW()
          WHERE table_id = $1 AND id = $4`,
        [TICKETS_TABLE_ID, reason, ts, ticketIdNum]
      );
    } catch (err) {
      if (stats) stats.errors = (stats.errors || 0) + 1;
      log.error({ err, ticket_id: ticketIdNum }, 'stall: failed to flip → failed');
      return { outcome: 'error' };
    }
    if (stats) stats.stall_failed = (stats.stall_failed || 0) + 1;
    log.warn(
      { ticket_id: ticketIdNum, attempt: attemptNum, source, reason },
      'stall: max_attempts exhausted — terminal failure'
    );
  }

  // 4. Audit. ADR-150 P0: include `mode` so observers can tell the smart
  // and legacy paths apart in the run history.
  await _audit(ticketIdNum, {
    at: ts,
    from: prevState,
    to: toState,
    attempt: attemptNum,
    reason,
    mode: mode || (config?.smart_liveness_enabled === false ? 'legacy' : 'smart'),
    source: 'tick_part_a',
    tick_source: source,
  });

  // 5. Free slot + clear in-mem FSM cache.
  _activeMap.delete(ticketIdNum);
  _fsmState.delete(ticketIdNum);

  return { outcome: toState };
}

/**
 * ADR-150 P0 §AC5 — legacy conflated-clock stall path.
 *
 * Used when `config.smart_liveness_enabled === false`. Heartbeat and
 * meaningful event share `run_last_event_at`, so the threshold check is
 * effectively a wall-clock cap. Pure-DB filter (no /proc, no FSM): any
 * ticket in (preparing|running|streaming) whose `run_last_event_at` is
 * older than the threshold gets the kill path.
 *
 * Threshold lookup chain (matches the smart path so the same
 * `_workflow_config` row works in both modes):
 *   `meaningful_event_idle_ms` → `stall_timeout_ms` (legacy alias) →
 *   default 600_000.
 *
 * Returns the count of rows handled.
 */
export async function legacyStallKillIfStuck({ source = 'manual', stats, config, deps = {} } = {}) {
  const stallTimeoutMs =
    Number(config?.meaningful_event_idle_ms) ||
    Number(config?.stall_timeout_ms) ||
    DEFAULT_CONFIG.meaningful_event_idle_ms;
  const stallStateList = STALL_CHECK_RUN_STATES.map((s) => `'${s}'`).join(',');
  const _dbAll         = deps.dbAllFn   || dbAll;

  const rows = await _dbAll(
    `
    SELECT id,
           COALESCE((data->>'run_attempt')::int, 0) AS run_attempt,
           data->>'run_state'                       AS run_state,
           (data->>'run_last_event_at')::timestamptz AS last_event_at
      FROM table_rows
     WHERE table_id = $1
       AND data->>'run_state' IN (${stallStateList})
       AND (data->>'run_last_event_at') IS NOT NULL
       AND (data->>'run_last_event_at')::timestamptz < NOW() - ($2::bigint * INTERVAL '1 millisecond')
     ORDER BY id ASC
     FOR UPDATE SKIP LOCKED
    `,
    [TICKETS_TABLE_ID, stallTimeoutMs]
  );

  if (rows.length === 0) return 0;

  log.info({ source, count: rows.length, stallTimeoutMs, mode: 'legacy' },
    'stall_detect (legacy): found stalled tickets');

  for (const row of rows) {
    if (stats) stats.stalled = (stats.stalled || 0) + 1;
    await killTicket(row.id, {
      prevState: row.run_state,
      attempt:   row.run_attempt,
      reason:    'stall',
      mode:      'legacy',
      source,
      config,
      stats,
      deps,
    });
  }
  return rows.length;
}

/**
 * ADR-150 P0 — split-timestamp stall detection tick.
 *
 * The check is a pure-DB filter on two JSONB timestamps:
 *   - `run_last_event_at`         must be older than `meaningful_event_idle_ms`
 *   - `run_completion_intent_at`  must be NULL or older than `closing_grace_ms`
 *
 * The completion-intent grace clause is the key change vs. the legacy path:
 * once the agent has fired `send_chat_message` / `update_ticket_status` →
 * Done / etc., it gets a 90 s window where no kill path can fire even if
 * `run_last_event_at` is technically over the idle threshold. This eliminates
 * the false-positive kill that used to happen during the natural quiet tail
 * of a turn.
 *
 * Threshold lookup chain (matches `legacyStallKillIfStuck` so the same
 * `_workflow_config` row works in both modes):
 *   `meaningful_event_idle_ms` → `stall_timeout_ms` (legacy alias) →
 *   default 600_000.
 *
 * AC8 rollback: `smart_liveness_enabled === false` skips this entirely and
 * delegates to `legacyStallKillIfStuck`.
 *
 * `deps` injection lets tests bypass the real DB / runner registry. Returns
 * the count of rows examined.
 */
async function _runStallDetect({ source, stats, config, deps = {} } = {}) {
  // AC8 — operator can flip back to legacy conflated-clock path. Two
  // escape hatches are wired:
  //   - per-config `smart_liveness_enabled === false` (ADR-150 P0 carry-over)
  //   - env `AGENT_LIVENESS_LEGACY=1` for runtime override without DB write
  if (
    config?.smart_liveness_enabled === false
    || process.env.AGENT_LIVENESS_LEGACY === '1'
  ) {
    return legacyStallKillIfStuck({ source, stats, config, deps });
  }

  const _dbAll         = deps.dbAllFn   || dbAll;
  const _classify      = deps.classify  || classifySecondarySignals;
  const _activeMap     = deps.activeMap || _activeAttempts;
  const _now           = typeof deps.nowMs === 'number' ? deps.nowMs : Date.now();
  const stallStateList = STALL_CHECK_RUN_STATES.map((s) => `'${s}'`).join(',');
  const meaningfulIdleMs =
    Number(config?.meaningful_event_idle_ms) ||
    Number(config?.stall_timeout_ms) ||
    DEFAULT_CONFIG.meaningful_event_idle_ms;
  const closingGraceMs =
    Number(config?.closing_grace_ms) || DEFAULT_CONFIG.closing_grace_ms;
  // ADR-0042 §10 — stuck_check observation window. Resolved through the
  // canonical `effectiveThresholdMs` so an operator override on
  // `stuck_check_window_ms` propagates without code changes.
  const stuckCheckWindowMs = effectiveThresholdMs(
    STATES.STUCK_CHECK, null, config
  );

  // ─── Sweep 1 (ADR-0042 Task 5b) — stuck_check rows with a baseline ──
  // For tickets the FSM has parked in `stuck_check`, the second-opinion
  // gate runs here: re-sample /proc, compare to the captured baseline,
  // and let the verdict drive the kill decision before the meaningful-
  // event idle clock or 4h backstop ever fire.
  //
  // We query separately from the legacy event-idle sweep because the
  // semantics differ: this sweep doesn't care whether `last_event_at` is
  // stale (the FSM has already decided the run is suspect by promoting
  // it into stuck_check); it cares about the verdict + state-window.
  const stuckRows = await _dbAll(
    `
    SELECT id,
           COALESCE((data->>'run_attempt')::int, 0)              AS run_attempt,
           data->>'run_state'                                     AS run_state,
           data->>'run_liveness_state'                            AS liveness_state,
           (data->>'run_state_entered_at')::timestamptz           AS state_entered_at,
           data->'run_stuck_check_baseline'                       AS stuck_check_baseline
      FROM table_rows
     WHERE table_id = $1
       AND data->>'run_state' IN (${stallStateList})
       AND data->>'run_liveness_state' = 'stuck_check'
     ORDER BY id ASC
     FOR UPDATE SKIP LOCKED
    `,
    [TICKETS_TABLE_ID]
  );

  let secondarySignalKills = 0;
  for (const row of stuckRows) {
    const tracked = _activeMap.get(row.id);
    const pid = tracked && Number.isFinite(tracked.pid) ? tracked.pid : null;
    const baseline = row.stuck_check_baseline
      && typeof row.stuck_check_baseline === 'object'
      ? row.stuck_check_baseline.baseline
      : null;
    const baselineCpu = baseline && typeof baseline === 'object' ? baseline.cpu : null;
    if (!pid || !baselineCpu) {
      // Missing pid (ticket re-claimed by another worker?) or missing
      // baseline blob (entry path failed quietly). Fall through to the
      // legacy event-idle sweep — it will eventually catch this row.
      continue;
    }

    let verdict;
    let reasons = [];
    try {
      const result = await _classify(pid, baselineCpu);
      verdict = result?.verdict;
      reasons = Array.isArray(result?.reasons) ? result.reasons : [];
    } catch (err) {
      log.warn(
        { err: err.message, ticket_id: row.id, pid },
        'stall_detect: classify threw — leaving row for next tick'
      );
      continue;
    }

    if (verdict === VERDICT.ALIVE) {
      // FSM will lift the row out of stuck_check on the next meaningful
      // event (`stuck_check is a hold` per state-machine.js). No-op here.
      log.debug(
        { ticket_id: row.id, pid, reasons },
        'stall_detect: stuck_check verdict=alive — keep waiting'
      );
      continue;
    }

    if (verdict === VERDICT.DEAD) {
      if (stats) stats.stalled = (stats.stalled || 0) + 1;
      await killTicket(row.id, {
        prevState: row.run_state,
        attempt:   row.run_attempt,
        reason:    'secondary_signals_dead',
        mode:      'smart',
        source,
        config,
        stats,
        deps,
      });
      log.warn(
        { ticket_id: row.id, pid, reasons },
        'stall_detect: stuck_check verdict=dead — killed'
      );
      secondarySignalKills += 1;
      continue;
    }

    // verdict === INCONCLUSIVE — kill only if the stuck_check window
    // has elapsed. Otherwise let the next tick re-sample (the baseline
    // stays put; classify() will get a fresh delta against it).
    const enteredAtMs = row.state_entered_at
      ? new Date(row.state_entered_at).getTime()
      : null;
    const dwellMs = enteredAtMs ? (_now - enteredAtMs) : null;
    if (dwellMs !== null && dwellMs > stuckCheckWindowMs) {
      if (stats) stats.stalled = (stats.stalled || 0) + 1;
      await killTicket(row.id, {
        prevState: row.run_state,
        attempt:   row.run_attempt,
        reason:    'inconclusive_timeout',
        mode:      'smart',
        source,
        config,
        stats,
        deps,
      });
      log.warn(
        { ticket_id: row.id, pid, dwellMs, stuckCheckWindowMs, reasons },
        'stall_detect: stuck_check inconclusive past window — killed'
      );
      secondarySignalKills += 1;
      continue;
    }

    log.debug(
      { ticket_id: row.id, pid, dwellMs, stuckCheckWindowMs, reasons },
      'stall_detect: stuck_check verdict=inconclusive within window — re-sample next tick'
    );
  }

  // ─── Sweep 2 — meaningful-event idle (legacy ADR-150 P0 path) ──────
  // Catches rows that never made it into stuck_check (FSM never promoted
  // them) but whose meaningful-event clock has nevertheless elapsed.
  // This path remains the safety net and matches pre-Task-5b behavior.
  const rows = await _dbAll(
    `
    SELECT id,
           COALESCE((data->>'run_attempt')::int, 0)             AS run_attempt,
           data->>'run_state'                                    AS run_state,
           (data->>'run_last_event_at')::timestamptz             AS last_event_at,
           (data->>'run_completion_intent_at')::timestamptz      AS completion_intent_at
      FROM table_rows
     WHERE table_id = $1
       AND data->>'run_state' IN (${stallStateList})
       AND (data->>'run_last_event_at') IS NOT NULL
       AND (data->>'run_last_event_at')::timestamptz
           < NOW() - ($2::bigint * INTERVAL '1 millisecond')
       AND (
         (data->>'run_completion_intent_at') IS NULL
         OR (data->>'run_completion_intent_at')::timestamptz
            < NOW() - ($3::bigint * INTERVAL '1 millisecond')
       )
     ORDER BY id ASC
     FOR UPDATE SKIP LOCKED
    `,
    [TICKETS_TABLE_ID, meaningfulIdleMs, closingGraceMs]
  );

  if (rows.length === 0) return secondarySignalKills;

  log.info(
    { source, count: rows.length, meaningfulIdleMs, closingGraceMs, mode: 'smart' },
    'stall_detect (smart): found stalled tickets'
  );

  for (const row of rows) {
    if (stats) stats.stalled = (stats.stalled || 0) + 1;
    await killTicket(row.id, {
      prevState: row.run_state,
      attempt:   row.run_attempt,
      reason:    'stall',
      mode:      'smart',
      source,
      config,
      stats,
      deps,
    });
  }
  return rows.length + secondarySignalKills;
}

// Test-only export: lets the table-driven test drive the tick with mocked
// classify/dbAll/activeMap/now without monkey-patching the live module.
export const _runStallDetectImpl = (opts) => _runStallDetect(opts);

/**
 * Atomically claim up to `limit` tickets ready for a run.
 *
 * Strategy: PostgreSQL `FOR UPDATE SKIP LOCKED` on table_rows row-locks,
 * combined with a JSONB-merge UPDATE that flips run_state to 'preparing'
 * and bumps run_attempt. Two concurrent dispatchers will never claim the
 * same row — the second sees `SKIP LOCKED` and moves on.
 *
 * Filter:
 *   - table_id = TICKETS_TABLE_ID
 *   - run_state IS NULL OR IN CLAIMABLE_RUN_STATES
 *   - assigned_to IS NOT NULL AND assigned_to != ''
 *   - state NOT IN terminal states (don't claim closed tickets)
 *   - run_next_attempt_after IS NULL OR <= now() (retry backoff respect)
 *
 * Returns: claimed rows with id, agent_id, prev_run_state, new run_attempt.
 */
async function claimReady(limit) {
  if (limit <= 0) return [];

  const terminalList = TICKET_TERMINAL_STATES.map(s => `'${s}'`).join(',');
  const claimableList = CLAIMABLE_RUN_STATES.map(s => `'${s}'`).join(',');

  // The CTE selects + locks candidate rows; the UPDATE then flips them.
  // Doing it in one statement keeps the lock window minimal.
  //
  // Tickets must explicitly opt into the run loop by setting
  // run_state='idle' (or queued/retry_after). NULL run_state means
  // "not enrolled" and is intentionally excluded — most pre-P1 tickets
  // have no run_state field and must not be auto-claimed.
  const sql = `
    WITH candidate AS (
      SELECT id
        FROM table_rows
       WHERE table_id = $1
         AND data->>'run_state' IN (${claimableList})
         AND COALESCE(NULLIF(data->>'assigned_to', ''), NULL) IS NOT NULL
         AND COALESCE(data->>'state', '') NOT IN (${terminalList})
         AND (
              data->>'run_next_attempt_after' IS NULL
              OR (data->>'run_next_attempt_after')::timestamptz <= NOW()
             )
       ORDER BY created_at ASC, id ASC
       LIMIT $2
       FOR UPDATE SKIP LOCKED
    )
    UPDATE table_rows tr
       SET data = tr.data
                  || jsonb_build_object(
                       'run_state', 'preparing',
                       'run_started_at', COALESCE(tr.data->>'run_started_at', to_char(NOW() AT TIME ZONE 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS"Z"')),
                       'run_last_event_at', to_char(NOW() AT TIME ZONE 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS"Z"'),
                       'run_attempt', COALESCE((tr.data->>'run_attempt')::int, 0) + 1
                     ),
           updated_at = NOW()
      FROM candidate c
     WHERE tr.id = c.id
     RETURNING tr.id,
               tr.data->>'assigned_to'  AS assigned_to,
               (tr.data->>'run_attempt')::int AS run_attempt
  `;
  return await dbAll(sql, [TICKETS_TABLE_ID, limit]);
}

/**
 * Dry-run terminal: flip preparing → canceled with phase2_dryrun reason.
 * Phase 3 retains this for `RUN_DISPATCHER_PHASE='dryrun'` (default) and
 * for rollback. Phase 4 will replace the workspace path with handler spawn.
 */
async function dryRunCancel(ticketId) {
  const ts = new Date().toISOString();
  await dbRun(
    `UPDATE table_rows
        SET data = data
                   || jsonb_build_object(
                        'run_state', 'canceled',
                        'run_terminal_reason', 'phase2_dryrun',
                        'run_finished_at', $2::text,
                        'run_last_event_at', $2::text
                      ),
            updated_at = NOW()
      WHERE table_id = $1 AND id = $3`,
    [TICKETS_TABLE_ID, ts, ticketId]
  );
}

/**
 * Phase 3 terminal: workspace was materialized, store its path, then flip
 * preparing → canceled with phase3_workspace_only reason. The cancel keeps
 * the loop terminating cleanly until Phase 4 wires the claude CLI.
 */
async function phase3WorkspaceOnlyCancel(ticketId, workspacePath) {
  const ts = new Date().toISOString();
  await dbRun(
    `UPDATE table_rows
        SET data = data
                   || jsonb_build_object(
                        'run_state', 'canceled',
                        'run_terminal_reason', 'phase3_workspace_only',
                        'run_workspace_path', $4::text,
                        'run_finished_at', $2::text,
                        'run_last_event_at', $2::text
                      ),
            updated_at = NOW()
      WHERE table_id = $1 AND id = $3`,
    [TICKETS_TABLE_ID, ts, ticketId, workspacePath]
  );
}

/**
 * Phase 4 helpers.
 *
 * Flip preparing → running. We don't carry prev state back from RETURNING
 * in claimReady, so we always set 'running' unconditionally — concurrency
 * is already protected by the row-level lock.
 */
async function transitionToRunning(ticketId) {
  const ts = new Date().toISOString();
  await dbRun(
    `UPDATE table_rows
        SET data = data
                   || jsonb_build_object(
                        'run_state', 'running',
                        'run_last_event_at', $2::text
                      ),
            updated_at = NOW()
      WHERE table_id = $1 AND id = $3`,
    [TICKETS_TABLE_ID, ts, ticketId]
  );
}

/**
 * ADR-150 P0 — bump `run_last_event_at`. Fires for every meaningful NDJSON
 * event the runner emits; this is the freshness signal stall detection
 * actually reads. Cheap UPDATE: single jsonb_set, fire-and-forget.
 */
export async function bumpLastEventAt(ticketId) {
  if (!ticketId) return;
  const ts = new Date().toISOString();
  try {
    await dbRun(
      `UPDATE table_rows
          SET data = jsonb_set(data, '{run_last_event_at}', to_jsonb($2::text), true)
        WHERE table_id = $1 AND id = $3`,
      [TICKETS_TABLE_ID, ts, ticketId]
    );
  } catch (err) {
    log.warn({ err: err.message, ticket_id: ticketId }, 'bumpLastEventAt failed (non-blocking)');
  }
}

/**
 * Apply terminal Phase 4 transition based on stream handler summary.
 * Does NOT touch ticket `state` — only run_* columns. The ticket's
 * lifecycle stays under the human's control.
 */
async function applyRunTerminal(ticketId, summary) {
  const ts = new Date().toISOString();
  let runState;
  let terminalReason;
  if (summary.finalStatus === 'success') {
    runState = 'succeeded';
    terminalReason = 'completed';
  } else if (summary.finalStatus === 'timeout') {
    runState = 'failed';
    // ADR-150 P0: stream handler now emits terminalReason='backstop' on
    // wall-clock kill. Honor it; otherwise keep the historical 'timeout'.
    terminalReason = summary.terminalReason || 'timeout';
  } else {
    runState = 'failed';
    terminalReason = summary.terminalReason || 'runner_failed';
  }
  const updates = {
    run_state: runState,
    run_terminal_reason: terminalReason,
    run_finished_at: ts,
    run_last_event_at: ts,
    run_duration_ms: summary.durationMs,
    run_event_count: summary.eventCount,
  };
  if (summary.lastError) {
    // Truncate to 500 chars per brief.
    updates.run_terminal_error = String(summary.lastError).slice(-500);
  }
  if (summary.exitCode != null) {
    updates.run_exit_code = summary.exitCode;
  }
  await dbRun(
    `UPDATE table_rows
        SET data = data || $2::jsonb,
            updated_at = NOW()
      WHERE table_id = $1 AND id = $3`,
    [TICKETS_TABLE_ID, JSON.stringify(updates), ticketId]
  );
}

/**
 * Resolve the conversation_id bound to a given ticket. Returns null if no
 * conversation is attached — caller should silently skip the chat post.
 *
 * Two strategies, in order:
 *   1. ticket.data.chat_conversation_id (explicit denorm — preferred).
 *   2. SELECT id FROM conversations WHERE bound_table_id=1708 AND bound_row_id=<ticketId>
 *      ORDER BY id ASC LIMIT 1 (fallback — works for tickets that had a
 *      chat created via the standard binding path).
 */
async function findTicketConversationId(ticketId, ticketRow) {
  const explicit = ticketRow?.chat_conversation_id;
  if (explicit) {
    const n = Number(explicit);
    if (Number.isInteger(n) && n > 0) return n;
  }
  try {
    const row = await dbGet(
      `SELECT id FROM conversations
        WHERE bound_table_id = $1 AND bound_row_id = $2
        ORDER BY id ASC LIMIT 1`,
      [TICKETS_TABLE_ID, ticketId]
    );
    if (row?.id) return Number(row.id);
  } catch (err) {
    log.debug({ err: err.message, ticket_id: ticketId }, 'conversation lookup failed');
  }
  return null;
}

/**
 * Post a system-style approval message to the ticket's chat. Sender is
 * 'system'/null user. Returns true if posted, false if no conversation.
 *
 * IMPORTANT: caller passes plaintext code — this function sends it once
 * to the chat (the only acceptable destination) and never persists or
 * logs it elsewhere. The redaction marker `[approval-code]` lets a future
 * hook scrub it from chat history if needed.
 */
async function postApprovalCodeToChat({ ticketId, ticketRow, code, expiresAt }) {
  const conversationId = await findTicketConversationId(ticketId, ticketRow);
  if (!conversationId) {
    log.warn(
      { ticket_id: ticketId },
      'no conversation bound to ticket — approval code cannot be posted to chat'
    );
    return false;
  }
  const expiresMin = Math.round(APPROVAL_CONSTANTS.TTL_MS / 60_000);
  const content =
    `🔐 Approval required for run on ticket T-${ticketId}.\n\n` +
    `[approval-code]\`${code}\`[/approval-code]\n\n` +
    `Expires in ${expiresMin} min. Submit:\n` +
    `\`POST /api/v3/admin/agent-run-dispatcher/approve/${ticketId}\` ` +
    `with body \`{"code":"${code}"}\`. ` +
    `Expires at ${expiresAt}.`;
  try {
    await dbRun(
      `INSERT INTO messages
         (conversation_id, sender_id, sender_type, role, content, content_type, agent_id, created_at, updated_at)
       VALUES ($1, NULL, 'system', 'system', $2, 'markdown', NULL, NOW(), NOW())`,
      [conversationId, content]
    );
    log.info(
      { ticket_id: ticketId, conversation_id: conversationId, code: '<redacted>' },
      'approval code posted to chat'
    );
    return true;
  } catch (err) {
    log.warn(
      { err: err.message, ticket_id: ticketId },
      'failed to post approval code to chat'
    );
    return false;
  }
}

/**
 * Post the agent's text output as a single message into the ticket chat.
 * Best-effort — never throws to caller. If conversation doesn't exist, log
 * + skip (per Phase 4 brief: must not crash on missing chat).
 */
async function postRunOutputToChat({ ticketId, ticketRow, agentId, content }) {
  if (!content || !String(content).trim()) {
    log.debug({ ticket_id: ticketId }, 'no content to post to chat — skipping');
    return;
  }
  const conversationId = await findTicketConversationId(ticketId, ticketRow);
  if (!conversationId) {
    log.warn({ ticket_id: ticketId }, 'no conversation bound to ticket — skipping chat post');
    return;
  }
  try {
    await dbRun(
      `INSERT INTO messages
         (conversation_id, sender_id, sender_type, role, content, content_type, agent_id, created_at, updated_at)
       VALUES ($1, NULL, 'agent', 'assistant', $2, 'markdown', $3, NOW(), NOW())`,
      [conversationId, String(content), agentId != null ? Number(agentId) || null : null]
    );
    log.info({ ticket_id: ticketId, conversation_id: conversationId, content_len: content.length }, 'posted run output to chat');
  } catch (err) {
    log.warn({ err: err.message, ticket_id: ticketId }, 'failed to post run output to chat (non-blocking)');
  }
}

/**
 * Phase 4 live execution path. Runs a single ticket end-to-end:
 *   createWorkspace → preparing→running → buildPrompt → runStreamHandler →
 *   apply terminal → post chat → destroyWorkspace.
 *
 * Always destroys workspace + decrements active map in finally so a crash
 * never leaks state. Returns one of: 'succeeded', 'failed', 'timeout',
 * 'workspace_create_failed'.
 */
async function processLive({ row, source, stats }) {
  let ws = null;
  let outputContent = '';
  let agentId = row.assigned_to;
  try {
    // 1. Workspace.
    ws = await createWorkspace(row.id);
    stats.workspaces_created++;
    _activeAttempts.set(row.id, {
      agent_id: agentId,
      claimedAt: Date.now(),
      workspacePath: ws.path,
    });
    log.info(
      { ticket_id: row.id, workspace_path: ws.path, branch: ws.branch, reused: !!ws.reused },
      'phase4: workspace materialized'
    );

    // Persist workspace path on the ticket so observers can see it.
    await dbRun(
      `UPDATE table_rows
          SET data = jsonb_set(data, '{run_workspace_path}', to_jsonb($2::text), true),
              updated_at = NOW()
        WHERE table_id = $1 AND id = $3`,
      [TICKETS_TABLE_ID, ws.path, row.id]
    );

    // 2. Build prompt.
    const { prompt, ticketRow } = await buildRunPrompt({
      ticketId: row.id,
      agentId,
    });

    // 2.5 Approval gate (Phase 5). Default ON in 'live' phase. Blocks here
    // until human approves via POST /admin/agent-run-dispatcher/approve/:id
    // (or hits 10-min TTL / 5 wrong attempts). Skipped entirely when
    // RUN_REQUIRE_APPROVAL=false (smoke tests + opt-out).
    if (RUN_REQUIRE_APPROVAL) {
      const { code, code_hash, expires_at, generated_at } = generateApprovalCode();
      await persistApprovalRequest(row.id, { code_hash, expires_at, generated_at });
      stats.awaiting_approval++;
      await appendAuditLog(row.id, {
        at: new Date().toISOString(),
        from: 'preparing',
        to: 'awaiting_approval',
        attempt: row.run_attempt,
        reason: 'approval_required',
        source,
        expires_at,
        // code/code_hash intentionally omitted — never logged.
      });
      // Post the code to chat. The plaintext `code` lives only in this
      // local frame; we drop the reference immediately after.
      await postApprovalCodeToChat({
        ticketId: row.id,
        ticketRow,
        code,
        expiresAt: expires_at,
      });

      const { outcome, attempts } = await awaitApproval(row.id);
      if (outcome === APPROVAL_OUTCOMES.APPROVED) {
        stats.approval_approved++;
        await appendAuditLog(row.id, {
          at: new Date().toISOString(),
          from: 'awaiting_approval',
          to: 'preparing',
          attempt: row.run_attempt,
          reason: 'approval_granted',
          source,
          attempts,
        });
        // resolveApproval already flipped run_state to 'preparing'.
      } else {
        // denied or expired: terminal failure, skip runStreamHandler.
        const terminalReason = outcome === APPROVAL_OUTCOMES.DENIED
          ? 'approval_denied'
          : 'approval_timeout';
        stats[outcome === APPROVAL_OUTCOMES.DENIED ? 'approval_denied' : 'approval_expired']++;
        await applyRunTerminal(row.id, {
          finalStatus: 'failed',
          durationMs: 0,
          eventCount: 0,
          exitCode: null,
          lastError: terminalReason,
        });
        // applyRunTerminal sets run_terminal_reason='runner_failed' for
        // generic failed → patch it to the approval-specific reason.
        await dbRun(
          `UPDATE table_rows
              SET data = jsonb_set(data, '{run_terminal_reason}', to_jsonb($2::text), true),
                  updated_at = NOW()
            WHERE table_id = $1 AND id = $3`,
          [TICKETS_TABLE_ID, terminalReason, row.id]
        );
        await appendAuditLog(row.id, {
          at: new Date().toISOString(),
          from: 'awaiting_approval',
          to: 'failed',
          attempt: row.run_attempt,
          reason: terminalReason,
          source,
          attempts,
        });
        stats.live_failed++;
        return 'failed';
      }
    }

    // 3. preparing → running.
    await transitionToRunning(row.id);
    await appendAuditLog(row.id, {
      at: new Date().toISOString(),
      from: 'preparing',
      to: 'running',
      attempt: row.run_attempt,
      reason: 'live_run_started',
      source,
      workspace_path: ws.path,
    });

    // 4. Stream handler. ADR-150 P0 wiring:
    //    - onEvent           → capture output, audit (and FSM tracker for
    //                          observability — does not drive stall detection
    //                          anymore).
    //    - onMeaningfulEvent → bumpLastEventAt + (if completion-intent)
    //                          bumpCompletionIntentAt. This is the freshness
    //                          signal stall detection actually reads.
    //    - onHeartbeat       → bumpHeartbeatAt (writes run_last_heartbeat_at).
    //                          When smart_liveness is OFF, ALSO bumps
    //                          run_last_event_at (legacy conflated mode).
    //    - backstopMs        → 4h smart, 30 min legacy.
    let lastAuditAt = 0;
    warnLegacyOnce();
    const config = await loadConfig();
    const smartLivenessEnabled = config?.smart_liveness_enabled !== false;
    const terminalStates = Array.isArray(config?.completion_intent_terminal_states)
      ? config.completion_intent_terminal_states
      : DEFAULT_CONFIG.completion_intent_terminal_states;
    const closingTools = Array.isArray(config?.completion_intent_tools)
      ? config.completion_intent_tools
      : DEFAULT_CONFIG.completion_intent_tools;
    const backstopMs = smartLivenessEnabled
      ? (Number(config?.runner_backstop_ms) || DEFAULT_CONFIG.runner_backstop_ms)
      : 30 * 60 * 1000;  // legacy conflated mode reverts to 30-min cap (AC5)
    // ADR-0042 Task 6 — pass-through of the warn ratio. Out-of-range values
    // are sanitized inside `runStreamHandler`; we just forward the config
    // value verbatim and let the runner fall back to DEFAULT_BACKSTOP_WARN_RATIO.
    const backstopWarnRatio = Number(config?.backstop_warn_ratio)
      || DEFAULT_CONFIG.backstop_warn_ratio;

    const summary = await runStreamHandler({
      ticketId: row.id,
      workspacePath: ws.path,
      prompt,
      agentId,
      backstopMs,
      backstopWarnRatio,
      onSpawn: ({ pid }) => {
        // Phase 6: record the PID so a future tick's stall detector can
        // signal it. The same _activeAttempts entry was set above; we
        // mutate in place so concurrent reads stay consistent.
        const entry = _activeAttempts.get(row.id);
        if (entry) entry.pid = pid;
      },
      onEvent: (evt) => {
        // Capture model output content for chat posting.
        if (evt && evt.type === 'output' && typeof evt.content === 'string') {
          outputContent = evt.content;
        }
        // FSM tracker — observability only; stall path no longer reads
        // run_liveness_state under ADR-150 P0. Fire-and-forget.
        onMeaningfulEvent(row.id, evt).catch((err) => {
          log.debug({ err: err.message, ticket_id: row.id }, 'onMeaningfulEvent threw');
        });
        // Audit rate limit — at most 1 entry per AUDIT_RATE_LIMIT_MS per ticket.
        const now = Date.now();
        if (now - lastAuditAt >= AUDIT_RATE_LIMIT_MS) {
          lastAuditAt = now;
          appendAuditLog(row.id, {
            at: new Date().toISOString(),
            event_type: evt?.type || 'unknown',
            event_message: typeof evt?.message === 'string' ? evt.message.slice(0, 200) : undefined,
            source: 'stream_event',
          }).catch(() => {});
        }
      },
      onMeaningfulEvent: (evt) => {
        // ADR-150 P0: bump the meaningful-event freshness clock. This is
        // the ONLY place run_last_event_at moves during a live run.
        bumpLastEventAt(row.id).catch(() => {});
        // Completion-intent edge: stamp run_completion_intent_at so the
        // stall tick's grace clause can shield the closing window.
        if (evtIsCompletionIntent(evt, { closingTools, terminalStates })) {
          bumpCompletionIntentAt(row.id).catch(() => {});
        }
      },
      onHeartbeat: () => {
        // ADR-150 P0: heartbeat writes ONLY run_last_heartbeat_at. The
        // 15s timer must NOT keep run_last_event_at fresh — that was the
        // bug that masked real stalls.
        bumpHeartbeatAt(row.id).catch(() => {});
        if (!smartLivenessEnabled) {
          // Legacy conflated mode: heartbeat ALSO bumps run_last_event_at
          // so the legacy threshold check has the same shape it always did.
          bumpLastEventAt(row.id).catch(() => {});
        }
      },
      onBackstopWarn: (warn) => {
        // ADR-0042 Task 6 — single audit edge ~3h before the 4h backstop.
        // `persistBackstopWarn` returns true only when the conditional
        // UPDATE actually flipped the row (DB-level idempotency); on a
        // resumed run where the flag is already set, we skip the audit.
        persistBackstopWarn(row.id, warn)
          .then((firstWarn) => {
            if (!firstWarn) return;
            return appendAuditLog(row.id, {
              at: warn.at,
              event_type: 'backstop_warn',
              elapsed_ms: warn.elapsedMs,
              ratio: warn.ratio,
              threshold_ms: warn.thresholdMs,
              backstop_ms: warn.backstopMs,
              source: 'run_stream_handler',
            });
          })
          .catch(() => {});
      },
    });

    // 5. Terminal transition.
    await applyRunTerminal(row.id, summary);
    await appendAuditLog(row.id, {
      at: new Date().toISOString(),
      from: 'running',
      to: summary.finalStatus === 'success' ? 'succeeded' : 'failed',
      attempt: row.run_attempt,
      reason: summary.finalStatus === 'success' ? 'completed'
            : summary.finalStatus === 'timeout' ? 'timeout'
            : 'runner_failed',
      source,
      duration_ms: summary.durationMs,
      event_count: summary.eventCount,
      exit_code: summary.exitCode,
    });

    // 6. Chat post (success only — failures don't get posted to user-facing
    // chat; they live in the audit log + run_terminal_error).
    if (summary.finalStatus === 'success' && outputContent) {
      await postRunOutputToChat({
        ticketId: row.id,
        ticketRow,
        agentId,
        content: outputContent,
      });
    }

    if (summary.finalStatus === 'success') stats.live_succeeded++;
    else if (summary.finalStatus === 'timeout') stats.live_timeout++;
    else stats.live_failed++;

    return summary.finalStatus;
  } catch (err) {
    stats.errors++;
    log.error({ err, ticket_id: row.id }, 'phase4: processLive failed — flipping to failed');
    try {
      await applyRunTerminal(row.id, {
        finalStatus: 'failed',
        durationMs: 0,
        eventCount: 0,
        exitCode: null,
        lastError: err.message || String(err),
      });
      await appendAuditLog(row.id, {
        at: new Date().toISOString(),
        from: 'preparing',
        to: 'failed',
        attempt: row.run_attempt,
        reason: 'live_path_exception',
        source,
        error: err.message,
      });
    } catch (innerErr) {
      log.error({ err: innerErr, ticket_id: row.id }, 'phase4: failed to record terminal failure');
    }
    stats.live_failed++;
    return 'failed';
  } finally {
    // Always cleanup workspace + active map, even on failure paths.
    if (ws) {
      try {
        await destroyWorkspace(row.id);
      } catch (cleanupErr) {
        log.warn({ err: cleanupErr.message, ticket_id: row.id }, 'phase4: workspace cleanup failed');
      }
    }
    _activeAttempts.delete(row.id);
    _fsmState.delete(row.id);
  }
}

/**
 * One full tick. Safe to call manually from admin route.
 */
export async function runTick({ source = 'interval' } = {}) {
  if (isTicking) {
    log.debug({ source }, 'tick already in progress; skipping');
    return { skipped: true, reason: 'already_ticking' };
  }
  isTicking = true;
  const startedAt = Date.now();
  const stats = {
    picked: 0,
    transitioned: 0,
    canceled_dryrun: 0,
    canceled_workspace_only: 0,
    workspaces_created: 0,
    live_succeeded: 0,
    live_failed: 0,
    live_timeout: 0,
    awaiting_approval: 0,
    approval_approved: 0,
    approval_denied: 0,
    approval_expired: 0,
    // Phase 6 stall detection counters.
    stalled: 0,
    stall_retried: 0,
    stall_failed: 0,
    errors: 0,
    paused: false,
  };

  try {
    const config = await loadConfig();

    // Update poll interval if config changed (will take effect next tick).
    if (typeof config.poll_interval_ms === 'number' && config.poll_interval_ms !== currentPollIntervalMs) {
      log.info(
        { old_ms: currentPollIntervalMs, new_ms: config.poll_interval_ms },
        'poll_interval_ms changed — re-arming interval'
      );
      currentPollIntervalMs = config.poll_interval_ms;
      _rearmInterval();
    }

    if (config.paused === true) {
      stats.paused = true;
      log.debug({ source }, 'config.paused=true — skipping tick');
      return { ...stats, duration_ms: Date.now() - startedAt };
    }

    // ─── Part A: stall detection (Phase 6) ─────────────────────
    // Find tickets stuck in (preparing|running|streaming) past
    // stall_timeout_ms with no recent run_last_event_at, kill any tracked
    // PID, retry with exponential backoff, or terminally fail when
    // attempts are exhausted. awaiting_approval is excluded by design.
    try {
      await _runStallDetect({ source, stats, config });
    } catch (err) {
      stats.errors++;
      log.error({ err }, 'tick: stall_detect failed');
    }

    // ─── Part B: claim ready work ──────────────────────────────
    const headroom = Math.max(0, (config.max_concurrent_runs || 3) - _activeAttempts.size);
    if (headroom === 0) {
      log.debug({ active: _activeAttempts.size }, 'no headroom — skipping claim');
      return { ...stats, duration_ms: Date.now() - startedAt };
    }

    const claimed = await claimReady(headroom);
    stats.picked = claimed.length;

    for (const row of claimed) {
      try {
        // Audit: idle/queued/retry_after → preparing
        await appendAuditLog(row.id, {
          at: new Date().toISOString(),
          from: '<previous>',  // we don't carry it back from RETURNING; cheap to omit
          to: 'preparing',
          attempt: row.run_attempt,
          reason: 'claimed_by_dispatcher',
          source,
        });
        stats.transitioned++;

        if (RUN_DISPATCHER_PHASE === 'live') {
          // Phase 4: full pipeline — workspace + prompt + claude + chat post.
          // processLive owns its own audit + cleanup; we just await it here.
          await processLive({ row, source, stats });
          continue;
        }

        if (RUN_DISPATCHER_PHASE === 'workspace_only') {
          // Phase 3: materialize git worktree, store path, then cancel.
          let ws;
          try {
            ws = await createWorkspace(row.id);
            stats.workspaces_created++;
            // Track in active map so shutdown() can clean it up.
            _activeAttempts.set(row.id, {
              agent_id: row.assigned_to,
              claimedAt: Date.now(),
              workspacePath: ws.path,
            });
            log.info(
              { ticket_id: row.id, workspace_path: ws.path, branch: ws.branch, reused: !!ws.reused },
              'phase3: workspace materialized'
            );
          } catch (wsErr) {
            stats.errors++;
            log.error({ err: wsErr, ticket_id: row.id }, 'phase3: workspace creation failed — falling back to dryrun cancel');
            await dryRunCancel(row.id);
            await appendAuditLog(row.id, {
              at: new Date().toISOString(),
              from: 'preparing',
              to: 'canceled',
              attempt: row.run_attempt,
              reason: 'workspace_create_failed',
              source,
              error: wsErr.message,
            });
            stats.canceled_dryrun++;
            continue;
          }

          await phase3WorkspaceOnlyCancel(row.id, ws.path);
          await appendAuditLog(row.id, {
            at: new Date().toISOString(),
            from: 'preparing',
            to: 'canceled',
            attempt: row.run_attempt,
            reason: 'phase3_workspace_only',
            source,
            workspace_path: ws.path,
            branch: ws.branch,
          });
          stats.canceled_workspace_only++;
          // Workspace is intentionally LEFT on disk for inspection/debug.
          // Phase 4 will own its lifecycle (destroy after handler exits).
          // shutdown() still cleans up _activeAttempts entries on graceful stop.
        } else {
          // Phase 2 dry-run: immediately cancel (default).
          await dryRunCancel(row.id);
          await appendAuditLog(row.id, {
            at: new Date().toISOString(),
            from: 'preparing',
            to: 'canceled',
            attempt: row.run_attempt,
            reason: 'phase2_dryrun',
            source,
          });
          stats.canceled_dryrun++;
        }
      } catch (err) {
        stats.errors++;
        log.error({ err, ticket_id: row.id }, 'tick: error processing claimed ticket');
      }
    }
  } catch (err) {
    stats.errors++;
    log.error({ err }, 'tick: unhandled error');
  } finally {
    isTicking = false;
  }

  const duration_ms = Date.now() - startedAt;
  if (stats.picked > 0 || stats.errors > 0) {
    log.info({ ...stats, duration_ms, source }, 'agent_run_dispatcher tick complete');
  } else {
    log.debug({ ...stats, duration_ms, source }, 'agent_run_dispatcher tick (idle)');
  }
  return { ...stats, duration_ms };
}

function _rearmInterval() {
  if (intervalHandle) {
    clearInterval(intervalHandle);
    intervalHandle = setInterval(() => {
      runTick({ source: 'interval' }).catch((err) => log.error({ err }, 'periodic tick failed'));
    }, currentPollIntervalMs);
    intervalHandle.unref?.();
  }
}

// ─── Module Lifecycle ──────────────────────────────────────────

export async function init() {
  if (intervalHandle) return health();

  if (process.env.AGENT_RUN_DISPATCHER_ENABLED !== 'true') {
    log.info(
      { flag: 'AGENT_RUN_DISPATCHER_ENABLED', value: process.env.AGENT_RUN_DISPATCHER_ENABLED || '<unset>' },
      'agent_run_dispatcher disabled by feature flag — not starting'
    );
    return { ok: false, running: false, disabled: true };
  }

  // Pre-load config so the first tick uses real poll_interval_ms.
  await loadConfig({ force: true });
  currentPollIntervalMs = cachedConfig?.poll_interval_ms || FALLBACK_POLL_INTERVAL_MS;

  log.info(
    { intervalMs: currentPollIntervalMs, firstDelayMs: FIRST_TICK_DELAY_MS, phase: RUN_DISPATCHER_PHASE },
    `starting agent_run_dispatcher (RUN_DISPATCHER_PHASE=${RUN_DISPATCHER_PHASE})`
  );

  // ADR-150 P0: log resolved values of the four smart-liveness knobs once
  // at boot so operators can confirm the active configuration without
  // tailing per-tick output.
  log.info(
    {
      smart_liveness_enabled:
        cachedConfig?.smart_liveness_enabled !== false,
      meaningful_event_idle_ms:
        Number(cachedConfig?.meaningful_event_idle_ms)
        || Number(cachedConfig?.stall_timeout_ms)
        || DEFAULT_CONFIG.meaningful_event_idle_ms,
      closing_grace_ms:
        Number(cachedConfig?.closing_grace_ms) || DEFAULT_CONFIG.closing_grace_ms,
      runner_backstop_ms:
        Number(cachedConfig?.runner_backstop_ms) || DEFAULT_CONFIG.runner_backstop_ms,
    },
    'agent_run_dispatcher: ADR-150 P0 smart-liveness knobs (resolved)'
  );

  firstTickTimer = setTimeout(() => {
    runTick({ source: 'first_tick' }).catch((err) => log.error({ err }, 'first tick failed'));
  }, FIRST_TICK_DELAY_MS);

  intervalHandle = setInterval(() => {
    runTick({ source: 'interval' }).catch((err) => log.error({ err }, 'periodic tick failed'));
  }, currentPollIntervalMs);

  intervalHandle.unref?.();
  firstTickTimer.unref?.();

  return health();
}

export async function shutdown() {
  if (firstTickTimer) {
    clearTimeout(firstTickTimer);
    firstTickTimer = null;
  }
  if (intervalHandle) {
    clearInterval(intervalHandle);
    intervalHandle = null;
    log.info('agent_run_dispatcher stopped');
  }

  // Phase 3: clean up any worktrees we still consider "active" so we don't
  // leak orphaned dirs across restarts. Use Promise.allSettled — one
  // failure mustn't block the rest.
  if (_activeAttempts.size > 0) {
    const ticketIds = Array.from(_activeAttempts.keys());
    log.info({ count: ticketIds.length }, 'shutdown: cleaning up active workspaces');
    const results = await Promise.allSettled(
      ticketIds.map((id) => destroyWorkspace(id))
    );
    let cleaned = 0;
    let failed = 0;
    results.forEach((r, idx) => {
      if (r.status === 'fulfilled') cleaned++;
      else {
        failed++;
        log.warn({ ticket_id: ticketIds[idx], err: r.reason?.message }, 'shutdown: workspace cleanup failed');
      }
    });
    _activeAttempts.clear();
    log.info({ cleaned, failed }, 'shutdown: workspace cleanup complete');
  }
}

export function health() {
  // In Phase 4 every entry in _activeAttempts represents a live runner
  // currently executing (we delete on terminal). For dryrun/workspace_only
  // phases the count is a no-op signal.
  const liveAttemptsCount = RUN_DISPATCHER_PHASE === 'live' ? _activeAttempts.size : 0;
  // Phase 5: count tickets currently in awaiting_approval. Best-effort:
  // failures fall back to undefined so health endpoint never crashes.
  let awaitingApprovalCount;
  return {
    ok: !!intervalHandle,
    running: !!intervalHandle,
    isTicking,
    intervalMs: currentPollIntervalMs,
    activeAttempts: _activeAttempts.size,
    liveAttemptsCount,
    awaitingApprovalCount, // Filled async via _awaitingApprovalCount when polled; sync default undefined.
    phase: RUN_DISPATCHER_PHASE,
    requireApproval: RUN_REQUIRE_APPROVAL,
    config: cachedConfig
      ? {
          poll_interval_ms: cachedConfig.poll_interval_ms,
          max_concurrent_runs: cachedConfig.max_concurrent_runs,
          paused: cachedConfig.paused,
        }
      : null,
  };
}

/**
 * Async health flavor — adds awaitingApprovalCount. Use from admin /health
 * route when you want the live count. Sync `health()` stays cheap.
 */
export async function healthAsync() {
  const base = health();
  try {
    const row = await dbGet(
      `SELECT COUNT(*)::int AS n
         FROM table_rows
        WHERE table_id = $1 AND data->>'run_state' = 'awaiting_approval'`,
      [TICKETS_TABLE_ID]
    );
    base.awaitingApprovalCount = Number(row?.n) || 0;
  } catch (err) {
    log.debug({ err: err.message }, 'healthAsync: awaitingApprovalCount lookup failed');
    base.awaitingApprovalCount = null;
  }
  return base;
}

// Phase 6 smoke-test affordance: expose the in-memory active-attempts map
// so tests can inject a PID for the SIGTERM/SIGKILL code path. Production
// code MUST NOT mutate this directly; `processLive` owns it.
export function _getActiveAttemptsForTest() {
  return _activeAttempts;
}

// ─── T-148528 (WP-B) — user-initiated abort ──────────────────
//
// The /stop button in chat lands here (via streamController). The legacy
// `cancelJob(jobId)` path only kills `agent_jobs` workers; dispatcher runs
// never create that row, so without this hook the child claude-code keeps
// eating tokens after a user pressed Stop.
//
// Contract:
//   - `abortRun(rowId, { reason })` is idempotent. Re-entry on an already-
//     dead/missing run is a no-op that resolves `{ aborted:false }`.
//   - Process kill uses the process-group trick (`process.kill(-pid, ...)`)
//     so child shells / MCP spawns get cleaned up too. Falls back to
//     direct-pid kill if the PGID path raises ESRCH.
//   - Terminal DB writes mark `run_state='canceled'`, `run_terminal_reason
//     ='user_stop'`, `run_finished_at=now()` — independently of whether
//     applyRunTerminal eventually fires from processLive's catch block.
//   - The conversation lookup intentionally does NOT depend on the
//     `_agent_runs` table (that name was floated in ADR-0030 drafts but
//     never landed). We walk `conversations.bound_row_id` → ticket.id.

const ABORT_SIGKILL_GRACE_MS = 5_000;

async function _persistRunCanceled(rowId, reason) {
  const nowIso = new Date().toISOString();
  try {
    await dbRun(
      `UPDATE table_rows
          SET data = COALESCE(data, '{}'::jsonb)
                     || jsonb_build_object(
                          'run_state',           'canceled'::text,
                          'run_terminal_reason', $2::text,
                          'run_finished_at',     $3::text,
                          'run_last_event_at',   $3::text
                        ),
              updated_at = NOW()
        WHERE table_id = $1 AND id = $4`,
      [TICKETS_TABLE_ID, reason || 'user_stop', nowIso, rowId]
    );
  } catch (err) {
    log.warn({ err: err.message, ticket_id: rowId }, 'abortRun: persist canceled state failed (non-blocking)');
  }
  await appendAuditLog(rowId, {
    at: nowIso,
    event_type: 'aborted',
    reason: reason || 'user_stop',
    source: 'abort_run',
  });
}

/**
 * Abort an active dispatcher run for a single ticket row.
 *
 * Looks up the live PID in `_activeAttempts`. Sends SIGTERM to the
 * process group, waits up to 5s, then escalates to SIGKILL if the child
 * is still alive. Always writes `run_state='canceled'` so observers
 * (incl. the next runTick) treat this as terminal even if the kill
 * raced with a natural exit.
 *
 * @param {number} rowId - ticket row id (table 1708)
 * @param {{reason?: string}} [opts]
 * @returns {Promise<{aborted: boolean, pid?: number, reason: string, escalated?: boolean}>}
 */
export async function abortRun(rowId, { reason = 'user_stop' } = {}) {
  if (!rowId || typeof rowId !== 'number') {
    return { aborted: false, reason: 'invalid_row_id' };
  }
  const entry = _activeAttempts.get(rowId);
  const pid = entry?.pid;

  // Persist terminal state regardless — if a stale `_activeAttempts` entry
  // is missing the PID, we still flip the ticket so future ticks don't
  // re-pick it up.
  await _persistRunCanceled(rowId, reason);

  // Drop the active-attempts entry up front so concurrent ticks don't
  // treat this run as still live.
  _activeAttempts.delete(rowId);

  if (!pid || typeof pid !== 'number' || pid <= 0) {
    log.info({ ticket_id: rowId, reason }, 'abortRun: no live PID — terminal state persisted, nothing to signal');
    return { aborted: true, reason, pid: null };
  }

  if (!probeAlive(pid)) {
    log.info({ ticket_id: rowId, pid, reason }, 'abortRun: pid already gone — nothing to signal');
    return { aborted: true, reason, pid, escalated: false };
  }

  // SIGTERM the process group so child shells / MCP spawns get cleaned up.
  try {
    process.kill(-pid, 'SIGTERM');
    log.info({ ticket_id: rowId, pid, reason }, 'abortRun: SIGTERM sent to process group');
  } catch (err) {
    if (err && err.code === 'ESRCH') {
      log.debug({ ticket_id: rowId, pid }, 'abortRun: PGID kill ESRCH — falling back to direct PID');
      try { process.kill(pid, 'SIGTERM'); }
      catch (innerErr) { /* already gone */ void innerErr; }
    } else {
      log.warn({ err: err.message, ticket_id: rowId, pid }, 'abortRun: SIGTERM failed (non-fatal)');
    }
  }

  // SIGKILL grace timer. We do NOT await it — abortRun must return
  // quickly so the HTTP handler can release. The timer unref()s so it
  // doesn't keep the event loop alive on shutdown.
  const grace = setTimeout(() => {
    if (!probeAlive(pid)) return;
    try {
      process.kill(-pid, 'SIGKILL');
      log.warn({ ticket_id: rowId, pid }, 'abortRun: SIGKILL sent (SIGTERM grace expired)');
    } catch (err) {
      if (!err || err.code !== 'ESRCH') {
        log.warn({ err: err.message, ticket_id: rowId, pid }, 'abortRun: SIGKILL failed');
      }
    }
  }, ABORT_SIGKILL_GRACE_MS);
  grace.unref?.();

  return { aborted: true, reason, pid };
}

/**
 * Abort any active dispatcher run associated with a chat conversation.
 *
 * Walks `conversations.bound_row_id` → ticket row id, then checks whether
 * that ticket is currently in `_activeAttempts`. If yes, delegates to
 * `abortRun`. Safe to call for conversations that have no bound ticket
 * (returns `{aborted: false}` without touching anything).
 *
 * @param {number|string} conversationId
 * @param {{reason?: string}} [opts]
 * @returns {Promise<{aborted: boolean, reason: string, pid?: number|null, ticket_id?: number|null}>}
 */
export async function abortRunByConversation(conversationId, { reason = 'user_stop' } = {}) {
  if (!conversationId) {
    return { aborted: false, reason: 'invalid_conversation_id' };
  }
  let conv;
  try {
    conv = await dbGet(
      `SELECT bound_row_id, bound_table_id FROM conversations WHERE id = $1`,
      [conversationId]
    );
  } catch (err) {
    log.warn({ err: err.message, conversationId }, 'abortRunByConversation: conversation lookup failed');
    return { aborted: false, reason: 'lookup_failed' };
  }
  const ticketRowId = conv?.bound_row_id ? Number(conv.bound_row_id) : null;
  // Tolerate both bound_table_id present (canonical) and missing (older rows).
  // The lookup is the authoritative signal — `_activeAttempts` keys are ticket
  // row ids regardless of which table they live in.
  if (!ticketRowId || !_activeAttempts.has(ticketRowId)) {
    return { aborted: false, reason: 'no_active_run', ticket_id: ticketRowId };
  }
  const result = await abortRun(ticketRowId, { reason });
  return { ...result, ticket_id: ticketRowId };
}

export default {
  init,
  shutdown,
  health,
  healthAsync,
  runTick,
  loadConfig,
  runStallDetect,
  computeBackoff,
  abortRun,
  abortRunByConversation,
};