feat: Minions job queue — queue, worker, backoff, types

BullMQ-inspired Postgres-native job queue built into GBrain. No Redis.
No external dependencies. Postgres transactions replace Lua scripts.

- MinionQueue: submit, claim (FOR UPDATE SKIP LOCKED), complete/fail
  (token-fenced), atomic stall detection (CTE), delayed promotion,
  parent-child resolution, prune, stats
- MinionWorker: handler registry, lock renewal, graceful SIGTERM,
  exponential backoff with jitter, UnrecoverableError bypass
- MinionJobContext: updateProgress(), log(), isActive() for handlers
- 8-state machine: waiting/active/completed/failed/delayed/dead/
  cancelled/waiting-children

Patterns stolen from: BullMQ (lock tokens, stall detection, flows),
Sidekiq (dead set, backoff formula), Inngest (checkpoint/resume).

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

src/core/minions/backoff.ts

@@ -0,0 +1,26 @@
+/**
+ * Backoff calculation for job retries.
+ * Exponential: 2^(attempts-1) * delay, with jitter.
+ * Fixed: constant delay, with jitter.
+ * From Sidekiq's formula, with BullMQ-style jitter parameter.
+ */
+
+import type { MinionJob } from './types.ts';
+
+export function calculateBackoff(job: Pick<MinionJob, 'backoff_type' | 'backoff_delay' | 'backoff_jitter' | 'attempts_made'>): number {
+  const { backoff_type, backoff_delay, backoff_jitter, attempts_made } = job;
+
+  let delay: number;
+  if (backoff_type === 'exponential') {
+    delay = Math.pow(2, Math.max(attempts_made - 1, 0)) * backoff_delay;
+  } else {
+    delay = backoff_delay;
+  }
+
+  if (backoff_jitter > 0) {
+    const jitterRange = delay * backoff_jitter;
+    delay += Math.random() * jitterRange * 2 - jitterRange;
+  }
+
+  return Math.max(delay, 0);
+}

src/core/minions/index.ts

@@ -0,0 +1,8 @@
+export { MinionQueue } from './queue.ts';
+export { MinionWorker } from './worker.ts';
+export { calculateBackoff } from './backoff.ts';
+export { UnrecoverableError, rowToMinionJob } from './types.ts';
+export type {
+  MinionJob, MinionJobInput, MinionJobStatus, MinionJobContext,
+  MinionHandler, MinionWorkerOpts, BackoffType, ChildFailPolicy,
+} from './types.ts';

src/core/minions/queue.ts

@@ -0,0 +1,377 @@
+/**
+ * MinionQueue — Postgres-native job queue inspired by BullMQ.
+ *
+ * Usage:
+ *   const queue = new MinionQueue(engine);
+ *   const job = await queue.add('sync', { full: true });
+ *   const status = await queue.getJob(job.id);
+ *   await queue.prune({ olderThan: new Date(Date.now() - 30 * 86400000) });
+ */
+
+import type { BrainEngine } from '../engine.ts';
+import type { MinionJob, MinionJobInput, MinionJobStatus } from './types.ts';
+import { rowToMinionJob } from './types.ts';
+
+const MIGRATION_VERSION = 5;
+
+export class MinionQueue {
+  constructor(private engine: BrainEngine) {}
+
+  /** Verify minion_jobs table exists (migration v5+). Call before first operation. */
+  async ensureSchema(): Promise<void> {
+    const ver = await this.engine.getConfig('version');
+    const current = parseInt(ver || '1', 10);
+    if (current < MIGRATION_VERSION) {
+      throw new Error(
+        `minion_jobs table not found (schema version ${current}, need ${MIGRATION_VERSION}). Run 'gbrain init' to apply migrations.`
+      );
+    }
+  }
+
+  /** Submit a new job. Returns the created job record. */
+  async add(name: string, data?: Record<string, unknown>, opts?: Partial<MinionJobInput>): Promise<MinionJob> {
+    if (!name || name.trim().length === 0) {
+      throw new Error('Job name cannot be empty');
+    }
+    await this.ensureSchema();
+
+    const status: MinionJobStatus = opts?.delay ? 'delayed' : (opts?.parent_job_id ? 'waiting-children' : 'waiting');
+    const delayUntil = opts?.delay ? new Date(Date.now() + opts.delay) : null;
+
+    const rows = await this.engine.executeRaw<Record<string, unknown>>(
+      `INSERT INTO minion_jobs (name, queue, status, priority, data, max_attempts, backoff_type,
+        backoff_delay, backoff_jitter, delay_until, parent_job_id, on_child_fail)
+       VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)
+       RETURNING *`,
+      [
+        name.trim(),
+        opts?.queue ?? 'default',
+        status,
+        opts?.priority ?? 0,
+        JSON.stringify(data ?? {}),
+        opts?.max_attempts ?? 3,
+        opts?.backoff_type ?? 'exponential',
+        opts?.backoff_delay ?? 1000,
+        opts?.backoff_jitter ?? 0.2,
+        delayUntil?.toISOString() ?? null,
+        opts?.parent_job_id ?? null,
+        opts?.on_child_fail ?? 'fail_parent',
+      ]
+    );
+    return rowToMinionJob(rows[0]);
+  }
+
+  /** Get a job by ID. Returns null if not found. */
+  async getJob(id: number): Promise<MinionJob | null> {
+    const rows = await this.engine.executeRaw<Record<string, unknown>>(
+      'SELECT * FROM minion_jobs WHERE id = $1',
+      [id]
+    );
+    return rows.length > 0 ? rowToMinionJob(rows[0]) : null;
+  }
+
+  /** List jobs with optional filters. */
+  async getJobs(opts?: {
+    status?: MinionJobStatus;
+    queue?: string;
+    name?: string;
+    limit?: number;
+    offset?: number;
+  }): Promise<MinionJob[]> {
+    const conditions: string[] = [];
+    const params: unknown[] = [];
+    let idx = 1;
+
+    if (opts?.status) {
+      conditions.push(`status = $${idx++}`);
+      params.push(opts.status);
+    }
+    if (opts?.queue) {
+      conditions.push(`queue = $${idx++}`);
+      params.push(opts.queue);
+    }
+    if (opts?.name) {
+      conditions.push(`name = $${idx++}`);
+      params.push(opts.name);
+    }
+
+    const where = conditions.length > 0 ? `WHERE ${conditions.join(' AND ')}` : '';
+    const limit = opts?.limit ?? 50;
+    const offset = opts?.offset ?? 0;
+
+    const rows = await this.engine.executeRaw<Record<string, unknown>>(
+      `SELECT * FROM minion_jobs ${where} ORDER BY created_at DESC LIMIT $${idx++} OFFSET $${idx}`,
+      [...params, limit, offset]
+    );
+    return rows.map(rowToMinionJob);
+  }
+
+  /** Remove a job. Only terminal statuses can be removed. */
+  async removeJob(id: number): Promise<boolean> {
+    const rows = await this.engine.executeRaw<Record<string, unknown>>(
+      `DELETE FROM minion_jobs WHERE id = $1 AND status IN ('completed', 'dead', 'cancelled', 'failed') RETURNING id`,
+      [id]
+    );
+    return rows.length > 0;
+  }
+
+  /** Cancel a waiting, active, or delayed job. */
+  async cancelJob(id: number): Promise<MinionJob | null> {
+    const rows = await this.engine.executeRaw<Record<string, unknown>>(
+      `UPDATE minion_jobs SET status = 'cancelled', lock_token = NULL, lock_until = NULL,
+        finished_at = now(), updated_at = now()
+       WHERE id = $1 AND status IN ('waiting', 'active', 'delayed', 'waiting-children')
+       RETURNING *`,
+      [id]
+    );
+    return rows.length > 0 ? rowToMinionJob(rows[0]) : null;
+  }
+
+  /** Re-queue a failed or dead job for retry. */
+  async retryJob(id: number): Promise<MinionJob | null> {
+    const rows = await this.engine.executeRaw<Record<string, unknown>>(
+      `UPDATE minion_jobs SET status = 'waiting', error_text = NULL,
+        lock_token = NULL, lock_until = NULL, delay_until = NULL,
+        finished_at = NULL, updated_at = now()
+       WHERE id = $1 AND status IN ('failed', 'dead')
+       RETURNING *`,
+      [id]
+    );
+    return rows.length > 0 ? rowToMinionJob(rows[0]) : null;
+  }
+
+  /** Prune old jobs in terminal statuses. Returns count of deleted rows. */
+  async prune(opts?: { olderThan?: Date; status?: MinionJobStatus[] }): Promise<number> {
+    const statuses = opts?.status ?? ['completed', 'dead', 'cancelled'];
+    const olderThan = opts?.olderThan ?? new Date(Date.now() - 30 * 86400000);
+
+    const rows = await this.engine.executeRaw<{ count: string }>(
+      `WITH pruned AS (
+         DELETE FROM minion_jobs
+         WHERE status = ANY($1) AND updated_at < $2
+         RETURNING id
+       )
+       SELECT count(*)::text as count FROM pruned`,
+      [statuses, olderThan.toISOString()]
+    );
+    return parseInt(rows[0]?.count ?? '0', 10);
+  }
+
+  /** Get job statistics. */
+  async getStats(opts?: { since?: Date }): Promise<{
+    by_status: Record<string, number>;
+    by_type: Array<{ name: string; total: number; completed: number; failed: number; dead: number; avg_duration_ms: number | null }>;
+    queue_health: { waiting: number; active: number; stalled: number };
+  }> {
+    const since = opts?.since ?? new Date(Date.now() - 86400000);
+
+    // Status counts
+    const statusRows = await this.engine.executeRaw<{ status: string; count: string }>(
+      `SELECT status, count(*)::text as count FROM minion_jobs GROUP BY status`
+    );
+    const by_status: Record<string, number> = {};
+    for (const r of statusRows) by_status[r.status] = parseInt(r.count, 10);
+
+    // Type breakdown (within time window)
+    const typeRows = await this.engine.executeRaw<Record<string, unknown>>(
+      `SELECT name,
+        count(*)::text as total,
+        count(*) FILTER (WHERE status = 'completed')::text as completed,
+        count(*) FILTER (WHERE status = 'failed')::text as failed,
+        count(*) FILTER (WHERE status = 'dead')::text as dead,
+        avg(EXTRACT(EPOCH FROM (finished_at - started_at)) * 1000) FILTER (WHERE finished_at IS NOT NULL AND started_at IS NOT NULL) as avg_duration_ms
+       FROM minion_jobs WHERE created_at >= $1
+       GROUP BY name ORDER BY total DESC`,
+      [since.toISOString()]
+    );
+    const by_type = typeRows.map(r => ({
+      name: r.name as string,
+      total: parseInt(r.total as string, 10),
+      completed: parseInt(r.completed as string, 10),
+      failed: parseInt(r.failed as string, 10),
+      dead: parseInt(r.dead as string, 10),
+      avg_duration_ms: r.avg_duration_ms != null ? Math.round(r.avg_duration_ms as number) : null,
+    }));
+
+    // Queue health: stalled = active with expired lock
+    const stalledRows = await this.engine.executeRaw<{ count: string }>(
+      `SELECT count(*)::text as count FROM minion_jobs WHERE status = 'active' AND lock_until < now()`
+    );
+    const stalled = parseInt(stalledRows[0]?.count ?? '0', 10);
+
+    return {
+      by_status,
+      by_type,
+      queue_health: {
+        waiting: by_status['waiting'] ?? 0,
+        active: by_status['active'] ?? 0,
+        stalled,
+      },
+    };
+  }
+
+  /** Claim the next waiting job for a worker. Token-fenced, filters by registered names. */
+  async claim(lockToken: string, lockDurationMs: number, queue: string, registeredNames: string[]): Promise<MinionJob | null> {
+    if (registeredNames.length === 0) return null;
+
+    const rows = await this.engine.executeRaw<Record<string, unknown>>(
+      `UPDATE minion_jobs SET
+        status = 'active',
+        lock_token = $1,
+        lock_until = now() + ($2::double precision * interval '1 millisecond'),
+        attempts_started = attempts_started + 1,
+        started_at = COALESCE(started_at, now()),
+        updated_at = now()
+       WHERE id = (
+         SELECT id FROM minion_jobs
+         WHERE queue = $3 AND status = 'waiting' AND name = ANY($4)
+         ORDER BY priority ASC, created_at ASC
+         FOR UPDATE SKIP LOCKED
+         LIMIT 1
+       )
+       RETURNING *`,
+      [lockToken, lockDurationMs, queue, registeredNames]
+    );
+    return rows.length > 0 ? rowToMinionJob(rows[0]) : null;
+  }
+
+  /** Complete a job (token-fenced). Returns null if token mismatch. */
+  async completeJob(id: number, lockToken: string, result?: Record<string, unknown>): Promise<MinionJob | null> {
+    const rows = await this.engine.executeRaw<Record<string, unknown>>(
+      `UPDATE minion_jobs SET status = 'completed', result = $1,
+        finished_at = now(), lock_token = NULL, lock_until = NULL, updated_at = now()
+       WHERE id = $2 AND status = 'active' AND lock_token = $3
+       RETURNING *`,
+      [result ? JSON.stringify(result) : null, id, lockToken]
+    );
+    return rows.length > 0 ? rowToMinionJob(rows[0]) : null;
+  }
+
+  /** Fail a job (token-fenced). Sets delayed for retry or dead/failed for terminal. */
+  async failJob(
+    id: number,
+    lockToken: string,
+    errorText: string,
+    newStatus: 'delayed' | 'failed' | 'dead',
+    backoffMs?: number
+  ): Promise<MinionJob | null> {
+    const rows = await this.engine.executeRaw<Record<string, unknown>>(
+      `UPDATE minion_jobs SET
+        status = $1, error_text = $2, attempts_made = attempts_made + 1,
+        stacktrace = COALESCE(stacktrace, '[]'::jsonb) || to_jsonb($3::text),
+        delay_until = CASE WHEN $1 = 'delayed' THEN now() + ($4::double precision * interval '1 millisecond') ELSE NULL END,
+        finished_at = CASE WHEN $1 IN ('failed', 'dead') THEN now() ELSE NULL END,
+        lock_token = NULL, lock_until = NULL, updated_at = now()
+       WHERE id = $5 AND status = 'active' AND lock_token = $6
+       RETURNING *`,
+      [newStatus, errorText, errorText, backoffMs ?? 0, id, lockToken]
+    );
+    return rows.length > 0 ? rowToMinionJob(rows[0]) : null;
+  }
+
+  /** Update job progress (token-fenced). */
+  async updateProgress(id: number, lockToken: string, progress: unknown): Promise<boolean> {
+    const rows = await this.engine.executeRaw<Record<string, unknown>>(
+      `UPDATE minion_jobs SET progress = $1, updated_at = now()
+       WHERE id = $2 AND status = 'active' AND lock_token = $3
+       RETURNING id`,
+      [JSON.stringify(progress), id, lockToken]
+    );
+    return rows.length > 0;
+  }
+
+  /** Renew lock (token-fenced). Returns false if token mismatch (job was reclaimed). */
+  async renewLock(id: number, lockToken: string, lockDurationMs: number): Promise<boolean> {
+    const rows = await this.engine.executeRaw<Record<string, unknown>>(
+      `UPDATE minion_jobs SET lock_until = now() + ($1::double precision * interval '1 millisecond'), updated_at = now()
+       WHERE id = $2 AND lock_token = $3 AND status = 'active'
+       RETURNING id`,
+      [lockDurationMs, id, lockToken]
+    );
+    return rows.length > 0;
+  }
+
+  /** Promote delayed jobs whose delay_until has passed. Returns promoted jobs. */
+  async promoteDelayed(): Promise<MinionJob[]> {
+    const rows = await this.engine.executeRaw<Record<string, unknown>>(
+      `UPDATE minion_jobs SET status = 'waiting', delay_until = NULL,
+        lock_token = NULL, lock_until = NULL, updated_at = now()
+       WHERE status = 'delayed' AND delay_until <= now()
+       RETURNING *`
+    );
+    return rows.map(rowToMinionJob);
+  }
+
+  /** Detect and handle stalled jobs. Single CTE, no off-by-one. Returns affected jobs. */
+  async handleStalled(): Promise<{ requeued: MinionJob[]; dead: MinionJob[] }> {
+    const rows = await this.engine.executeRaw<Record<string, unknown> & { action: string }>(
+      `WITH stalled AS (
+        SELECT id, stalled_counter, max_stalled
+        FROM minion_jobs
+        WHERE status = 'active' AND lock_until < now()
+        FOR UPDATE SKIP LOCKED
+      ),
+      requeued AS (
+        UPDATE minion_jobs SET
+          status = 'waiting', stalled_counter = stalled_counter + 1,
+          lock_token = NULL, lock_until = NULL, updated_at = now()
+        WHERE id IN (SELECT id FROM stalled WHERE stalled_counter + 1 < max_stalled)
+        RETURNING *, 'requeued' as action
+      ),
+      dead_lettered AS (
+        UPDATE minion_jobs SET
+          status = 'dead', stalled_counter = stalled_counter + 1,
+          error_text = 'max stalled count exceeded',
+          lock_token = NULL, lock_until = NULL, finished_at = now(), updated_at = now()
+        WHERE id IN (SELECT id FROM stalled WHERE stalled_counter + 1 >= max_stalled)
+        RETURNING *, 'dead' as action
+      )
+      SELECT * FROM requeued UNION ALL SELECT * FROM dead_lettered`
+    );
+
+    const requeued: MinionJob[] = [];
+    const dead: MinionJob[] = [];
+    for (const r of rows) {
+      const job = rowToMinionJob(r);
+      if (r.action === 'requeued') requeued.push(job);
+      else dead.push(job);
+    }
+    return { requeued, dead };
+  }
+
+  /** Check if all children of a parent are done. If so, unblock parent. */
+  async resolveParent(parentId: number): Promise<MinionJob | null> {
+    const rows = await this.engine.executeRaw<Record<string, unknown>>(
+      `UPDATE minion_jobs SET status = 'waiting', updated_at = now()
+       WHERE id = $1 AND status = 'waiting-children'
+         AND NOT EXISTS (
+           SELECT 1 FROM minion_jobs
+           WHERE parent_job_id = $1
+             AND status NOT IN ('completed', 'dead', 'cancelled')
+         )
+       RETURNING *`,
+      [parentId]
+    );
+    return rows.length > 0 ? rowToMinionJob(rows[0]) : null;
+  }
+
+  /** Fail the parent when a child fails with fail_parent policy. */
+  async failParent(parentId: number, childId: number, errorText: string): Promise<MinionJob | null> {
+    const rows = await this.engine.executeRaw<Record<string, unknown>>(
+      `UPDATE minion_jobs SET status = 'failed',
+        error_text = $1, finished_at = now(), updated_at = now()
+       WHERE id = $2 AND status = 'waiting-children'
+       RETURNING *`,
+      [`child job ${childId} failed: ${errorText}`, parentId]
+    );
+    return rows.length > 0 ? rowToMinionJob(rows[0]) : null;
+  }
+
+  /** Remove a child's dependency on its parent. */
+  async removeChildDependency(childId: number): Promise<void> {
+    await this.engine.executeRaw(
+      `UPDATE minion_jobs SET parent_job_id = NULL, updated_at = now() WHERE id = $1`,
+      [childId]
+    );
+  }
+}

src/core/minions/types.ts

@@ -0,0 +1,160 @@
+/**
+ * Minions — BullMQ-inspired Postgres-native job queue for GBrain.
+ *
+ * Usage:
+ *   const queue = new MinionQueue(engine);
+ *   const job = await queue.add('sync', { full: true });
+ *
+ *   const worker = new MinionWorker(engine);
+ *   worker.register('sync', async (job) => {
+ *     await runSync(engine, job.data);
+ *     return { pages_synced: 42 };
+ *   });
+ *   await worker.start();
+ */
+
+// --- Status & Type Unions ---
+
+export type MinionJobStatus =
+  | 'waiting'
+  | 'active'
+  | 'completed'
+  | 'failed'
+  | 'delayed'
+  | 'dead'
+  | 'cancelled'
+  | 'waiting-children';
+
+export type BackoffType = 'fixed' | 'exponential';
+
+export type ChildFailPolicy = 'fail_parent' | 'remove_dep' | 'ignore' | 'continue';
+
+// --- Job Record ---
+
+export interface MinionJob {
+  id: number;
+  name: string;
+  queue: string;
+  status: MinionJobStatus;
+  priority: number;
+  data: Record<string, unknown>;
+
+  // Retry
+  max_attempts: number;
+  attempts_made: number;
+  attempts_started: number;
+  backoff_type: BackoffType;
+  backoff_delay: number;
+  backoff_jitter: number;
+
+  // Stall detection
+  stalled_counter: number;
+  max_stalled: number;
+  lock_token: string | null;
+  lock_until: Date | null;
+
+  // Scheduling
+  delay_until: Date | null;
+
+  // Dependencies
+  parent_job_id: number | null;
+  on_child_fail: ChildFailPolicy;
+
+  // Results
+  result: Record<string, unknown> | null;
+  progress: unknown | null;
+  error_text: string | null;
+  stacktrace: string[];
+
+  // Timestamps
+  created_at: Date;
+  started_at: Date | null;
+  finished_at: Date | null;
+  updated_at: Date;
+}
+
+// --- Input Types ---
+
+export interface MinionJobInput {
+  name: string;
+  data?: Record<string, unknown>;
+  queue?: string;
+  priority?: number;
+  max_attempts?: number;
+  backoff_type?: BackoffType;
+  backoff_delay?: number;
+  backoff_jitter?: number;
+  delay?: number; // ms delay before eligible
+  parent_job_id?: number;
+  on_child_fail?: ChildFailPolicy;
+}
+
+export interface MinionWorkerOpts {
+  queue?: string;
+  concurrency?: number; // default 1
+  lockDuration?: number; // ms, default 30000
+  stalledInterval?: number; // ms, default 30000
+  maxStalledCount?: number; // default 1
+  pollInterval?: number; // ms, default 5000 (for PGLite fallback)
+}
+
+// --- Job Context (passed to handlers) ---
+
+export interface MinionJobContext {
+  id: number;
+  name: string;
+  data: Record<string, unknown>;
+  attempts_made: number;
+  /** Update structured progress (not just 0-100). */
+  updateProgress(progress: unknown): Promise<void>;
+  /** Append a log message to the job's stacktrace array. */
+  log(message: string): Promise<void>;
+  /** Check if the lock is still held (for long-running jobs). */
+  isActive(): Promise<boolean>;
+}
+
+export type MinionHandler = (job: MinionJobContext) => Promise<unknown>;
+
+// --- Errors ---
+
+/** Throw this from a handler to skip all retry logic and go straight to 'dead'. */
+export class UnrecoverableError extends Error {
+  constructor(message: string) {
+    super(message);
+    this.name = 'UnrecoverableError';
+  }
+}
+
+// --- Row Mapping ---
+
+export function rowToMinionJob(row: Record<string, unknown>): MinionJob {
+  return {
+    id: row.id as number,
+    name: row.name as string,
+    queue: row.queue as string,
+    status: row.status as MinionJobStatus,
+    priority: row.priority as number,
+    data: (typeof row.data === 'string' ? JSON.parse(row.data) : row.data ?? {}) as Record<string, unknown>,
+    max_attempts: row.max_attempts as number,
+    attempts_made: row.attempts_made as number,
+    attempts_started: row.attempts_started as number,
+    backoff_type: row.backoff_type as BackoffType,
+    backoff_delay: row.backoff_delay as number,
+    backoff_jitter: row.backoff_jitter as number,
+    stalled_counter: row.stalled_counter as number,
+    max_stalled: row.max_stalled as number,
+    lock_token: (row.lock_token as string) || null,
+    lock_until: row.lock_until ? new Date(row.lock_until as string) : null,
+    delay_until: row.delay_until ? new Date(row.delay_until as string) : null,
+    parent_job_id: (row.parent_job_id as number) || null,
+    on_child_fail: row.on_child_fail as ChildFailPolicy,
+    result: row.result ? (typeof row.result === 'string' ? JSON.parse(row.result) : row.result) as Record<string, unknown> : null,
+    progress: row.progress ? (typeof row.progress === 'string' ? JSON.parse(row.progress) : row.progress) : null,
+    error_text: (row.error_text as string) || null,
+    stacktrace: row.stacktrace ? (typeof row.stacktrace === 'string' ? JSON.parse(row.stacktrace) : row.stacktrace) as string[] : [],
+    created_at: new Date(row.created_at as string),
+    started_at: row.started_at ? new Date(row.started_at as string) : null,
+    finished_at: row.finished_at ? new Date(row.finished_at as string) : null,
+    updated_at: new Date(row.updated_at as string),
+  };
+}

src/core/minions/worker.ts

@@ -0,0 +1,257 @@
+/**
+ * MinionWorker — In-process job worker with BullMQ-inspired patterns.
+ *
+ * Usage:
+ *   const worker = new MinionWorker(engine);
+ *   worker.register('sync', async (job) => { ... });
+ *   worker.register('embed', async (job) => { ... });
+ *   await worker.start(); // polls until SIGTERM
+ */
+
+import type { BrainEngine } from '../engine.ts';
+import type { MinionJob, MinionJobContext, MinionHandler, MinionWorkerOpts } from './types.ts';
+import { UnrecoverableError } from './types.ts';
+import { MinionQueue } from './queue.ts';
+import { calculateBackoff } from './backoff.ts';
+import { randomUUID } from 'crypto';
+
+export class MinionWorker {
+  private queue: MinionQueue;
+  private handlers = new Map<string, MinionHandler>();
+  private running = false;
+  private currentJob: MinionJob | null = null;
+  private lockRenewalTimer: ReturnType<typeof setInterval> | null = null;
+  private workerId = randomUUID();
+
+  private opts: Required<MinionWorkerOpts>;
+
+  constructor(
+    private engine: BrainEngine,
+    opts?: MinionWorkerOpts,
+  ) {
+    this.queue = new MinionQueue(engine);
+    this.opts = {
+      queue: opts?.queue ?? 'default',
+      concurrency: opts?.concurrency ?? 1,
+      lockDuration: opts?.lockDuration ?? 30000,
+      stalledInterval: opts?.stalledInterval ?? 30000,
+      maxStalledCount: opts?.maxStalledCount ?? 1,
+      pollInterval: opts?.pollInterval ?? 5000,
+    };
+  }
+
+  /** Register a handler for a job type. */
+  register(name: string, handler: MinionHandler): void {
+    this.handlers.set(name, handler);
+  }
+
+  /** Get registered handler names (used by claim query). */
+  get registeredNames(): string[] {
+    return Array.from(this.handlers.keys());
+  }
+
+  /** Start the worker loop. Blocks until stopped. */
+  async start(): Promise<void> {
+    if (this.handlers.size === 0) {
+      throw new Error('No handlers registered. Call worker.register(name, handler) before start().');
+    }
+
+    await this.queue.ensureSchema();
+    this.running = true;
+
+    // Graceful shutdown
+    const shutdown = () => {
+      console.log('Minion worker shutting down...');
+      this.running = false;
+    };
+    process.on('SIGTERM', shutdown);
+    process.on('SIGINT', shutdown);
+
+    // Stall detection on interval
+    const stalledTimer = setInterval(async () => {
+      try {
+        const { requeued, dead } = await this.queue.handleStalled();
+        if (requeued.length > 0) console.log(`Stall detector: requeued ${requeued.length} jobs`);
+        if (dead.length > 0) console.log(`Stall detector: dead-lettered ${dead.length} jobs`);
+      } catch (e) {
+        console.error('Stall detection error:', e instanceof Error ? e.message : String(e));
+      }
+    }, this.opts.stalledInterval);
+
+    try {
+      while (this.running) {
+        // Promote delayed jobs
+        try {
+          await this.queue.promoteDelayed();
+        } catch (e) {
+          console.error('Promotion error:', e instanceof Error ? e.message : String(e));
+        }
+
+        // Claim and execute
+        const lockToken = `${this.workerId}:${Date.now()}`;
+        const job = await this.queue.claim(
+          lockToken,
+          this.opts.lockDuration,
+          this.opts.queue,
+          this.registeredNames,
+        );
+
+        if (job) {
+          this.currentJob = job;
+          await this.executeJob(job, lockToken);
+          this.currentJob = null;
+        } else {
+          // No jobs available, poll
+          await new Promise(resolve => setTimeout(resolve, this.opts.pollInterval));
+        }
+      }
+    } finally {
+      clearInterval(stalledTimer);
+      process.removeListener('SIGTERM', shutdown);
+      process.removeListener('SIGINT', shutdown);
+
+      // Wait for current job to finish (graceful shutdown)
+      if (this.currentJob && this.lockRenewalTimer) {
+        console.log('Waiting for current job to finish (30s timeout)...');
+        await new Promise(resolve => setTimeout(resolve, 30000));
+      }
+
+      console.log('Minion worker stopped.');
+    }
+  }
+
+  /** Stop the worker gracefully. */
+  stop(): void {
+    this.running = false;
+  }
+
+  private async executeJob(job: MinionJob, lockToken: string): Promise<void> {
+    const handler = this.handlers.get(job.name);
+    if (!handler) {
+      // This shouldn't happen (claim filters by registered names), but be safe
+      await this.queue.failJob(job.id, lockToken, `No handler for job type '${job.name}'`, 'dead');
+      return;
+    }
+
+    // Start lock renewal
+    this.lockRenewalTimer = setInterval(async () => {
+      const renewed = await this.queue.renewLock(job.id, lockToken, this.opts.lockDuration);
+      if (!renewed) {
+        // Lock was stolen (stall detector reclaimed it)
+        console.warn(`Lock lost for job ${job.id}, stopping execution`);
+        this.lockRenewalTimer && clearInterval(this.lockRenewalTimer);
+        this.lockRenewalTimer = null;
+      }
+    }, this.opts.lockDuration / 2);
+
+    // Build job context
+    const context: MinionJobContext = {
+      id: job.id,
+      name: job.name,
+      data: job.data,
+      attempts_made: job.attempts_made,
+      updateProgress: async (progress: unknown) => {
+        await this.queue.updateProgress(job.id, lockToken, progress);
+      },
+      log: async (message: string) => {
+        // Append to stacktrace as a log entry
+        await this.engine.executeRaw(
+          `UPDATE minion_jobs SET stacktrace = COALESCE(stacktrace, '[]'::jsonb) || to_jsonb($1::text),
+            updated_at = now()
+           WHERE id = $2 AND status = 'active' AND lock_token = $3`,
+          [message, job.id, lockToken]
+        );
+      },
+      isActive: async () => {
+        const rows = await this.engine.executeRaw<{ id: number }>(
+          `SELECT id FROM minion_jobs WHERE id = $1 AND status = 'active' AND lock_token = $2`,
+          [job.id, lockToken]
+        );
+        return rows.length > 0;
+      },
+    };
+
+    try {
+      const result = await handler(context);
+
+      // Clear renewal timer
+      if (this.lockRenewalTimer) {
+        clearInterval(this.lockRenewalTimer);
+        this.lockRenewalTimer = null;
+      }
+
+      // Complete the job (token-fenced)
+      const completed = await this.queue.completeJob(
+        job.id,
+        lockToken,
+        result != null ? (typeof result === 'object' ? result as Record<string, unknown> : { value: result }) : undefined,
+      );
+
+      if (!completed) {
+        console.warn(`Job ${job.id} completion dropped (lock token mismatch, job was reclaimed)`);
+        return;
+      }
+
+      // Resolve parent if this is a child job
+      if (job.parent_job_id) {
+        await this.queue.resolveParent(job.parent_job_id);
+      }
+    } catch (err) {
+      // Clear renewal timer
+      if (this.lockRenewalTimer) {
+        clearInterval(this.lockRenewalTimer);
+        this.lockRenewalTimer = null;
+      }
+
+      const errorText = err instanceof Error ? err.message : String(err);
+      const isUnrecoverable = err instanceof UnrecoverableError;
+      const attemptsExhausted = job.attempts_made + 1 >= job.max_attempts;
+
+      let newStatus: 'delayed' | 'failed' | 'dead';
+      if (isUnrecoverable || attemptsExhausted) {
+        newStatus = 'dead';
+      } else {
+        newStatus = 'delayed';
+      }
+
+      const backoffMs = newStatus === 'delayed' ? calculateBackoff({
+        backoff_type: job.backoff_type,
+        backoff_delay: job.backoff_delay,
+        backoff_jitter: job.backoff_jitter,
+        attempts_made: job.attempts_made + 1,
+      }) : 0;
+
+      const failed = await this.queue.failJob(job.id, lockToken, errorText, newStatus, backoffMs);
+      if (!failed) {
+        console.warn(`Job ${job.id} failure dropped (lock token mismatch)`);
+        return;
+      }
+
+      // Handle parent-child failure policies
+      if (job.parent_job_id && (newStatus === 'dead' || newStatus === 'failed')) {
+        const parentJob = await this.queue.getJob(job.parent_job_id);
+        if (parentJob && parentJob.status === 'waiting-children') {
+          switch (job.on_child_fail) {
+            case 'fail_parent':
+              await this.queue.failParent(job.parent_job_id, job.id, errorText);
+              break;
+            case 'remove_dep':
+              await this.queue.removeChildDependency(job.id);
+              await this.queue.resolveParent(job.parent_job_id);
+              break;
+            case 'ignore':
+            case 'continue':
+              await this.queue.resolveParent(job.parent_job_id);
+              break;
+          }
+        }
+      }
+
+      if (newStatus === 'delayed') {
+        console.log(`Job ${job.id} (${job.name}) failed, retrying in ${Math.round(backoffMs)}ms (attempt ${job.attempts_made + 1}/${job.max_attempts})`);
+      } else {
+        console.log(`Job ${job.id} (${job.name}) permanently failed: ${errorText}`);
+      }
+    }
+  }
+}