feat(minions): queue v7 — depth, child cap, timeouts, cascade, idempotency, child_done

Wraps completeJob and failJob in engine.transaction() so parent hook
invocations (resolveParent, failParent, removeChildDependency) fold into
the same transaction as the child update. A process crash between child
and parent can't strand the parent in waiting-children anymore.

Adds v7 behaviors:
- Depth tracking. add() computes depth = parent.depth + 1 and rejects
  past maxSpawnDepth (default 5).
- Per-parent child cap. add() takes SELECT ... FOR UPDATE on the parent,
  counts non-terminal children, rejects when count >= max_children.
  NULL max_children = no cap.
- Per-job wall-clock timeout. claim() populates timeout_at when
  timeout_ms is set. New handleTimeouts() dead-letters expired rows with
  error_text='timeout exceeded'. Terminal — no retry.
- Cascade cancel. cancelJob() walks descendants via recursive CTE with
  depth-100 runaway cap. Returns the root row. Re-parented descendants
  (parent_job_id NULL) are naturally excluded.
- Idempotency. add() uses INSERT ... ON CONFLICT (idempotency_key) DO
  NOTHING RETURNING; falls back to SELECT when RETURNING is empty. Same
  key always yields the same job id.
- child_done inbox. completeJob inserts {type:'child_done', child_id,
  job_name, result} into the parent's inbox in the same transaction as
  the token rollup, guarded by EXISTS so terminal/deleted parents skip
  without FK violation. New readChildCompletions(parent_id, lock_token,
  since?) helper; token-fenced like readInbox.
- removeOnComplete / removeOnFail. Deletes the row after the parent hook
  fires, so parent policy sees consistent state.
- Attachment methods. addAttachment validates via validateAttachment
  then INSERTs; UNIQUE (job_id, filename) backs the JS dup check.
  listAttachments, getAttachment, deleteAttachment round out the API.

Fixes pre-existing inverted status bug: add() now puts children in
waiting/delayed (not waiting-children) and atomically flips the parent
to waiting-children in the same transaction. Tests no longer need
manual UPDATE workarounds.

Two correctness fixes:
- Sibling completion race. Under READ COMMITTED, two grandchildren
  completing concurrently each saw the other as still-active in the
  pre-commit snapshot and neither flipped the parent. Fixed by taking
  SELECT ... FOR UPDATE on the parent row at the start of completeJob
  and failJob transactions, serializing siblings on the parent lock.
- JSONB double-encode. postgres.js conn.unsafe(sql, params) auto-
  JSON-encodes parameters. Calling JSON.stringify(obj) first stored a
  JSON string literal (jsonb_typeof=string) and broke payload->>'key'
  queries silently. Removed JSON.stringify from three call sites
  (child_done inbox post, updateProgress, sendMessage). PGLite tolerated
  both forms so unit tests missed it — real-PG E2E caught it.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

src/core/minions/queue.ts

@@ -9,13 +9,28 @@
  */
 
 import type { BrainEngine } from '../engine.ts';
-import type { MinionJob, MinionJobInput, MinionJobStatus, InboxMessage, TokenUpdate } from './types.ts';
-import { rowToMinionJob, rowToInboxMessage } from './types.ts';
+import type {
+  MinionJob, MinionJobInput, MinionJobStatus, InboxMessage, TokenUpdate,
+  MinionQueueOpts, ChildDoneMessage, Attachment, AttachmentInput,
+} from './types.ts';
+import { rowToMinionJob, rowToInboxMessage, rowToAttachment } from './types.ts';
+import { validateAttachment } from './attachments.ts';
 
-const MIGRATION_VERSION = 6;
+const MIGRATION_VERSION = 7;
+
+const DEFAULT_MAX_SPAWN_DEPTH = 5;
+const DEFAULT_MAX_ATTACHMENT_BYTES = 5 * 1024 * 1024; // 5 MiB
+
+const TERMINAL_STATUSES = ['completed', 'failed', 'dead', 'cancelled'] as const;
 
 export class MinionQueue {
-  constructor(private engine: BrainEngine) {}
+  readonly maxSpawnDepth: number;
+  readonly maxAttachmentBytes: number;
+
+  constructor(private engine: BrainEngine, opts: MinionQueueOpts = {}) {
+    this.maxSpawnDepth = opts.maxSpawnDepth ?? DEFAULT_MAX_SPAWN_DEPTH;
+    this.maxAttachmentBytes = opts.maxAttachmentBytes ?? DEFAULT_MAX_ATTACHMENT_BYTES;
+  }
 
   /** Verify minion_jobs table exists (migration v5+). Call before first operation. */
   async ensureSchema(): Promise<void> {
@@ -28,25 +43,89 @@ export class MinionQueue {
     }
   }
 
-  /** Submit a new job. Returns the created job record. */
+  /**
+   * Submit a new job.
+   *
+   * Wrapped in engine.transaction(): when parent_job_id is set, takes
+   * SELECT ... FOR UPDATE on the parent so concurrent submissions serialize
+   * on the cap check. Without this, two concurrent submissions could both
+   * see count = N-1 and both insert, blowing max_children.
+   *
+   * Child status is 'waiting' (or 'delayed') — claimable. Parent is flipped
+   * to 'waiting-children' atomically. Idempotency_key dedups via PG unique
+   * partial index; same key returns the existing row (no second insert).
+   */
   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 childStatus: MinionJobStatus = opts?.delay ? 'delayed' : 'waiting';
     const delayUntil = opts?.delay ? new Date(Date.now() + opts.delay) : null;
+    const maxSpawnDepth = opts?.max_spawn_depth ?? this.maxSpawnDepth;
 
-    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 *`,
-      [
+    return this.engine.transaction(async (tx) => {
+      // 1. Idempotency fast path — if a row already exists for this key, return it
+      //    without doing any other work. The unique partial index guarantees
+      //    no second row can be inserted with the same non-null key.
+      if (opts?.idempotency_key) {
+        const existing = await tx.executeRaw<Record<string, unknown>>(
+          `SELECT * FROM minion_jobs WHERE idempotency_key = $1`,
+          [opts.idempotency_key]
+        );
+        if (existing.length > 0) return rowToMinionJob(existing[0]);
+      }
+
+      // 2. Parent lock + depth/cap validation
+      let depth = 0;
+      if (opts?.parent_job_id) {
+        const parentRows = await tx.executeRaw<Record<string, unknown>>(
+          `SELECT * FROM minion_jobs WHERE id = $1 FOR UPDATE`,
+          [opts.parent_job_id]
+        );
+        if (parentRows.length === 0) {
+          throw new Error(`parent_job_id ${opts.parent_job_id} not found`);
+        }
+        const parent = rowToMinionJob(parentRows[0]);
+
+        depth = parent.depth + 1;
+        if (depth > maxSpawnDepth) {
+          throw new Error(`spawn depth ${depth} exceeds maxSpawnDepth ${maxSpawnDepth}`);
+        }
+
+        if (parent.max_children !== null) {
+          const countRows = await tx.executeRaw<{ count: string }>(
+            `SELECT count(*)::text as count FROM minion_jobs
+             WHERE parent_job_id = $1 AND status NOT IN ('completed','failed','dead','cancelled')`,
+            [opts.parent_job_id]
+          );
+          const live = parseInt(countRows[0]?.count ?? '0', 10);
+          if (live >= parent.max_children) {
+            throw new Error(`parent ${opts.parent_job_id} already has ${live} live children (max_children=${parent.max_children})`);
+          }
+        }
+      }
+
+      // 3. Insert child. Use ON CONFLICT for idempotency; if a concurrent submit
+      //    raced past the fast-path SELECT, the unique index catches it here.
+      const insertSql = opts?.idempotency_key
+        ? `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,
+            depth, max_children, timeout_ms, remove_on_complete, remove_on_fail, idempotency_key)
+           VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18)
+           ON CONFLICT (idempotency_key) WHERE idempotency_key IS NOT NULL DO NOTHING
+           RETURNING *`
+        : `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,
+            depth, max_children, timeout_ms, remove_on_complete, remove_on_fail, idempotency_key)
+           VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18)
+           RETURNING *`;
+
+      const params = [
         name.trim(),
         opts?.queue ?? 'default',
-        status,
+        childStatus,
         opts?.priority ?? 0,
         JSON.stringify(data ?? {}),
         opts?.max_attempts ?? 3,
@@ -56,9 +135,43 @@ export class MinionQueue {
         delayUntil?.toISOString() ?? null,
         opts?.parent_job_id ?? null,
         opts?.on_child_fail ?? 'fail_parent',
-      ]
-    );
-    return rowToMinionJob(rows[0]);
+        depth,
+        opts?.max_children ?? null,
+        opts?.timeout_ms ?? null,
+        opts?.remove_on_complete ?? false,
+        opts?.remove_on_fail ?? false,
+        opts?.idempotency_key ?? null,
+      ];
+
+      const inserted = await tx.executeRaw<Record<string, unknown>>(insertSql, params);
+
+      // ON CONFLICT DO NOTHING returns 0 rows — fall back to SELECT to fetch the
+      // existing row that won the race.
+      if (inserted.length === 0 && opts?.idempotency_key) {
+        const existing = await tx.executeRaw<Record<string, unknown>>(
+          `SELECT * FROM minion_jobs WHERE idempotency_key = $1`,
+          [opts.idempotency_key]
+        );
+        if (existing.length === 0) {
+          throw new Error(`idempotency_key ${opts.idempotency_key} insert returned no row and no existing row found`);
+        }
+        return rowToMinionJob(existing[0]);
+      }
+
+      const child = rowToMinionJob(inserted[0]);
+
+      // 4. Flip parent to waiting-children if this is a fresh child insert.
+      //    Only transition from non-terminal, non-already-waiting-children states.
+      if (opts?.parent_job_id) {
+        await tx.executeRaw(
+          `UPDATE minion_jobs SET status = 'waiting-children', updated_at = now()
+           WHERE id = $1 AND status IN ('waiting','active','delayed')`,
+          [opts.parent_job_id]
+        );
+      }
+
+      return child;
+    });
   }
 
   /** Get a job by ID. Returns null if not found. */
@@ -115,16 +228,45 @@ export class MinionQueue {
     return rows.length > 0;
   }
 
-  /** Cancel a waiting, active, or delayed job. */
+  /**
+   * Cancel a job and cascade-kill all descendants in one statement.
+   *
+   * Honest scope: this is BullMQ-style best-effort cancel. The recursive CTE
+   * snapshots the parent_job_id chain at statement start. A descendant
+   * re-parented BEFORE the cancel call is excluded; one re-parented DURING
+   * the call may still get cancelled (cancel wins if seen in the snapshot).
+   * Re-parented descendants whose parent_job_id is NULL'd by
+   * removeChildDependency naturally fall out of the recursive walk.
+   *
+   * Active descendants get lock_token = NULL — same path pause uses, so the
+   * worker's renewLock will fail next tick and AbortController fires.
+   *
+   * Returns the *root* (the job matching id), not an arbitrary descendant.
+   */
   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', 'paused')
+      `WITH RECURSIVE descendants AS (
+        SELECT id, 0 AS d FROM minion_jobs WHERE id = $1
+        UNION ALL
+        SELECT m.id, descendants.d + 1
+          FROM minion_jobs m
+          JOIN descendants ON m.parent_job_id = descendants.id
+          WHERE descendants.d < 100
+      )
+      UPDATE minion_jobs SET
+        status = 'cancelled',
+        lock_token = NULL,
+        lock_until = NULL,
+        finished_at = now(),
+        updated_at = now()
+       WHERE id IN (SELECT id FROM descendants)
+         AND status IN ('waiting','active','delayed','waiting-children','paused')
        RETURNING *`,
       [id]
     );
-    return rows.length > 0 ? rowToMinionJob(rows[0]) : null;
+    if (rows.length === 0) return null;
+    const root = rows.find(r => (r.id as number) === id);
+    return root ? rowToMinionJob(root) : null;
   }
 
   /** Re-queue a failed or dead job for retry. */
@@ -210,7 +352,12 @@ export class MinionQueue {
     };
   }
 
-  /** Claim the next waiting job for a worker. Token-fenced, filters by registered names. */
+  /**
+   * Claim the next waiting job for a worker. Token-fenced, filters by registered names.
+   *
+   * Sets timeout_at = now() + timeout_ms when the job has a per-job deadline,
+   * so handleTimeouts() can dead-letter expired jobs without rereading timeout_ms.
+   */
   async claim(lockToken: string, lockDurationMs: number, queue: string, registeredNames: string[]): Promise<MinionJob | null> {
     if (registeredNames.length === 0) return null;
 
@@ -219,6 +366,9 @@ export class MinionQueue {
         status = 'active',
         lock_token = $1,
         lock_until = now() + ($2::double precision * interval '1 millisecond'),
+        timeout_at = CASE WHEN timeout_ms IS NOT NULL
+                          THEN now() + (timeout_ms::double precision * interval '1 millisecond')
+                          ELSE NULL END,
         attempts_started = attempts_started + 1,
         started_at = COALESCE(started_at, now()),
         updated_at = now()
@@ -235,36 +385,155 @@ export class MinionQueue {
     return rows.length > 0 ? rowToMinionJob(rows[0]) : null;
   }
 
-  /** Complete a job (token-fenced). Rolls up token counts to parent if applicable. Returns null if token mismatch. */
-  async completeJob(id: number, lockToken: string, result?: Record<string, unknown>): Promise<MinionJob | null> {
+  /**
+   * Dead-letter active jobs whose timeout_at has passed.
+   *
+   * The lock_until > now() guard is critical: a stalled job (lock_until < now)
+   * is being requeued by handleStalled, NOT timed out terminally. Stall →
+   * retry, timeout → dead. Order in worker loop: handleStalled() before
+   * handleTimeouts() to give stall recovery first crack.
+   *
+   * Honest scope: 1-tick TOCTOU window remains. A job whose lock_until
+   * expires between handleStalled and handleTimeouts may miss this tick
+   * but will be caught the next one (after re-claim). Never double-handled.
+   */
+  async handleTimeouts(): Promise<MinionJob[]> {
     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]
+      `UPDATE minion_jobs SET
+        status = 'dead',
+        error_text = 'timeout exceeded',
+        lock_token = NULL,
+        lock_until = NULL,
+        finished_at = now(),
+        updated_at = now()
+       WHERE status = 'active'
+         AND timeout_at IS NOT NULL
+         AND timeout_at < now()
+         AND lock_until > now()
+       RETURNING *`
     );
-    if (rows.length === 0) return null;
-
-    const completed = rowToMinionJob(rows[0]);
-
-    // Roll up token counts to parent (if parent exists and is not in a terminal state)
-    if (completed.parent_job_id && (completed.tokens_input > 0 || completed.tokens_output > 0 || completed.tokens_cache_read > 0)) {
-      await this.engine.executeRaw(
-        `UPDATE minion_jobs SET
-          tokens_input = tokens_input + $1,
-          tokens_output = tokens_output + $2,
-          tokens_cache_read = tokens_cache_read + $3,
-          updated_at = now()
-         WHERE id = $4 AND status NOT IN ('completed', 'dead', 'cancelled')`,
-        [completed.tokens_input, completed.tokens_output, completed.tokens_cache_read, completed.parent_job_id]
-      );
-    }
-
-    return completed;
+    return rows.map(rowToMinionJob);
   }
 
-  /** Fail a job (token-fenced). Sets delayed for retry or dead/failed for terminal. */
+  /**
+   * Complete a job (token-fenced). All side effects atomic in one transaction:
+   *   1. UPDATE child to 'completed' with result
+   *   2. Roll up token counts to parent (skipped if parent is terminal)
+   *   3. Insert child_done message into parent's inbox (skipped if parent terminal)
+   *   4. Resolve parent (flip waiting-children → waiting if all kids done)
+   *   5. If remove_on_complete, DELETE the child row (cascades inbox + attachments)
+   *
+   * Returns the completed job (the in-memory snapshot before any delete), or
+   * null if the lock_token mismatched (e.g., reclaimed mid-completion).
+   *
+   * The fold-in of resolveParent eliminates the crash window where a process
+   * died between completeJob and worker's prior post-call resolveParent,
+   * stranding the parent in waiting-children forever.
+   */
+  async completeJob(id: number, lockToken: string, result?: Record<string, unknown>): Promise<MinionJob | null> {
+    return this.engine.transaction(async (tx) => {
+      // Peek at parent_job_id before the UPDATE so we can lock the parent row
+      // FIRST. Without this SELECT FOR UPDATE, two siblings completing
+      // concurrently each see the other as still active (pre-commit snapshot
+      // under read-committed), neither flips the parent, and the parent is
+      // stuck in waiting-children forever.
+      const peek = await tx.executeRaw<{ parent_job_id: number | null }>(
+        `SELECT parent_job_id FROM minion_jobs WHERE id = $1`,
+        [id]
+      );
+      const parentId = peek[0]?.parent_job_id ?? null;
+      if (parentId) {
+        await tx.executeRaw(
+          `SELECT id FROM minion_jobs WHERE id = $1 FOR UPDATE`,
+          [parentId]
+        );
+      }
+
+      const rows = await tx.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]
+      );
+      if (rows.length === 0) return null;
+
+      const completed = rowToMinionJob(rows[0]);
+
+      if (completed.parent_job_id) {
+        // Roll up token counts. Guarded against parent already being terminal.
+        if (completed.tokens_input > 0 || completed.tokens_output > 0 || completed.tokens_cache_read > 0) {
+          await tx.executeRaw(
+            `UPDATE minion_jobs SET
+              tokens_input = tokens_input + $1,
+              tokens_output = tokens_output + $2,
+              tokens_cache_read = tokens_cache_read + $3,
+              updated_at = now()
+             WHERE id = $4 AND status NOT IN ('completed', 'failed', 'dead', 'cancelled')`,
+            [completed.tokens_input, completed.tokens_output, completed.tokens_cache_read, completed.parent_job_id]
+          );
+        }
+
+        // Auto-post child_done into parent's inbox. EXISTS guard skips if parent
+        // was deleted or hit a terminal state mid-flight (no FK violation, no
+        // contradiction with the token rollup guard).
+        const childDone: ChildDoneMessage = {
+          type: 'child_done',
+          child_id: completed.id,
+          job_name: completed.name,
+          result: result ?? null,
+        };
+        await tx.executeRaw(
+          `INSERT INTO minion_inbox (job_id, sender, payload)
+           SELECT $1, 'minions', $2::jsonb
+           WHERE EXISTS (
+             SELECT 1 FROM minion_jobs
+             WHERE id = $1 AND status NOT IN ('completed','failed','dead','cancelled')
+           )`,
+          [completed.parent_job_id, childDone]
+        );
+
+        // Fold-in resolveParent: flip parent to waiting once all children done.
+        await tx.executeRaw(
+          `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')
+             )`,
+          [completed.parent_job_id]
+        );
+      }
+
+      // remove_on_complete cleanup AFTER all parent-side bookkeeping.
+      // The child_done we just inserted lives in the *parent's* inbox row,
+      // so it survives the child cascade-delete.
+      if (completed.remove_on_complete) {
+        await tx.executeRaw(
+          `DELETE FROM minion_jobs WHERE id = $1`,
+          [completed.id]
+        );
+      }
+
+      return completed;
+    });
+  }
+
+  /**
+   * Fail a job (token-fenced). All side effects atomic in one transaction:
+   *   1. UPDATE child to 'delayed' (retry) | 'failed' | 'dead'
+   *   2. If terminal AND parent_job_id, run on_child_fail policy:
+   *      - 'fail_parent' → mark parent 'failed' (via failParent SQL)
+   *      - 'remove_dep'  → null out parent_job_id (via removeChildDependency SQL)
+   *      - 'ignore' / 'continue' → no parent action
+   *   3. If remove_on_fail AND terminal, DELETE the child row (parent hook
+   *      already ran in this txn using in-memory state, so child deletion is safe)
+   *
+   * Folding the parent hook into this transaction eliminates the crash window
+   * where a process died between failJob and worker's prior post-call hook,
+   * leaving the parent stuck in waiting-children.
+   */
   async failJob(
     id: number,
     lockToken: string,
@@ -272,18 +541,76 @@ export class MinionQueue {
     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;
+    return this.engine.transaction(async (tx) => {
+      // Lock the parent row first so concurrent sibling completions/failures
+      // serialize on the parent — same race fix as completeJob.
+      const peek = await tx.executeRaw<{ parent_job_id: number | null }>(
+        `SELECT parent_job_id FROM minion_jobs WHERE id = $1`,
+        [id]
+      );
+      const parentId = peek[0]?.parent_job_id ?? null;
+      if (parentId) {
+        await tx.executeRaw(
+          `SELECT id FROM minion_jobs WHERE id = $1 FOR UPDATE`,
+          [parentId]
+        );
+      }
+
+      const rows = await tx.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]
+      );
+      if (rows.length === 0) return null;
+
+      const failed = rowToMinionJob(rows[0]);
+      const terminal = newStatus === 'failed' || newStatus === 'dead';
+
+      // Parent hook on terminal failure.
+      if (terminal && failed.parent_job_id) {
+        if (failed.on_child_fail === 'fail_parent') {
+          await tx.executeRaw(
+            `UPDATE minion_jobs SET status = 'failed',
+              error_text = $1, finished_at = now(), updated_at = now()
+             WHERE id = $2 AND status = 'waiting-children'`,
+            [`child job ${failed.id} failed: ${errorText}`, failed.parent_job_id]
+          );
+        } else if (failed.on_child_fail === 'remove_dep') {
+          await tx.executeRaw(
+            `UPDATE minion_jobs SET parent_job_id = NULL, updated_at = now() WHERE id = $1`,
+            [failed.id]
+          );
+          // After dropping the dep, try to resolve the parent if all OTHER kids are done.
+          await tx.executeRaw(
+            `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')
+               )`,
+            [failed.parent_job_id]
+          );
+        }
+        // 'ignore' / 'continue' → parent stays in waiting-children waiting on siblings
+      }
+
+      // remove_on_fail cleanup AFTER parent hook.
+      if (terminal && failed.remove_on_fail) {
+        await tx.executeRaw(
+          `DELETE FROM minion_jobs WHERE id = $1`,
+          [failed.id]
+        );
+      }
+
+      return failed;
+    });
   }
 
   /** Update job progress (token-fenced). */
@@ -292,7 +619,7 @@ export class MinionQueue {
       `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]
+      [progress, id, lockToken]
     );
     return rows.length > 0;
   }
@@ -425,7 +752,7 @@ export class MinionQueue {
       `INSERT INTO minion_inbox (job_id, sender, payload)
        VALUES ($1, $2, $3)
        RETURNING *`,
-      [jobId, sender, JSON.stringify(payload)]
+      [jobId, sender, payload]
     );
     return rows.length > 0 ? rowToInboxMessage(rows[0]) : null;
   }
@@ -490,4 +817,137 @@ export class MinionQueue {
       [childId]
     );
   }
+
+  /**
+   * Read child_done messages from a parent's inbox. Token-fenced (the parent
+   * job must currently hold lockToken — same fence as readInbox to prevent a
+   * stale process polling completions for jobs it no longer owns).
+   *
+   * Does NOT mark messages read (parent may want to poll repeatedly with a
+   * cursor). Use `since` to fetch only newer entries.
+   */
+  async readChildCompletions(
+    parentId: number,
+    lockToken: string,
+    opts?: { since?: Date }
+  ): Promise<ChildDoneMessage[]> {
+    // Verify the caller holds the parent's lock.
+    const lockCheck = await this.engine.executeRaw<{ id: number }>(
+      `SELECT id FROM minion_jobs WHERE id = $1 AND lock_token = $2 AND status = 'active'`,
+      [parentId, lockToken]
+    );
+    if (lockCheck.length === 0) return [];
+
+    const params: unknown[] = [parentId];
+    let sinceClause = '';
+    if (opts?.since) {
+      sinceClause = ` AND sent_at > $2`;
+      params.push(opts.since.toISOString());
+    }
+
+    const rows = await this.engine.executeRaw<Record<string, unknown>>(
+      `SELECT payload FROM minion_inbox
+       WHERE job_id = $1 AND (payload->>'type') = 'child_done'${sinceClause}
+       ORDER BY sent_at ASC`,
+      params
+    );
+
+    return rows.map(r => {
+      const p = typeof r.payload === 'string' ? JSON.parse(r.payload) : r.payload;
+      return p as ChildDoneMessage;
+    });
+  }
+
+  /**
+   * Attach a file to a job. Validates size, base64, filename safety, and
+   * duplicate filename. Returns the persisted attachment metadata (not the
+   * bytes — use getAttachment to fetch).
+   *
+   * The DB UNIQUE (job_id, filename) constraint is the authoritative duplicate
+   * fence; the in-memory check just gives a faster error.
+   */
+  async addAttachment(jobId: number, input: AttachmentInput): Promise<Attachment> {
+    await this.ensureSchema();
+
+    // Verify job exists (FK guarantees this on insert too, but explicit error is clearer)
+    const exists = await this.engine.executeRaw<{ id: number }>(
+      `SELECT id FROM minion_jobs WHERE id = $1`,
+      [jobId]
+    );
+    if (exists.length === 0) {
+      throw new Error(`job ${jobId} not found`);
+    }
+
+    const existingRows = await this.engine.executeRaw<{ filename: string }>(
+      `SELECT filename FROM minion_attachments WHERE job_id = $1`,
+      [jobId]
+    );
+    const existingFilenames = new Set(existingRows.map(r => r.filename));
+
+    const result = validateAttachment(input, {
+      maxBytes: this.maxAttachmentBytes,
+      existingFilenames,
+    });
+    if (!result.ok) {
+      throw new Error(`attachment validation failed: ${result.error}`);
+    }
+    const { filename, content_type, bytes, size_bytes, sha256 } = result.normalized;
+
+    const rows = await this.engine.executeRaw<Record<string, unknown>>(
+      `INSERT INTO minion_attachments (job_id, filename, content_type, content, size_bytes, sha256)
+       VALUES ($1, $2, $3, $4, $5, $6)
+       RETURNING id, job_id, filename, content_type, storage_uri, size_bytes, sha256, created_at`,
+      [jobId, filename, content_type, bytes, size_bytes, sha256]
+    );
+    return rowToAttachment(rows[0]);
+  }
+
+  /** List attachments for a job (metadata only, no bytes). */
+  async listAttachments(jobId: number): Promise<Attachment[]> {
+    const rows = await this.engine.executeRaw<Record<string, unknown>>(
+      `SELECT id, job_id, filename, content_type, storage_uri, size_bytes, sha256, created_at
+       FROM minion_attachments
+       WHERE job_id = $1
+       ORDER BY created_at ASC, id ASC`,
+      [jobId]
+    );
+    return rows.map(rowToAttachment);
+  }
+
+  /**
+   * Fetch a single attachment with bytes. Returns null if not found.
+   * The bytes are returned as a Buffer (Uint8Array under the hood).
+   */
+  async getAttachment(jobId: number, filename: string): Promise<{ meta: Attachment; bytes: Buffer } | null> {
+    const rows = await this.engine.executeRaw<Record<string, unknown>>(
+      `SELECT id, job_id, filename, content_type, storage_uri, size_bytes, sha256, created_at, content
+       FROM minion_attachments
+       WHERE job_id = $1 AND filename = $2`,
+      [jobId, filename]
+    );
+    if (rows.length === 0) return null;
+    const row = rows[0];
+    const meta = rowToAttachment(row);
+    const raw = row.content;
+    let bytes: Buffer;
+    if (raw == null) {
+      bytes = Buffer.alloc(0);
+    } else if (Buffer.isBuffer(raw)) {
+      bytes = raw;
+    } else if (raw instanceof Uint8Array) {
+      bytes = Buffer.from(raw);
+    } else {
+      bytes = Buffer.from(raw as ArrayBuffer);
+    }
+    return { meta, bytes };
+  }
+
+  /** Delete an attachment by job + filename. Returns true if a row was removed. */
+  async deleteAttachment(jobId: number, filename: string): Promise<boolean> {
+    const rows = await this.engine.executeRaw<{ id: number }>(
+      `DELETE FROM minion_attachments WHERE job_id = $1 AND filename = $2 RETURNING id`,
+      [jobId, filename]
+    );
+    return rows.length > 0;
+  }
 }