Merge remote-tracking branch 'origin/master' into garrytan/cathedral-1

# Conflicts:
#	CHANGELOG.md
#	TODOS.md
#	VERSION
#	package.json

src/commands/agent.ts

@@ -74,8 +74,12 @@ SUBMITTING
     --follow                     Tail status until terminal (default on TTY)
     --detach                     Submit + print job id, exit immediately
 
-  Flags after \`run\` up to the first unrecognized token are parsed; the
-  remainder is the prompt. Use \`--\` to explicitly terminate flag parsing.
+  Flags before the prompt are parsed normally. The no-value switches
+  --detach, --follow and --no-follow are ALSO recognized when they trail
+  the prompt, so \`gbrain agent run "do X" --detach\` detaches. Any other
+  --word is treated as prompt text (no error). Use \`--\` to end flag
+  parsing and pass the rest verbatim:
+    gbrain agent run -- "literally --detach this, with --flags"
 
 VIEWING
   gbrain agent logs <job_id>
@@ -102,31 +106,86 @@ interface RunFlags {
   detach: boolean;
 }
 
+/** No-value switches that may also trail the prompt and get hoisted out (#1738). */
+const BOOLEAN_TAIL_FLAGS = new Set(['--follow', '--no-follow', '--detach']);
+
+function applyBooleanFlag(flags: RunFlags, a: string): void {
+  if (a === '--follow') flags.follow = true;
+  else if (a === '--no-follow') flags.follow = false;
+  else { flags.detach = true; flags.follow = false; } // --detach
+}
+
+/** Read the value for a value-flag, rejecting a missing or flag-shaped value. */
+function requireFlagValue(args: string[], i: number, flag: string): string {
+  const v = args[i];
+  if (v === undefined || v.startsWith('--')) {
+    throw new Error(`gbrain agent run: ${flag} requires a value. Run \`gbrain agent run --help\`.`);
+  }
+  return v;
+}
+
+function parseIntFlagValue(v: string, flag: string): number {
+  const n = parseInt(v, 10);
+  if (Number.isNaN(n)) {
+    throw new Error(`gbrain agent run: ${flag} expects a number, got "${v}".`);
+  }
+  return n;
+}
+
+/**
+ * Parse `agent run` args into flags + prompt (#1738).
+ *
+ *   args ──► [ leading flag zone ] [ ── ? ] [ prompt … (trailing booleans) ]
+ *
+ * Leading zone: known flags (value + boolean) are consumed left-to-right until
+ * the first positional token, an UNKNOWN --flag, or an explicit `--`. An
+ * unknown --flag is NOT an error — it begins the freeform prompt, so
+ * `agent run "--note: do X"` works without `--`. Value-flags missing their
+ * value throw a usage error instead of silently capturing `undefined`/`NaN`.
+ *
+ * Prompt zone: a trailing run of the no-value switches (--detach/--follow/
+ * --no-follow) is hoisted out so `agent run "do X" --detach` detaches. Only
+ * trailing switches are hoisted; a `--word` elsewhere in the prompt stays
+ * verbatim. After an explicit `--`, nothing is hoisted.
+ */
 function parseRunFlags(args: string[]): { flags: RunFlags; rest: string[] } {
   const flags: RunFlags = {
     follow: process.stdout.isTTY === true,
     detach: false,
   };
   let i = 0;
-  while (i < args.length) {
-    const a = args[i];
-    if (a === '--') { i++; break; }
-    if (!isKnownFlag(a!)) break;
+  let escaped = false;
+  for (; i < args.length; i++) {
+    const a = args[i]!;
+    if (a === '--') { i++; escaped = true; break; }
+    if (!a.startsWith('--')) break;
+    let known = true;
     switch (a) {
-      case '--subagent-def': flags.subagentDef = args[++i]; i++; break;
-      case '--model':        flags.model = args[++i]; i++; break;
-      case '--max-turns':    flags.maxTurns = parseInt(args[++i] ?? '', 10); i++; break;
-      case '--tools':        flags.tools = (args[++i] ?? '').split(',').map(s => s.trim()).filter(Boolean); i++; break;
-      case '--timeout-ms':   flags.timeoutMs = parseInt(args[++i] ?? '', 10); i++; break;
-      case '--fanout-manifest': flags.fanoutManifest = args[++i]; i++; break;
-      case '--follow':       flags.follow = true; i++; break;
-      case '--no-follow':    flags.follow = false; i++; break;
-      case '--detach':       flags.detach = true; flags.follow = false; i++; break;
-      default:
-        throw new Error(`unknown flag: ${a}. Run \`gbrain agent run --help\` for usage.`);
+      case '--subagent-def':    flags.subagentDef = requireFlagValue(args, ++i, a); break;
+      case '--model':           flags.model = requireFlagValue(args, ++i, a); break;
+      case '--max-turns':       flags.maxTurns = parseIntFlagValue(requireFlagValue(args, ++i, a), a); break;
+      case '--tools':           flags.tools = requireFlagValue(args, ++i, a).split(',').map(s => s.trim()).filter(Boolean); break;
+      case '--timeout-ms':      flags.timeoutMs = parseIntFlagValue(requireFlagValue(args, ++i, a), a); break;
+      case '--fanout-manifest': flags.fanoutManifest = requireFlagValue(args, ++i, a); break;
+      case '--follow':          flags.follow = true; break;
+      case '--no-follow':       flags.follow = false; break;
+      case '--detach':          flags.detach = true; flags.follow = false; break;
+      default:                  known = false; break;
+    }
+    if (!known) break; // unknown --flag → first token of the (freeform) prompt
+  }
+  const rest = args.slice(i);
+  // An explicit `--` terminates flag parsing wherever it appears — leading
+  // zone (escaped) OR after a positional (the leading loop breaks before it,
+  // so `escaped` stays false). Honor both: when the prompt carries a literal
+  // `--`, hoist nothing, so `agent run note -- --detach` keeps `--detach`
+  // verbatim instead of silently flipping detach mode.
+  if (!escaped && !rest.includes('--')) {
+    while (rest.length > 0 && BOOLEAN_TAIL_FLAGS.has(rest[rest.length - 1]!)) {
+      applyBooleanFlag(flags, rest.pop()!);
     }
   }
-  return { flags, rest: args.slice(i) };
+  return { flags, rest };
 }
 
 export async function runAgentRun(engine: BrainEngine, args: string[]): Promise<void> {
@@ -251,7 +310,7 @@ async function runFanout(engine: BrainEngine, queue: MinionQueue, flags: RunFlag
   // do this after submission because each add() returns the committed
   // row's id; the aggregator's seed started with an empty array.
   await engine.executeRaw(
-    `UPDATE minion_jobs SET data = jsonb_set(data, '{children_ids}', $1::jsonb) WHERE id = $2`,
+    `UPDATE minion_jobs SET data = jsonb_set(data, '{children_ids}', $1::text::jsonb) WHERE id = $2`,
     [JSON.stringify(childIds), aggregator.id],
   );
 

src/commands/autopilot-fanout.ts

@@ -32,9 +32,26 @@
 
 import type { BrainEngine, SourceRow } from '../core/engine.ts';
 import type { MinionQueue } from '../core/minions/queue.ts';
+import { NON_GLOBAL_PHASES, GLOBAL_PHASES, LAST_GLOBAL_AT_KEY } from '../core/cycle.ts';
 
 const FULL_CYCLE_FLOOR_MIN = 60;
 
+// #2194 fix #2: failure cooldown. A source whose autopilot-cycle keeps
+// failing/timing-out re-dispatches every tick today (only SUCCESS gates
+// dispatch), so the same handful of sources fail and re-fan-out forever — the
+// self-perpetuating dead-job storm. Back a failed source off with bounded
+// exponential cooldown so a chronically-slow source can't re-dispatch every
+// tick. Disabled with autopilot.failure_cooldown_min=0.
+const FAILURE_COOLDOWN_BASE_MIN = 10;
+const FAILURE_COOLDOWN_CAP_MIN = 120;
+const FAILURE_COOLDOWN_EXP_CAP = 4; // 2^4 = 16× base before the cap clamps
+
+/** Recent-failure record for one source (from minion_jobs dead/failed rows). */
+export interface SourceFailure { count: number; lastFailedAt: Date; }
+
+/** Resolved cooldown knobs. baseMin <= 0 means the cooldown is disabled. */
+export interface CooldownOpts { baseMin: number; capMin: number; }
+
 export interface FanoutOpts {
   repoPath: string;
   slot: string;
@@ -58,6 +75,8 @@ export interface FanoutResult {
   skipped_fresh: string[];
   /** Source ids beyond the fanoutMax cap (will retry next tick). */
   skipped_cap: string[];
+  /** Source ids skipped because they're in failure cooldown (#2194 fix #2). */
+  skipped_cooldown: string[];
   /** True when this tick fell back to the legacy single-job path
    *  (no sources rows / engine empty). */
   legacy_fallback: boolean;
@@ -83,6 +102,62 @@ export async function resolveFanoutMax(engine: BrainEngine): Promise<number> {
   return engine.kind === 'pglite' ? 1 : 4;
 }
 
+/**
+ * Read the worker concurrency the supervisor most recently STARTED with, from
+ * its `started` audit event (the lowest-coupling source — no extra lock-row
+ * column). Filesystem read; returns null when no supervisor has ever started
+ * (or the event lacks concurrency). Filtered by queue so a `shell`-queue
+ * supervisor's concurrency doesn't leak into the `default`-queue decision.
+ *
+ * ADVISORY use only (doctor warning). Behavior-changing callers (the fanout
+ * clamp) must additionally gate on a LIVE supervisor — see
+ * resolveEffectiveFanoutMax — because a stale `started` row can otherwise
+ * shrink fan-out for a supervisor that isn't running that config (codex #9/D5).
+ */
+export async function readSupervisorConcurrency(queue = 'default'): Promise<number | null> {
+  try {
+    const { readSupervisorEvents } = await import('../core/minions/handlers/supervisor-audit.ts');
+    const events = readSupervisorEvents({ sinceMs: 24 * 60 * 60 * 1000 });
+    const started = events
+      .filter((e) => e.event === 'started' && (e.queue === undefined || e.queue === queue))
+      .pop();
+    const c = started?.concurrency;
+    return typeof c === 'number' && Number.isFinite(c) ? c : null;
+  } catch {
+    return null;
+  }
+}
+
+/**
+ * Resolve fanoutMax CLAMPED to the worker's effective concurrency (#2194 fix #1).
+ *
+ * Fanning out more cycles than the worker can run guarantees waiters that then
+ * race the stalled-sweeper. Clamp to `max(1, concurrency - 1)` — reserving ≥1
+ * slot for targeted sync/embed jobs that share the `default` queue.
+ *
+ * codex #9 / D5: the clamp is BEHAVIOR-changing, so it trusts only a
+ * proven-alive supervisor (live DB-lock holder, `ttl_expires_at`-gated). With
+ * no live holder the concurrency is UNKNOWN and we fall back to the unclamped
+ * default (4 pg / 1 pglite) — the safe direction (never starve on stale data).
+ * Operators can disable the clamp via `autopilot.fanout_clamp_to_concurrency`.
+ */
+export async function resolveEffectiveFanoutMax(engine: BrainEngine, queue = 'default'): Promise<number> {
+  const base = await resolveFanoutMax(engine);
+  const clampCfg = await engine.getConfig('autopilot.fanout_clamp_to_concurrency');
+  if (clampCfg === 'false' || clampCfg === '0') return base; // operator opt-out
+  try {
+    const { inspectLock, isLockHolderLive } = await import('../core/db-lock.ts');
+    const { supervisorLockId, SUPERVISOR_LOCK_TTL_MIN } = await import('../core/minions/supervisor.ts');
+    const snap = await inspectLock(engine, supervisorLockId(queue));
+    if (!snap || !isLockHolderLive(snap, SUPERVISOR_LOCK_TTL_MIN)) return base; // no live holder → unknown → no clamp
+    const concurrency = await readSupervisorConcurrency(queue);
+    if (concurrency === null) return base;
+    return Math.max(1, Math.min(base, concurrency - 1));
+  } catch {
+    return base;
+  }
+}
+
 /**
  * Read `last_full_cycle_at` ISO string from a source's config JSONB.
  * Returns null when missing or unparseable. Pure function over the row
@@ -111,6 +186,133 @@ export function isSourceStale(src: SourceRow, now = Date.now(), floorMin = FULL_
   return ageMin >= floorMin;
 }
 
+/**
+ * Most recent SUCCESSFUL cycle for a source. Prefers `last_source_cycle_at`
+ * (per-source phases, written by the split cycle) and falls back to the legacy
+ * `last_full_cycle_at`, so this works before AND after the cycle split.
+ */
+export function readLastSuccessAt(src: SourceRow): Date | null {
+  const c = src.config ?? {};
+  const raw = (typeof c.last_source_cycle_at === 'string' && c.last_source_cycle_at)
+    || (typeof c.last_full_cycle_at === 'string' && c.last_full_cycle_at)
+    || null;
+  if (!raw) return null;
+  const d = new Date(raw);
+  return Number.isFinite(d.getTime()) ? d : null;
+}
+
+/** Bounded exponential cooldown window (minutes) for a given failure count. */
+export function cooldownMinForCount(count: number, opts: CooldownOpts): number {
+  if (count <= 0 || opts.baseMin <= 0) return 0;
+  const mult = Math.pow(2, Math.min(count - 1, FAILURE_COOLDOWN_EXP_CAP));
+  return Math.min(opts.baseMin * mult, opts.capMin);
+}
+
+/**
+ * Is a source currently in failure cooldown? Pure — drives both the dispatch
+ * gate and the claim-time guard. A SUCCESS at-or-after the most recent failure
+ * clears the cooldown (codex #7: operator repair / manual cycle re-eligibility),
+ * so a recovered source is never suppressed by stale failure history.
+ */
+export function isInFailureCooldown(
+  failure: SourceFailure | undefined,
+  lastSuccessAt: Date | null,
+  now: number,
+  opts: CooldownOpts,
+): boolean {
+  if (opts.baseMin <= 0) return false;            // disabled
+  if (!failure || failure.count <= 0) return false;
+  if (lastSuccessAt && lastSuccessAt.getTime() >= failure.lastFailedAt.getTime()) return false;
+  const cooldownMs = cooldownMinForCount(failure.count, opts) * 60_000;
+  return (now - failure.lastFailedAt.getTime()) < cooldownMs;
+}
+
+/**
+ * Resolve cooldown knobs from config. `autopilot.failure_cooldown_min` overrides
+ * the base (0 = disable entirely — exactly today's behavior);
+ * `autopilot.failure_cooldown_cap_min` overrides the ceiling.
+ */
+export async function resolveFailureCooldownOpts(engine: BrainEngine): Promise<CooldownOpts> {
+  let baseMin = FAILURE_COOLDOWN_BASE_MIN;
+  let capMin = FAILURE_COOLDOWN_CAP_MIN;
+  const baseCfg = await engine.getConfig('autopilot.failure_cooldown_min');
+  if (baseCfg !== null && baseCfg !== undefined && baseCfg !== '') {
+    const n = parseInt(baseCfg, 10);
+    if (Number.isFinite(n) && n >= 0) baseMin = n;
+  }
+  const capCfg = await engine.getConfig('autopilot.failure_cooldown_cap_min');
+  if (capCfg) {
+    const n = parseInt(capCfg, 10);
+    if (Number.isFinite(n) && n >= 1) capMin = n;
+  }
+  return { baseMin, capMin };
+}
+
+/**
+ * Read recent dead/failed autopilot-cycle jobs grouped by source. Read-at-
+ * dispatch (NOT a write hook) because timeouts/RSS-kills/stalls dead-letter via
+ * SQL in queue.ts and never run handler code — a write-only cooldown would miss
+ * the exact failures that drive the storm. Engine-parity-safe via executeRaw
+ * (one query, both engines); cutoff is precomputed in JS to avoid INTERVAL
+ * portability concerns. codex #6: rows with a null source_id are excluded.
+ */
+export async function readRecentSourceFailures(
+  engine: BrainEngine,
+  opts: { sinceMin?: number; sourceId?: string } = {},
+): Promise<Map<string, SourceFailure>> {
+  const sinceMin = opts.sinceMin ?? FAILURE_COOLDOWN_CAP_MIN;
+  const cutoff = new Date(Date.now() - sinceMin * 60_000).toISOString();
+  const map = new Map<string, SourceFailure>();
+  try {
+    const params: unknown[] = [cutoff];
+    let sql =
+      `SELECT data->>'source_id' AS source_id,
+              count(*)::int AS fail_count,
+              max(finished_at) AS last_failed_at
+         FROM minion_jobs
+        WHERE name = 'autopilot-cycle'
+          AND status IN ('dead','failed')
+          AND data->>'source_id' IS NOT NULL
+          AND finished_at IS NOT NULL
+          AND finished_at > $1`;
+    if (opts.sourceId) { params.push(opts.sourceId); sql += ` AND data->>'source_id' = $${params.length}`; }
+    sql += ` GROUP BY data->>'source_id'`;
+    const rows = await engine.executeRaw<{ source_id: string | null; fail_count: number; last_failed_at: string | Date }>(sql, params);
+    for (const r of rows) {
+      if (!r.source_id) continue; // codex #6 null-source guard (defensive)
+      const last = r.last_failed_at instanceof Date ? r.last_failed_at : new Date(r.last_failed_at);
+      if (!Number.isFinite(last.getTime())) continue;
+      map.set(r.source_id, { count: Number(r.fail_count) || 0, lastFailedAt: last });
+    }
+  } catch {
+    // Pre-migration / transient DB error → no cooldown data (fail open: dispatch).
+  }
+  return map;
+}
+
+/**
+ * Claim-time cooldown guard (codex #5 / D4): a job already queued or retrying
+ * (max_attempts:2) can reach the worker after the dispatch gate decided. The
+ * handler calls this immediately before runCycle; an in-cooldown claim becomes
+ * a no-op skip (NOT a failure — it must not re-arm the cooldown). Shares the
+ * exact cooldown math with the dispatch gate (DRY).
+ */
+export async function isSourceInCooldown(engine: BrainEngine, sourceId: string, now = Date.now()): Promise<boolean> {
+  const opts = await resolveFailureCooldownOpts(engine);
+  if (opts.baseMin <= 0) return false;
+  const failures = await readRecentSourceFailures(engine, { sinceMin: opts.capMin, sourceId });
+  const failure = failures.get(sourceId);
+  if (!failure) return false;
+  let lastSuccessAt: Date | null = null;
+  try {
+    const rows = await engine.executeRaw<{ config: Record<string, unknown> | null }>(
+      `SELECT config FROM sources WHERE id = $1`, [sourceId],
+    );
+    if (rows[0]) lastSuccessAt = readLastSuccessAt({ config: rows[0].config ?? {} } as SourceRow);
+  } catch { /* treat as no success */ }
+  return isInFailureCooldown(failure, lastSuccessAt, now, opts);
+}
+
 /**
  * Decide which sources to dispatch this tick. Pure function so tests can
  * exercise the freshness gate + cap math without an engine.
@@ -126,11 +328,21 @@ export function selectSourcesForDispatch(
   fanoutMax: number,
   now = Date.now(),
   floorMin = FULL_CYCLE_FLOOR_MIN,
-): { dispatch: SourceRow[]; skippedFresh: SourceRow[]; skippedCap: SourceRow[] } {
+  recentFailures: Map<string, SourceFailure> = new Map(),
+  cooldownOpts: CooldownOpts = { baseMin: FAILURE_COOLDOWN_BASE_MIN, capMin: FAILURE_COOLDOWN_CAP_MIN },
+): { dispatch: SourceRow[]; skippedFresh: SourceRow[]; skippedCap: SourceRow[]; skippedCooldown: SourceRow[] } {
   const stale: SourceRow[] = [];
   const fresh: SourceRow[] = [];
+  const cooldown: SourceRow[] = [];
   for (const s of sources) {
-    (isSourceStale(s, now, floorMin) ? stale : fresh).push(s);
+    if (!isSourceStale(s, now, floorMin)) { fresh.push(s); continue; }
+    // #2194 fix #2: a stale source that recently failed is held in cooldown so
+    // it can't re-dispatch every tick (the storm). Success clears it.
+    if (isInFailureCooldown(recentFailures.get(s.id), readLastSuccessAt(s), now, cooldownOpts)) {
+      cooldown.push(s);
+      continue;
+    }
+    stale.push(s);
   }
   // Oldest-first ordering: NULL last_full_cycle_at sorts before any timestamp.
   stale.sort((a, b) => {
@@ -141,7 +353,7 @@ export function selectSourcesForDispatch(
   });
   const dispatch = stale.slice(0, fanoutMax);
   const skippedCap = stale.slice(fanoutMax);
-  return { dispatch, skippedFresh: fresh, skippedCap };
+  return { dispatch, skippedFresh: fresh, skippedCap, skippedCooldown: cooldown };
 }
 
 /**
@@ -193,10 +405,27 @@ export async function dispatchPerSource(
     } else {
       log(`[dispatch] job #${job.id} autopilot-cycle (legacy single-source)`);
     }
-    return { dispatched: [], skipped_fresh: [], skipped_cap: [], legacy_fallback: true };
+    return { dispatched: [], skipped_fresh: [], skipped_cap: [], skipped_cooldown: [], legacy_fallback: true };
   }
 
-  const { dispatch, skippedFresh, skippedCap } = selectSourcesForDispatch(sources, opts.fanoutMax);
+  // #2194 fix #2: load recent per-source failures + cooldown knobs so a
+  // chronically-failing source is backed off instead of re-dispatched every
+  // tick. Fail-open: cooldown is an optimization, not a correctness gate — if
+  // config/job-history reads fail (or the engine lacks them), dispatch proceeds
+  // with no cooldown rather than blocking.
+  let cooldownOpts: CooldownOpts = { baseMin: 0, capMin: FAILURE_COOLDOWN_CAP_MIN };
+  let recentFailures = new Map<string, SourceFailure>();
+  try {
+    cooldownOpts = await resolveFailureCooldownOpts(engine);
+    if (cooldownOpts.baseMin > 0) {
+      recentFailures = await readRecentSourceFailures(engine, { sinceMin: cooldownOpts.capMin });
+    }
+  } catch {
+    cooldownOpts = { baseMin: 0, capMin: FAILURE_COOLDOWN_CAP_MIN };
+  }
+
+  const { dispatch, skippedFresh, skippedCap, skippedCooldown } =
+    selectSourcesForDispatch(sources, opts.fanoutMax, Date.now(), FULL_CYCLE_FLOOR_MIN, recentFailures, cooldownOpts);
 
   const dispatched: string[] = [];
   for (const src of dispatch) {
@@ -208,6 +437,11 @@ export async function dispatchPerSource(
           repoPath: opts.repoPath,
           source_id: src.id,
           pull: !!remoteUrl,
+          // #2194 fix #3 (cycle split): per-source cycles run ONLY source-scoped
+          // (+ mixed) phases. The brain-wide global phases (embed, orphans,
+          // purge, …) run once in autopilot-global-maintenance, not N times
+          // concurrently here — the fix for the 4→10GB RSS blowout.
+          phases: NON_GLOBAL_PHASES,
         },
         {
           queue: 'default',
@@ -261,10 +495,77 @@ export async function dispatchPerSource(
     }));
   }
 
+  if (skippedCooldown.length > 0 && opts.jsonMode) {
+    emit(JSON.stringify({
+      event: 'fanout_cooldown_skipped',
+      sources: skippedCooldown.map(s => s.id),
+    }));
+  }
+
   return {
     dispatched,
     skipped_fresh: skippedFresh.map(s => s.id),
     skipped_cap: skippedCap.map(s => s.id),
+    skipped_cooldown: skippedCooldown.map(s => s.id),
     legacy_fallback: false,
   };
 }
+
+const GLOBAL_FLOOR_MIN = 60;
+
+/** Is the brain-wide maintenance overdue? Null/unparseable → overdue. */
+export function isGlobalMaintenanceStale(lastGlobalAtIso: string | null, now = Date.now(), floorMin = GLOBAL_FLOOR_MIN): boolean {
+  if (!lastGlobalAtIso) return true;
+  const d = new Date(lastGlobalAtIso);
+  if (!Number.isFinite(d.getTime())) return true;
+  return (now - d.getTime()) / 60_000 >= floorMin;
+}
+
+/**
+ * #2194 fix #3 / #2227 bug #3 — dispatch the single brain-wide maintenance job
+ * that runs the `global` cycle phases (embed, orphans, purge, …) ONCE per
+ * window, instead of N per-source cycles each running them concurrently (the
+ * RSS blowout). Single-flight is structural: one `idempotency_key` +
+ * `maxWaiting:1`, so a slow run never stacks. Gated on `autopilot.last_global_at`
+ * (stamped by the handler on success). Postgres-only fan-out concern; on PGLite
+ * the file lock already serializes, but the job is still correct there.
+ */
+export async function dispatchGlobalMaintenance(
+  engine: BrainEngine,
+  queue: MinionQueue,
+  opts: { repoPath: string; slot: string; timeoutMs: number; jsonMode: boolean; emit?: (l: string) => void; log?: (l: string) => void },
+): Promise<{ dispatched: boolean; reason: 'stale' | 'fresh' }> {
+  const emit = opts.emit ?? ((line) => process.stderr.write(line + '\n'));
+  const log = opts.log ?? ((line) => console.log(line));
+
+  let floorMin = GLOBAL_FLOOR_MIN;
+  const floorCfg = await engine.getConfig('autopilot.global_floor_min');
+  if (floorCfg) {
+    const n = parseInt(floorCfg, 10);
+    if (Number.isFinite(n) && n >= 1) floorMin = n;
+  }
+  const lastGlobalAt = await engine.getConfig(LAST_GLOBAL_AT_KEY);
+  if (!isGlobalMaintenanceStale(lastGlobalAt, Date.now(), floorMin)) {
+    return { dispatched: false, reason: 'fresh' };
+  }
+
+  const job = await queue.add(
+    'autopilot-global-maintenance',
+    { repoPath: opts.repoPath, phases: GLOBAL_PHASES },
+    {
+      queue: 'default',
+      // Structural single-flight: one global job per slot; maxWaiting:1 coalesces
+      // any surplus so a slow brain-wide pass never stacks duplicates.
+      idempotency_key: `autopilot-global:${opts.slot}`,
+      max_attempts: 2,
+      timeout_ms: opts.timeoutMs,
+      maxWaiting: 1,
+    },
+  );
+  if (opts.jsonMode) {
+    emit(JSON.stringify({ event: 'dispatched', job_id: job.id, mode: 'global_maintenance', slot: opts.slot }));
+  } else {
+    log(`[dispatch] job #${job.id} autopilot-global-maintenance (brain-wide phases)`);
+  }
+  return { dispatched: true, reason: 'stale' };
+}

src/commands/autopilot.ts

@@ -871,8 +871,12 @@ export async function runAutopilot(engine: BrainEngine, args: string[]) {
           // codex P1-3). Fresh-install brains with no sources rows fall
           // back to the legacy single autopilot-cycle so existing
           // behavior is preserved.
-          const { dispatchPerSource, resolveFanoutMax } = await import('./autopilot-fanout.ts');
-          const fanoutMax = await resolveFanoutMax(engine);
+          const { dispatchPerSource, dispatchGlobalMaintenance, resolveEffectiveFanoutMax } = await import('./autopilot-fanout.ts');
+          // #2194 fix #1: clamp fan-out to the worker's effective concurrency
+          // (reserve ≥1 slot), gated on a LIVE supervisor so a stale audit row
+          // can't shrink throughput (codex #9/D5). autopilot-cycle jobs run on
+          // the 'default' queue, so that's the concurrency we compare against.
+          const fanoutMax = await resolveEffectiveFanoutMax(engine, 'default');
           const result = await dispatchPerSource(engine, queue, {
             repoPath,
             slot,
@@ -880,6 +884,18 @@ export async function runAutopilot(engine: BrainEngine, args: string[]) {
             fanoutMax,
             jsonMode,
           });
+          // #2194 fix #3 / #2227 bug #3: dispatch the single brain-wide
+          // maintenance job (embed/orphans/purge/…) once per window — the per-
+          // source cycles above no longer run global phases, so this is where
+          // the brain-wide work happens (single-flight, no RSS blowout). Only on
+          // the per-source path (legacy single-source still runs everything).
+          if (!result.legacy_fallback) {
+            try {
+              await dispatchGlobalMaintenance(engine, queue, { repoPath, slot, timeoutMs, jsonMode });
+            } catch (e) {
+              if (jsonMode) process.stderr.write(JSON.stringify({ event: 'global_maintenance_dispatch_failed', error: e instanceof Error ? e.message : String(e) }) + '\n');
+            }
+          }
           if (result.dispatched.length > 0 || result.legacy_fallback) {
             lastFullCycleAt = Date.now();
           }
@@ -889,6 +905,7 @@ export async function runAutopilot(engine: BrainEngine, args: string[]) {
               dispatched: result.dispatched,
               skipped_fresh: result.skipped_fresh,
               skipped_cap: result.skipped_cap,
+              skipped_cooldown: result.skipped_cooldown,
               legacy_fallback: result.legacy_fallback,
               fanout_max: fanoutMax,
               score,
@@ -896,7 +913,8 @@ export async function runAutopilot(engine: BrainEngine, args: string[]) {
           } else if (!result.legacy_fallback) {
             console.log(
               `[dispatch] fanout: ${result.dispatched.length} dispatched, ` +
-              `${result.skipped_fresh.length} fresh, ${result.skipped_cap.length} capped ` +
+              `${result.skipped_fresh.length} fresh, ${result.skipped_cap.length} capped, ` +
+              `${result.skipped_cooldown.length} cooldown ` +
               `(score=${score}, max=${fanoutMax})`,
             );
           }

src/commands/doctor.ts

@@ -695,6 +695,9 @@ export async function doctorReportRemote(engine: BrainEngine): Promise<DoctorRep
   // issue #1801 — wedged_queue (cross-surface parity with buildChecks).
   checks.push(await computeWedgedQueueCheck(engine));
 
+  // #2194 fix #5 — warn when autopilot fan-out exceeds worker concurrency.
+  checks.push(await computeAutopilotFanoutConcurrencyCheck(engine));
+
   // v0.41 Bug 2 / Eng D8 — subagent_health surfaces rate-lease pressure to the operator.
   checks.push(await checkSubagentHealth(engine));
 
@@ -1563,6 +1566,49 @@ export async function computeWedgedQueueCheck(engine: BrainEngine): Promise<Chec
   }
 }
 
+/**
+ * #2194 fix #5: warn when autopilot's per-tick fan-out exceeds the worker's
+ * effective concurrency. Fanning out more cycles than there are worker slots
+ * guarantees waiters that race the stalled-sweeper — a silent misconfig today.
+ * Advisory (started-event concurrency is fine here; the behavior-changing clamp
+ * in resolveEffectiveFanoutMax is the one that gates on liveness). Surfaces only
+ * when a supervisor has actually started (no noise on never-supervised brains).
+ */
+export async function computeAutopilotFanoutConcurrencyCheck(engine: BrainEngine): Promise<Check> {
+  if (engine.kind !== 'postgres') {
+    return { name: 'autopilot_fanout_concurrency', status: 'ok', message: 'PGLite — single-writer, fan-out is 1' };
+  }
+  try {
+    const { resolveFanoutMax, readSupervisorConcurrency } = await import('./autopilot-fanout.ts');
+    const concurrency = await readSupervisorConcurrency('default');
+    if (concurrency === null) {
+      return { name: 'autopilot_fanout_concurrency', status: 'ok', message: 'No supervisor observed — skipping fan-out/concurrency check' };
+    }
+    const fanoutMax = await resolveFanoutMax(engine);
+    const effectiveSlots = Math.max(1, concurrency - 1);
+    if (fanoutMax > effectiveSlots) {
+      return {
+        name: 'autopilot_fanout_concurrency',
+        status: 'warn',
+        message:
+          `autopilot fan-out (${fanoutMax}/tick) exceeds worker concurrency (${concurrency}). ` +
+          `Surplus cycles queue behind the worker and race the stalled-sweeper. ` +
+          `Lower fan-out: \`gbrain config set autopilot.fanout_max_per_tick ${effectiveSlots}\`, ` +
+          `or raise the supervisor's \`--concurrency\` to ${fanoutMax + 1}. ` +
+          `(The clamp in autopilot does this automatically unless disabled.)`,
+        details: { fanout_max: fanoutMax, concurrency, effective_slots: effectiveSlots },
+      };
+    }
+    return {
+      name: 'autopilot_fanout_concurrency',
+      status: 'ok',
+      message: `fan-out ${fanoutMax}/tick within worker concurrency ${concurrency}`,
+    };
+  } catch (e) {
+    return { name: 'autopilot_fanout_concurrency', status: 'ok', message: `Skipped (${e instanceof Error ? e.message : String(e)})` };
+  }
+}
+
 export async function checkBatchRetryHealth(_engine: BrainEngine): Promise<Check> {
   try {
     // Codex M-10: surface bad env config at doctor time.
@@ -4395,7 +4441,22 @@ export async function buildChecks(
 
     const pidStatus = readSupervisorPid(DEFAULT_PID_FILE);
     const supervisorPid = pidStatus.pid;
-    const running = pidStatus.running;
+    const pidfileRunning = pidStatus.running;
+
+    // issue #2227 fix #1/#3: DEFAULT_PID_FILE is HOME-derived, so a supervisor
+    // started under a different $HOME reads as "not running" even when healthy.
+    // Consult the queue-scoped DB singleton lock (#1849, HOME-independent) before
+    // warning. PID-reuse-safe (isLockHolderLive keys on lock freshness).
+    let detectedViaDbLock = false;
+    if (!pidfileRunning && engine) {
+      try {
+        const { inspectLock, isLockHolderLive } = await import('../core/db-lock.ts');
+        const { supervisorLockId, SUPERVISOR_LOCK_TTL_MIN } = await import('../core/minions/supervisor.ts');
+        const snap = await inspectLock(engine, supervisorLockId('default'));
+        if (snap && isLockHolderLive(snap, SUPERVISOR_LOCK_TTL_MIN)) detectedViaDbLock = true;
+      } catch { /* pre-migration / transient: pidfile-only */ }
+    }
+    const running = pidfileRunning || detectedViaDbLock;
 
     const events = readSupervisorEvents({ sinceMs: 24 * 60 * 60 * 1000 });
     const lastStart = events.filter(e => e.event === 'started').pop()?.ts ?? null;
@@ -4443,7 +4504,7 @@ export async function buildChecks(
         checks.push({
           name: 'supervisor',
           status: 'ok',
-          message: `running=true pid=${supervisorPid} last_start=${lastStart ?? 'unknown'} crashes_24h=${crashes24h} clean_exits_24h=${summary.clean_exits}`,
+          message: `running=true${detectedViaDbLock ? ' (detected via DB lock; pidfile not at the HOME-derived path)' : ` pid=${supervisorPid}`} last_start=${lastStart ?? 'unknown'} crashes_24h=${crashes24h} clean_exits_24h=${summary.clean_exits}`,
         });
       }
     }
@@ -7218,6 +7279,9 @@ export async function buildChecks(
     // waiting, zero live-lock active, stale completions) as a health error.
     progress.heartbeat('wedged_queue');
     checks.push(await computeWedgedQueueCheck(engine));
+    // #2194 fix #5 — autopilot fan-out vs worker concurrency mismatch.
+    progress.heartbeat('autopilot_fanout_concurrency');
+    checks.push(await computeAutopilotFanoutConcurrencyCheck(engine));
     // v0.40.4 graph_signals_coverage — global inbound-link density when
     // graph_signals is enabled in the active mode bundle.
     progress.heartbeat('graph_signals_coverage');

src/commands/jobs.ts

@@ -1020,10 +1020,38 @@ HANDLER TYPES (built in)
 
         const pidStatus = readSupervisorPid(pidFile);
         const supervisorPid = pidStatus.pid;
-        const running = pidStatus.running;
+        const pidfileRunning = pidStatus.running;
 
         const events = readSupervisorEvents({ sinceMs: 24 * 60 * 60 * 1000 });
         const lastStart = events.filter(e => e.event === 'started').pop()?.ts ?? null;
+
+        // issue #2227 fix #1/#3: the pidfile is HOME-derived, so a supervisor
+        // started under a different $HOME (keeper=/root vs ops=/data) reads as
+        // "not running" here even when it is healthy — the false signal that
+        // makes an operator spawn a duplicate. Fall back to the queue-scoped DB
+        // singleton lock (#1849), the HOME-independent authority. PID-reuse-safe:
+        // isLockHolderLive keys on lock freshness, never process.kill.
+        const supQueue = parseFlag(args, '--queue') ?? 'default';
+        let detectedViaDbLock = false;
+        let dbLockHolder: { holder_pid: number; holder_host: string } | null = null;
+        if (!pidfileRunning) {
+          try {
+            const { inspectLock, isLockHolderLive } = await import('../core/db-lock.ts');
+            const { supervisorLockId, SUPERVISOR_LOCK_TTL_MIN } = await import('../core/minions/supervisor.ts');
+            const snap = await inspectLock(engine, supervisorLockId(supQueue));
+            if (snap && isLockHolderLive(snap, SUPERVISOR_LOCK_TTL_MIN)) {
+              detectedViaDbLock = true;
+              dbLockHolder = { holder_pid: snap.holder_pid, holder_host: snap.holder_host };
+            }
+          } catch {
+            // Pre-migration brains / transient DB errors: fall back to pidfile-only.
+          }
+        }
+        const running = pidfileRunning || detectedViaDbLock;
+        // Surface the supervisor's recorded config from the latest `started`
+        // event (concurrency + effective --max-rss) so split-$HOME deployments
+        // see what the live-but-pidfile-invisible supervisor is running.
+        const startedEvt = events.filter(e => e.event === 'started').pop() ?? null;
         // Shared classifier — same code path runs in `gbrain doctor` so the
         // two surfaces cannot drift on what counts as a crash. Supersedes
         // v0.35.4.0's binary `classifyWorkerExit({code})` on this surface;
@@ -1038,15 +1066,20 @@ HANDLER TYPES (built in)
           nice_requested: w.nice_requested,
           nice: w.nice_now,
         }));
-        const supervisorNice = running && supervisorPid !== null
+        const supervisorNice = pidfileRunning && supervisorPid !== null
           ? getEffectiveNiceness(supervisorPid)
           : null;
 
         const status = {
           running,
-          supervisor_pid: supervisorPid,
+          detected_via: detectedViaDbLock ? 'db_lock' : (pidfileRunning ? 'pidfile' : null),
+          supervisor_pid: supervisorPid ?? dbLockHolder?.holder_pid ?? null,
+          db_lock_holder: dbLockHolder,
           pid_file: pidFile,
+          queue: supQueue,
           last_start: lastStart,
+          concurrency: typeof startedEvt?.concurrency === 'number' ? startedEvt.concurrency : null,
+          max_rss_mb: typeof startedEvt?.max_rss_mb === 'number' ? startedEvt.max_rss_mb : null,
           crashes_24h: summary.total,
           clean_exits_24h: summary.clean_exits,
           crashes_by_cause: summary.by_cause,
@@ -1058,9 +1091,11 @@ HANDLER TYPES (built in)
         if (jsonMode) {
           console.log(JSON.stringify(status, null, 2));
         } else {
-          console.log(`Supervisor: ${running ? 'running' : 'not running'}`);
-          if (supervisorPid) console.log(`  PID:           ${supervisorPid}`);
+          const via = detectedViaDbLock ? ' (detected via DB lock; pidfile not found at the configured path)' : '';
+          console.log(`Supervisor: ${running ? 'running' : 'not running'}${via}`);
+          if (status.supervisor_pid) console.log(`  PID:           ${status.supervisor_pid}${detectedViaDbLock ? ` @ ${dbLockHolder?.holder_host}` : ''}`);
           console.log(`  PID file:      ${pidFile}`);
+          if (detectedViaDbLock && status.concurrency !== null) console.log(`  Concurrency:   ${status.concurrency}${status.max_rss_mb !== null ? ` (max-rss ${status.max_rss_mb}MB)` : ''}`);
           if (lastStart) console.log(`  Last start:    ${lastStart}`);
           console.log(`  Crashes (24h):     ${summary.total} (runtime=${summary.by_cause.runtime_error} oom=${summary.by_cause.oom_or_external_kill} unknown=${summary.by_cause.unknown} legacy=${summary.by_cause.legacy})`);
           console.log(`  Clean exits (24h): ${summary.clean_exits}`);
@@ -1607,6 +1642,13 @@ export async function registerBuiltinHandlers(
     //     archived between fan-out and worker claim, skip cleanly.
     const rawSourceId = job.data.source_id;
     let sourceId: string | undefined;
+    // issue #2227/#2194 (TODOS:634, codex #8): a per-source cycle must run its
+    // FILESYSTEM phases (sync/lint/extract) against the SOURCE's own checkout,
+    // not the global brain's. Pre-fix it inherited `repoPath` (the default
+    // checkout) while writing DB freshness for `source_id` — mixed scope that
+    // made cooldown/freshness attribute to the wrong source. We resolve the
+    // source's `local_path` here and use it as the cycle's brainDir below.
+    let sourceLocalPath: string | null = null;
     if (rawSourceId !== undefined && rawSourceId !== null) {
       if (typeof rawSourceId !== 'string') {
         throw new Error(`autopilot-cycle: invalid source_id (not a string): ${JSON.stringify(rawSourceId)}`);
@@ -1620,9 +1662,10 @@ export async function registerBuiltinHandlers(
       }
       // Archive recheck (codex r1 P1-5): cheap pre-cycle lookup. Returns
       // immediately if source is gone or archived; runCycle never even
-      // acquires a lock.
-      const rows = await engine.executeRaw<{ archived: boolean | null }>(
-        `SELECT archived FROM sources WHERE id = $1`,
+      // acquires a lock. Also fetches local_path so FS phases bind to the
+      // source's own checkout (the #2227/#2194 mixed-scope fix).
+      const rows = await engine.executeRaw<{ archived: boolean | null; local_path: string | null }>(
+        `SELECT archived, local_path FROM sources WHERE id = $1`,
         [rawSourceId],
       );
       if (rows.length === 0) {
@@ -1640,8 +1683,17 @@ export async function registerBuiltinHandlers(
         };
       }
       sourceId = rawSourceId;
+      sourceLocalPath = typeof rows[0].local_path === 'string' && rows[0].local_path.length > 0
+        ? rows[0].local_path
+        : null;
     }
 
+    // Effective checkout for FS phases. For a per-source cycle, bind to the
+    // SOURCE's local_path (or null → skip FS phases for a pure-DB source);
+    // NEVER fall through to the global repoPath, which would run sync/lint
+    // against the wrong tree. Legacy (no source_id) keeps the global repoPath.
+    const effectiveBrainDir: string | null = sourceId ? sourceLocalPath : repoPath;
+
     // Allow callers to select phases via job data (e.g. skip embed for
     // fast cycles). Validates against ALL_PHASES to prevent injection.
     const { ALL_PHASES } = await import('../core/cycle.ts');
@@ -1653,8 +1705,23 @@ export async function registerBuiltinHandlers(
     // Pull default: legacy `true` for back-compat; explicit boolean wins.
     const pull = typeof job.data.pull === 'boolean' ? job.data.pull : true;
 
+    // #2194 fix #2 / codex #5 (D4): claim-time cooldown guard. A job already
+    // queued or retrying (max_attempts:2) can reach the worker after the
+    // dispatch gate decided to back this source off. Skip it here as a NO-OP
+    // (status 'skipped', NOT a failure — a failure would re-arm the cooldown).
+    if (sourceId) {
+      const { isSourceInCooldown } = await import('./autopilot-fanout.ts');
+      if (await isSourceInCooldown(engine, sourceId)) {
+        return {
+          partial: false,
+          status: 'skipped',
+          report: { reason: 'source_in_cooldown', source_id: sourceId },
+        };
+      }
+    }
+
     const report = await runCycle(engine, {
-      brainDir: repoPath,
+      brainDir: effectiveBrainDir,
       pull,
       signal: job.signal, // propagate abort so cycle bails on timeout/cancel
       ...(sourceId ? { sourceId } : {}),
@@ -1672,6 +1739,49 @@ export async function registerBuiltinHandlers(
     };
   });
 
+  // #2194 fix #3 / #2227 bug #3 — brain-wide maintenance. Runs the `global`
+  // cycle phases (embed, orphans, purge, resolve_symbol_edges, grade_takes,
+  // calibration_profile, synthesize_concepts, skillopt) ONCE per window instead
+  // of N times concurrently across per-source cycles (the 4→10GB RSS blowout).
+  // No source_id → uses the legacy global cycle lock; stamps autopilot.last_global_at
+  // on success so the dispatch gate backs off.
+  worker.register('autopilot-global-maintenance', async (job) => {
+    const { runCycle, GLOBAL_PHASES, LAST_GLOBAL_AT_KEY, ALL_PHASES } = await import('../core/cycle.ts');
+    const repoPath: string | null = typeof job.data.repoPath === 'string'
+      ? job.data.repoPath
+      : (await engine.getConfig('sync.repo_path')) ?? null;
+
+    const validPhases = new Set(ALL_PHASES);
+    const requested = Array.isArray(job.data.phases)
+      ? (job.data.phases as string[]).filter((p) => validPhases.has(p as never))
+      : GLOBAL_PHASES;
+    const phases = (requested.length > 0 ? requested : GLOBAL_PHASES) as typeof GLOBAL_PHASES;
+
+    const report = await runCycle(engine, {
+      brainDir: repoPath,
+      pull: false, // brain-wide DB/maintenance work never git-pulls
+      signal: job.signal,
+      phases,
+      yieldBetweenPhases: async () => { await new Promise<void>((r) => setImmediate(r)); },
+    });
+
+    // Stamp last_global_at only on a non-failed run so a failed pass stays stale
+    // and re-dispatches next tick (self-healing retry).
+    if (report.status === 'ok' || report.status === 'clean' || report.status === 'partial') {
+      try {
+        await engine.setConfig(LAST_GLOBAL_AT_KEY, new Date().toISOString());
+      } catch (e) {
+        console.warn(`[autopilot-global-maintenance] failed to stamp last_global_at: ${e instanceof Error ? e.message : String(e)}`);
+      }
+    }
+
+    return {
+      partial: report.status === 'partial' || report.status === 'failed',
+      status: report.status,
+      report,
+    };
+  });
+
   // Shell handler is always registered. Runtime env guard lives inside the
   // handler so claimed jobs emit a clear rejection log on workers missing
   // GBRAIN_ALLOW_SHELL_JOBS=1.

src/commands/sources.ts

@@ -692,7 +692,7 @@ async function runFederate(engine: BrainEngine, args: string[], value: boolean):
   const config = parseConfig(src.config);
   config.federated = value;
   await engine.executeRaw(
-    `UPDATE sources SET config = $1::jsonb WHERE id = $2`,
+    `UPDATE sources SET config = $1::text::jsonb WHERE id = $2`,
     [JSON.stringify(config), id],
   );
   console.log(`Source "${id}" is now ${value ? 'federated (appears in cross-source default search)' : 'isolated (only searched when explicitly named)'}.`);
@@ -747,8 +747,26 @@ async function runStatus(engine: BrainEngine, args: string[]): Promise<void> {
   // caught-up source reports lag 0 instead of growing wall-clock (v0.41.32.0).
   const metrics = await computeAllSourceMetrics(engine, sources, { probeContent: true });
 
+  // #1950: a source holding a live (non-TTL-expired) per-source sync lock is
+  // actively syncing RIGHT NOW. Without this it printed "idle" while a sync
+  // proc was live (the bug reported). Read the SAME live-lock signal `gbrain
+  // doctor` uses, via the shared helper, so the two surfaces never disagree.
+  const { liveSyncStatus } = await import('../core/db-lock.ts');
+  const syncRunning = new Map<string, { holder_pid: number; holder_host: string }>();
+  await Promise.all(
+    metrics.map(async (m) => {
+      const live = await liveSyncStatus(engine, m.source_id);
+      if (live) syncRunning.set(m.source_id, live);
+    }),
+  );
+
   if (json) {
-    console.log(JSON.stringify({ schema_version: 1, sources: metrics }, null, 2));
+    const enriched = metrics.map((m) => ({
+      ...m,
+      sync_running: syncRunning.has(m.source_id),
+      sync_holder: syncRunning.get(m.source_id) ?? null,
+    }));
+    console.log(JSON.stringify({ schema_version: 1, sources: enriched }, null, 2));
     return;
   }
 
@@ -765,11 +783,15 @@ async function runStatus(engine: BrainEngine, args: string[]): Promise<void> {
     const embed = `${m.embed_coverage_pct.toFixed(0)}%`;
     // v0.41.31: embed-backfill state (active beats queued beats idle) so a
     // cron operator sees deferred embedding work after `sync --all`.
-    const backfill = m.backfill_active > 0
-      ? `active(${m.backfill_active})`
-      : m.backfill_queued > 0
-        ? `queued(${m.backfill_queued})`
-        : 'idle';
+    // #1950: a live sync lock wins the column — surfacing "actively syncing"
+    // matters more than deferred embed-backfill state.
+    const backfill = syncRunning.has(m.source_id)
+      ? 'running'
+      : m.backfill_active > 0
+        ? `active(${m.backfill_active})`
+        : m.backfill_queued > 0
+          ? `queued(${m.backfill_queued})`
+          : 'idle';
     const fails = String(m.failed_jobs_24h);
     const queue = String(m.queue_depth);
     const pages = m.total_pages.toLocaleString();
@@ -780,7 +802,10 @@ async function runStatus(engine: BrainEngine, args: string[]): Promise<void> {
   for (const m of metrics) {
     const warns: string[] = [];
     if (!m.local_path) warns.push('no local_path');
-    if (m.lag_seconds === null) warns.push(`never synced — run \`gbrain sync --source ${m.source_id}\``);
+    // #1950: don't cry "never synced" while a sync lock is live — it's syncing now.
+    if (m.lag_seconds === null && !syncRunning.has(m.source_id)) {
+      warns.push(`never synced — run \`gbrain sync --source ${m.source_id}\``);
+    }
     if (m.embed_coverage_pct < 95 && m.total_chunks > 100) {
       warns.push(`${(100 - m.embed_coverage_pct).toFixed(1)}% un-embedded — run \`gbrain embed --stale --source ${m.source_id}\``);
     }
@@ -854,7 +879,7 @@ async function runWebhookSet(engine: BrainEngine, args: string[]): Promise<void>
   cfg.webhook_secret = secret;
   cfg.github_repo = githubRepo;
   await engine.executeRaw(
-    `UPDATE sources SET config = $1::jsonb WHERE id = $2`,
+    `UPDATE sources SET config = $1::text::jsonb WHERE id = $2`,
     [JSON.stringify(cfg), id],
   );
 
@@ -910,7 +935,7 @@ async function runWebhookRotate(engine: BrainEngine, args: string[]): Promise<vo
   const cfg = parseConfig(src.config);
   cfg.webhook_secret = secret;
   await engine.executeRaw(
-    `UPDATE sources SET config = $1::jsonb WHERE id = $2`,
+    `UPDATE sources SET config = $1::text::jsonb WHERE id = $2`,
     [JSON.stringify(cfg), id],
   );
   console.log(`New webhook secret for source "${id}":`);
@@ -934,7 +959,7 @@ async function runWebhookClear(engine: BrainEngine, args: string[]): Promise<voi
   delete cfg.webhook_secret;
   delete cfg.github_repo;
   await engine.executeRaw(
-    `UPDATE sources SET config = $1::jsonb WHERE id = $2`,
+    `UPDATE sources SET config = $1::text::jsonb WHERE id = $2`,
     [JSON.stringify(cfg), id],
   );
   console.log(`Webhook configuration cleared for source "${id}".`);
@@ -959,7 +984,7 @@ async function runTrackedBranch(engine: BrainEngine, args: string[]): Promise<vo
   if (setArg) {
     cfg.tracked_branch = setArg;
     await engine.executeRaw(
-      `UPDATE sources SET config = $1::jsonb WHERE id = $2`,
+      `UPDATE sources SET config = $1::text::jsonb WHERE id = $2`,
       [JSON.stringify(cfg), id],
     );
     console.log(`Tracked branch for source "${id}" set to "${setArg}".`);
@@ -975,7 +1000,7 @@ async function runTrackedBranch(engine: BrainEngine, args: string[]): Promise<vo
       const branch = execFileSync('git', ['-C', src.local_path, 'rev-parse', '--abbrev-ref', 'HEAD'], { encoding: 'utf8' }).trim();
       cfg.tracked_branch = branch;
       await engine.executeRaw(
-        `UPDATE sources SET config = $1::jsonb WHERE id = $2`,
+        `UPDATE sources SET config = $1::text::jsonb WHERE id = $2`,
         [JSON.stringify(cfg), id],
       );
       console.log(`Detected branch "${branch}" for source "${id}"; persisted to config.tracked_branch.`);

src/commands/status.ts

@@ -39,6 +39,7 @@ import type { BrainEngine } from '../core/engine.ts';
 import { existsSync, readFileSync } from 'node:fs';
 import { gbrainPath, loadConfig, isThinClient } from '../core/config.ts';
 import { callRemoteTool, unpackToolResult } from '../core/mcp-client.ts';
+import { VERSION } from '../version.ts';
 import {
   buildSyncStatusReport,
   type SyncStatusReport,
@@ -104,6 +105,10 @@ export interface AutopilotStatus {
 
 export interface StatusReport {
   schema_version: typeof SCHEMA_VERSION;
+  /** #1984: the local gbrain CLI version, so a poller can pin behavior to a build. */
+  version: string;
+  /** #1984: the remote brain server's version (thin-client only; present when reported). */
+  remote_version?: string;
   generated_at: string;
   mode: 'local' | 'thin-client';
   sync?: SyncStatusReport;
@@ -113,6 +118,33 @@ export interface StatusReport {
   queue?: QueueCounts | { local_only_remote: true };
   autopilot?: AutopilotStatus | { local_only_remote: true };
   warnings?: string[];
+  /** #1984: true when a --deadline-ms budget elided one or more sections. */
+  partial?: boolean;
+  /** #1984: names of the sections skipped/timed-out under the deadline. */
+  stale_sections?: string[];
+}
+
+/**
+ * #1984: race a section's work against the remaining deadline budget. Returns
+ * the value, or undefined if the budget elapses first (the underlying query is
+ * abandoned — the process is a one-shot, so a stranded read is fine). `ms`
+ * undefined / <=0 means "no deadline" and the promise is awaited as-is.
+ */
+export async function withSectionDeadline<T>(
+  p: Promise<T>,
+  ms: number | undefined,
+  onTimeout: () => void,
+): Promise<T | undefined> {
+  if (ms === undefined || ms <= 0) return p;
+  let timer: ReturnType<typeof setTimeout> | undefined;
+  const timeout = new Promise<undefined>((resolve) => {
+    timer = setTimeout(() => { onTimeout(); resolve(undefined); }, ms);
+  });
+  try {
+    return await Promise.race([p, timeout]);
+  } finally {
+    if (timer) clearTimeout(timer);
+  }
 }
 
 // ---------------------------------------------------------------------------
@@ -305,6 +337,8 @@ function buildAutopilotStatus(): AutopilotStatus {
 
 interface BuildOpts {
   sections?: Set<Section>;
+  /** #1984: total wall-clock budget; sections share it, later ones get the remainder. */
+  deadlineMs?: number;
 }
 
 async function buildLocalReport(
@@ -313,23 +347,44 @@ async function buildLocalReport(
 ): Promise<StatusReport> {
   const want = (s: Section) => !opts.sections || opts.sections.has(s);
   const warnings: string[] = [];
+  const staleSections: string[] = [];
   const report: StatusReport = {
     schema_version: SCHEMA_VERSION,
+    version: VERSION,
     generated_at: new Date().toISOString(),
     mode: 'local',
   };
 
+  // #1984: a shared budget so `gbrain status --deadline-ms=N` never blocks a
+  // poller past N. Each async section is raced against the REMAINING budget, so
+  // one slow/hung section (cross-region DB, lock contention) can't strand the
+  // whole snapshot — it's marked stale and the rest still return.
+  const deadlineAt = opts.deadlineMs && opts.deadlineMs > 0 ? Date.now() + opts.deadlineMs : null;
+  const remaining = (): number | undefined =>
+    deadlineAt === null ? undefined : Math.max(1, deadlineAt - Date.now());
+  const markStale = (name: Section) => {
+    staleSections.push(name);
+    report.partial = true;
+    warnings.push(`${name} section exceeded the --deadline-ms budget (returned stale)`);
+  };
+
   if (want('sync')) {
     try {
-      const sources = await engine.executeRaw<{
-        id: string;
-        name: string;
-        local_path: string | null;
-        config: Record<string, unknown> | null;
-      }>(`SELECT id, name, local_path, config FROM sources ORDER BY id`);
-      report.sync = await buildSyncStatusReport(
-        engine,
-        sources.map((s) => ({ id: s.id, name: s.name, local_path: s.local_path, config: s.config ?? {} })),
+      report.sync = await withSectionDeadline(
+        (async () => {
+          const sources = await engine.executeRaw<{
+            id: string;
+            name: string;
+            local_path: string | null;
+            config: Record<string, unknown> | null;
+          }>(`SELECT id, name, local_path, config FROM sources ORDER BY id`);
+          return buildSyncStatusReport(
+            engine,
+            sources.map((s) => ({ id: s.id, name: s.name, local_path: s.local_path, config: s.config ?? {} })),
+          );
+        })(),
+        remaining(),
+        () => markStale('sync'),
       );
     } catch (err) {
       warnings.push(`sync section failed: ${(err as Error).message}`);
@@ -337,23 +392,24 @@ async function buildLocalReport(
   }
   if (want('cycle')) {
     try {
-      report.cycle = await buildCycleSnapshot(engine);
+      report.cycle = await withSectionDeadline(buildCycleSnapshot(engine), remaining(), () => markStale('cycle'));
     } catch (err) {
       warnings.push(`cycle section failed: ${(err as Error).message}`);
     }
   }
   if (want('locks')) {
-    report.locks = await buildLocks(engine);
+    report.locks = await withSectionDeadline(buildLocks(engine), remaining(), () => markStale('locks'));
   }
   if (want('workers')) {
     report.workers = buildWorkerSummary();
   }
   if (want('queue')) {
-    report.queue = await buildQueueCounts(engine);
+    report.queue = await withSectionDeadline(buildQueueCounts(engine), remaining(), () => markStale('queue'));
   }
   if (want('autopilot')) {
     report.autopilot = buildAutopilotStatus();
   }
+  if (staleSections.length > 0) report.stale_sections = staleSections;
   if (warnings.length > 0) report.warnings = warnings;
   return report;
 }
@@ -366,20 +422,51 @@ async function buildThinClientReport(
   const warnings: string[] = [];
   const report: StatusReport = {
     schema_version: SCHEMA_VERSION,
+    version: VERSION,
     generated_at: new Date().toISOString(),
     mode: 'thin-client',
   };
 
   if (want('sync') || want('cycle')) {
     try {
-      const raw = await callRemoteTool(cfg!, 'get_status_snapshot', {});
-      const payload = unpackToolResult<{
-        schema_version: number;
-        sync: SyncStatusReport;
-        cycle: CycleSnapshot;
-      }>(raw);
-      if (want('sync')) report.sync = payload.sync;
-      if (want('cycle')) report.cycle = payload.cycle;
+      const payload = await withSectionDeadline(
+        (async () => {
+          // #1984: pass the budget as the request timeout so the LOSING side of
+          // the race actually cancels the in-flight MCP call instead of leaking
+          // it (the section deadline only abandons the promise locally).
+          const raw = await callRemoteTool(
+            cfg!,
+            'get_status_snapshot',
+            {},
+            opts.deadlineMs && opts.deadlineMs > 0 ? { timeoutMs: opts.deadlineMs } : {},
+          );
+          return unpackToolResult<{
+            schema_version: number;
+            version?: string;
+            sync: SyncStatusReport;
+            cycle: CycleSnapshot;
+          }>(raw);
+        })(),
+        opts.deadlineMs && opts.deadlineMs > 0 ? opts.deadlineMs : undefined,
+        () => {
+          report.partial = true;
+          // Only name the sections the caller actually requested; the remote
+          // fetch backs both sync+cycle, but `--section sync` must not report
+          // `cycle` (a section it excluded) as stale. Matches the local path.
+          const elided: Section[] = [
+            ...(want('sync') ? (['sync'] as Section[]) : []),
+            ...(want('cycle') ? (['cycle'] as Section[]) : []),
+          ];
+          report.stale_sections = [...(report.stale_sections ?? []), ...elided];
+          warnings.push('remote snapshot exceeded the --deadline-ms budget (returned stale)');
+        },
+      );
+      if (payload) {
+        // #1984: surface the brain server's version for thin-client parity.
+        if (payload.version) report.remote_version = payload.version;
+        if (want('sync')) report.sync = payload.sync;
+        if (want('cycle')) report.cycle = payload.cycle;
+      }
     } catch (err) {
       warnings.push(`remote snapshot failed: ${(err as Error).message}`);
     }
@@ -401,7 +488,13 @@ function renderHuman(report: StatusReport): string {
   lines.push('');
   lines.push('GBrain Status');
   lines.push('=============');
-  lines.push(`Mode: ${report.mode}  ·  ${report.generated_at}`);
+  const ver = report.remote_version && report.remote_version !== report.version
+    ? `v${report.version} (remote v${report.remote_version})`
+    : `v${report.version}`;
+  lines.push(`Mode: ${report.mode}  ·  ${ver}  ·  ${report.generated_at}`);
+  if (report.partial) {
+    lines.push(`⚠ partial snapshot — stale sections: ${(report.stale_sections ?? []).join(', ')} (--deadline-ms budget hit)`);
+  }
   lines.push('');
 
   // Sync
@@ -528,6 +621,36 @@ function renderHuman(report: StatusReport): string {
  *   - Set<Section> → only these sections
  *   - 'usage_error' → bad section name (caller exits 2)
  */
+/** #1984: `--fast` preset budget (ms) when no explicit `--deadline-ms` is given. */
+export const FAST_DEADLINE_MS = 2000;
+
+/**
+ * #1984: parse the status deadline budget. `--deadline-ms=N` (or `--deadline-ms N`)
+ * wins; bare `--fast` applies FAST_DEADLINE_MS. Returns undefined (no budget),
+ * a positive ms value, or 'usage_error' on a non-positive / non-numeric value.
+ */
+export function parseDeadlineFlag(args: string[]): number | undefined | 'usage_error' {
+  let raw: string | undefined;
+  for (let i = 0; i < args.length; i++) {
+    const a = args[i];
+    if (a === '--deadline-ms') {
+      // #1984: a bare `--deadline-ms` with no following value is a usage error,
+      // not a silent fall-through to no-budget / --fast (which would mask a
+      // typo'd budget and let a poller hang it never meant to).
+      if (i + 1 >= args.length) return 'usage_error';
+      raw = args[i + 1];
+      break;
+    }
+    if (a.startsWith('--deadline-ms=')) { raw = a.slice('--deadline-ms='.length); break; }
+  }
+  if (raw == null) {
+    return args.includes('--fast') ? FAST_DEADLINE_MS : undefined;
+  }
+  const n = Number(raw);
+  if (!Number.isFinite(n) || n <= 0) return 'usage_error';
+  return n;
+}
+
 export function parseSectionFlag(args: string[]): Set<Section> | undefined | 'usage_error' {
   let raw: string | undefined;
   for (let i = 0; i < args.length; i++) {
@@ -575,19 +698,25 @@ export async function runStatus(
   const sections = sectionFlag;
   const json = args.includes('--json');
 
+  const deadlineMs = parseDeadlineFlag(args);
+  if (deadlineMs === 'usage_error') {
+    stderr('gbrain status: --deadline-ms must be a positive number of milliseconds\n');
+    return { exitCode: 2 };
+  }
+
   const cfg = loadConfig();
   const useThinClient = cfg ? isThinClient(cfg) : false;
 
   let report: StatusReport;
   try {
     if (useThinClient) {
-      report = await buildThinClientReport(cfg, { sections });
+      report = await buildThinClientReport(cfg, { sections, deadlineMs });
     } else {
       if (!engine) {
         stderr('gbrain status: no engine connected (DB unreachable?). Run `gbrain doctor` to diagnose.\n');
         return { exitCode: 1 };
       }
-      report = await buildLocalReport(engine, { sections });
+      report = await buildLocalReport(engine, { sections, deadlineMs });
     }
   } catch (err) {
     stderr(`gbrain status: snapshot failed: ${(err as Error).message}\n`);

src/commands/sync.ts

@@ -212,7 +212,7 @@ export interface SyncResult {
    * cron operators can disambiguate timeout vs pull-timeout in monitoring.
    */
   filesImported?: number;
-  reason?: 'timeout' | 'pull_timeout' | 'checkpoint_unavailable';
+  reason?: 'timeout' | 'pull_timeout' | 'stall_timeout' | 'checkpoint_unavailable';
   /**
    * v0.42.x (#1794): cumulative file paths durably banked to the checkpoint
    * across THIS run + prior resumed runs. Surfaced on every partial/blocked
@@ -1440,7 +1440,7 @@ function buildPartialResult(opts: {
   modified: number;
   deleted: number;
   renamed: number;
-  reason: 'timeout' | 'pull_timeout' | 'checkpoint_unavailable';
+  reason: 'timeout' | 'pull_timeout' | 'stall_timeout' | 'checkpoint_unavailable';
   bankedFiles?: number;
 }): SyncResult {
   return {
@@ -2050,7 +2050,7 @@ async function performSyncInner(engine: BrainEngine, opts: SyncOpts): Promise<Sy
   // `bankedFiles` is surfaced so a killed run shows banked progress instead of
   // looking like total loss. toCommit reports the PINNED target; last_commit is
   // never advanced on a partial (the next run resumes from the checkpoint).
-  const partial = async (reason: 'timeout' | 'pull_timeout'): Promise<SyncResult> => {
+  const partial = async (reason: 'timeout' | 'pull_timeout' | 'stall_timeout'): Promise<SyncResult> => {
     deregisterCheckpointCleanup();
     if (!checkpointDead) {
       try { await flushCheckpoint(); } catch { /* best effort — we're aborting */ }
@@ -2409,6 +2409,48 @@ async function performSyncInner(engine: BrainEngine, opts: SyncOpts): Promise<Sy
       pacer = createNoopPacer();
     }
 
+    // #1950: progress-aware stall watchdog for the import drain. The incident
+    // was a sync wedged ~29min while ALIVE — so the lock heartbeat kept
+    // refreshing (it fires on its own timer) and the wall-clock deadline hadn't
+    // hit yet, leaving only a manual `pkill`. This keys off FORWARD IMPORT
+    // PROGRESS (progress.tick below bumps `progressAt`), not the heartbeat: if no
+    // file completes for `resolveStallAbortSeconds()`, abort. The abort signal is
+    // composed into `opts.signal`, so the existing per-iteration abort checks,
+    // pacer.acquire/pace, and parallel-worker break-loops all observe it; the
+    // drain returns partial() (last_commit unchanged, next run resumes from the
+    // checkpoint) and withRefreshingLock's finally releases the lock. Limits
+    // (TODOS: #1950 follow-up — thread a cancellation signal through importFile):
+    // the abort is observed BETWEEN files (the per-iteration checks + the next
+    // importOnePath's pre-acquire check), so a hang INSIDE a single importFile
+    // call is not interrupted until that call returns — the watchdog fires and
+    // logs, but the in-flight file finishes (or the wall-clock hard deadline is
+    // the eventual backstop). This catches the documented #1950 incident shape
+    // (a slow-but-progressing drain, many files) and a stalled between-file
+    // drain; a single wedged file or a fully starved event loop is out of scope
+    // here. stallAborted distinguishes this from a user --timeout/SIGINT so the
+    // partial result reports `stall_timeout`, not `timeout`.
+    const stallSeconds = resolveStallAbortSeconds();
+    const progressAt = { last: Date.now() };
+    let stallAborted = false;
+    let stallTimer: ReturnType<typeof setInterval> | undefined;
+    if (stallSeconds > 0) {
+      const stallMs = stallSeconds * 1000;
+      const stallController = new AbortController();
+      stallTimer = setInterval(() => {
+        if (Date.now() - progressAt.last >= stallMs) {
+          serr(
+            `[sync] no import progress for ${stallSeconds}s — aborting (stall watchdog). ` +
+            `The per-source lock will release; the next 'gbrain sync' resumes from the checkpoint.`,
+          );
+          stallAborted = true;
+          stallController.abort();
+        }
+      }, Math.min(5000, stallMs));
+      // Don't keep the process alive on the watchdog alone.
+      (stallTimer as unknown as { unref?: () => void }).unref?.();
+      opts = { ...opts, signal: composeAbortSignals(opts.signal, stallController.signal) };
+    }
+
     async function importOnePath(eng: BrainEngine, path: string): Promise<void> {
       const filePath = join(syncRepoPath, path);
       if (!existsSync(filePath)) {
@@ -2429,6 +2471,7 @@ async function performSyncInner(engine: BrainEngine, opts: SyncOpts): Promise<Sy
         // its row so it can't age doctor to a permanent FAIL. (This covers the
         // net-zero add-then-delete range where the path isn't in filtered.deleted.)
         succeededPaths.push(path);
+        progressAt.last = Date.now(); // #1950: forward progress → reset stall watchdog
         progress.tick(1, `skip:${path}`);
         return;
       }
@@ -2484,6 +2527,7 @@ async function performSyncInner(engine: BrainEngine, opts: SyncOpts): Promise<Sy
       } finally {
         permit.release();
       }
+      progressAt.last = Date.now(); // #1950: forward progress → reset stall watchdog
       progress.tick(1, path);
       // v0.42.x (#1794): keep the lock-refresh heartbeat alive on big imports.
       await maybeYield();
@@ -2510,7 +2554,7 @@ async function performSyncInner(engine: BrainEngine, opts: SyncOpts): Promise<Sy
           // serial fallback inside the parallel branch (database_url unset).
           if (opts.signal?.aborted) {
             progress.finish();
-            return await partial('timeout');
+            return await partial(stallAborted ? 'stall_timeout' : 'timeout');
           }
           await importOnePath(engine, path);
         }
@@ -2574,7 +2618,7 @@ async function performSyncInner(engine: BrainEngine, opts: SyncOpts): Promise<Sy
         // primary serial site.
         if (opts.signal?.aborted) {
           progress.finish();
-          return await partial('timeout');
+          return await partial(stallAborted ? 'stall_timeout' : 'timeout');
         }
         await importOnePath(engine, path);
       }
@@ -2583,6 +2627,10 @@ async function performSyncInner(engine: BrainEngine, opts: SyncOpts): Promise<Sy
       // paced-backfill: release any blocked acquirers + clear pacer state on
       // every exit path (including the early partial('timeout') returns above).
       pacer.dispose();
+      // #1950: tear down the stall watchdog on every import-phase exit (normal,
+      // partial('timeout'), or throw). The try wrapping the import loop guarantees
+      // this runs before any post-import bookmark/anchor work.
+      if (stallTimer) clearInterval(stallTimer);
     }
 
     progress.finish();
@@ -2594,7 +2642,7 @@ async function performSyncInner(engine: BrainEngine, opts: SyncOpts): Promise<Sy
     // the bookmark write below. By returning partial here, we preserve
     // the D-V3-1 invariant that abort means "never advance last_commit."
     if (opts.signal?.aborted) {
-      return await partial('timeout');
+      return await partial(stallAborted ? 'stall_timeout' : 'timeout');
     }
   }
 
@@ -3144,6 +3192,26 @@ export interface HardDeadlineResolution {
   reason: string;
 }
 
+/**
+ * #1950: default no-import-progress window before the in-band stall watchdog
+ * aborts the drain. Generous on purpose — it must clear one legitimately large
+ * file (cross-region, big page) without false-tripping; one file taking longer
+ * than this trips it (documented limit). Distinct from the wall-clock hard
+ * deadline (whole-run cap) and the lock heartbeat (refreshes regardless of
+ * import progress). Env-tunable; <=0 disables.
+ */
+export const DEFAULT_SYNC_STALL_ABORT_SEC = 900;
+
+export function resolveStallAbortSeconds(
+  env: Record<string, string | undefined> = process.env,
+): number {
+  const raw = env.GBRAIN_SYNC_STALL_ABORT_SECONDS;
+  if (raw === undefined || raw === '') return DEFAULT_SYNC_STALL_ABORT_SEC;
+  const n = Number(raw);
+  if (!Number.isFinite(n)) return DEFAULT_SYNC_STALL_ABORT_SEC;
+  return n; // n <= 0 disables the watchdog
+}
+
 /**
  * Resolve the out-of-band hard-deadline for a `gbrain sync` invocation (#1633).
  * Pure + argv/env-only so it runs BEFORE `connectEngine` (so a connect-phase hang

src/core/code-intel/traversal-cache.ts

@@ -123,7 +123,7 @@ export async function putCachedTraversal<T>(
       `INSERT INTO code_traversal_cache
          (symbol_qualified, depth, source_id, response_json,
           max_chunk_updated_at, xmin_max, cluster_generation)
-       VALUES ($1, $2, $3, $4::jsonb, $5::timestamptz, $6, $7)
+       VALUES ($1, $2, $3, $4::text::jsonb, $5::timestamptz, $6, $7)
        ON CONFLICT (symbol_qualified, depth, source_id)
        DO UPDATE SET
          response_json = EXCLUDED.response_json,

src/core/conversation-parser/llm-base.ts

@@ -266,13 +266,13 @@ async function writeDbCache<T>(
 ): Promise<void> {
   const [, , contentSha] = splitCacheKey(key);
   if (!contentSha) return;
-  // executeRaw with positional binding for JSONB. Per the sql-query.ts
-  // contract: object values passed via positional params reach the
-  // wire as proper jsonb when cast.
+  // #2339 class: this binds JSON.stringify(value) (a STRING) positionally, so it
+  // must cast through $4::text::jsonb — a bare $4::jsonb double-encodes under
+  // postgres.js .unsafe() (PGLite hides it). Pass a raw object to use $N::jsonb.
   await engine.executeRaw(
     `INSERT INTO conversation_parser_llm_cache
        (content_sha256, model_id, call_shape, value_json)
-     VALUES ($1, $2, $3, $4::jsonb)
+     VALUES ($1, $2, $3, $4::text::jsonb)
      ON CONFLICT (content_sha256, model_id, call_shape) DO NOTHING`,
     [contentSha, modelStr, shape, JSON.stringify(value)],
   );

src/core/cycle.ts

@@ -242,6 +242,25 @@ export const PHASE_SCOPE: Record<CyclePhase, PhaseScope> = {
   skillopt: 'global',
 };
 
+/**
+ * #2194 fix #3 / #2227 bug #3 — the cycle split.
+ *
+ * Per-source autopilot cycles run ONLY the source-scoped (and mixed) phases;
+ * the brain-wide `global` phases (embed, orphans, purge, resolve_symbol_edges,
+ * grade_takes, calibration_profile, synthesize_concepts, skillopt) run ONCE in
+ * a separate `autopilot-global-maintenance` job instead of N times concurrently
+ * across per-source cycles (the 4→10GB RSS blowout). Single-flight is
+ * structural: one global job, not a skip-and-pretend-fresh hack (codex #1/#2).
+ *
+ * GLOBAL_PHASES ∪ NON_GLOBAL_PHASES == ALL_PHASES, with no overlap — pinned by
+ * test/autopilot-global-maintenance.test.ts.
+ */
+export const GLOBAL_PHASES: CyclePhase[] = ALL_PHASES.filter((p) => PHASE_SCOPE[p] === 'global');
+export const NON_GLOBAL_PHASES: CyclePhase[] = ALL_PHASES.filter((p) => PHASE_SCOPE[p] !== 'global');
+
+/** Config key holding the ISO timestamp of the last successful global-maintenance run. */
+export const LAST_GLOBAL_AT_KEY = 'autopilot.last_global_at';
+
 /**
  * Phases that mutate state (filesystem or DB) and therefore should
  * coordinate via the cycle lock. Only orphans is truly read-only
@@ -2305,12 +2324,21 @@ export async function runCycle(
   // the cost of missing a successful write (next cycle will redo work).
   if (opts.sourceId && engine && !dryRun && !aborted && (status === 'ok' || status === 'clean' || status === 'partial')) {
     try {
+      const nowIso = new Date().toISOString();
+      // #2194 fix #3 (the cycle split): `last_source_cycle_at` is the NEW gate
+      // for per-source dispatch (source-scoped phases done). We ALSO keep
+      // `last_full_cycle_at` current so doctor's cycle-freshness check and any
+      // legacy reader stay valid — it's no longer a *gate* for the brain-wide
+      // phases (those gate on autopilot.last_global_at), so writing it on a
+      // source-only cycle does not re-introduce the freshness poisoning codex
+      // flagged in the rejected skip-based design.
       await engine.updateSourceConfig(opts.sourceId, {
-        last_full_cycle_at: new Date().toISOString(),
+        last_source_cycle_at: nowIso,
+        last_full_cycle_at: nowIso,
       });
     } catch (e) {
       // Best-effort; cycle already succeeded by the time we get here.
-      console.warn(`[cycle] failed to write last_full_cycle_at for source ${opts.sourceId}: ${e instanceof Error ? e.message : String(e)}`);
+      console.warn(`[cycle] failed to write last_source_cycle_at for source ${opts.sourceId}: ${e instanceof Error ? e.message : String(e)}`);
     }
   }
 

src/core/cycle/calibration-profile.ts

@@ -356,7 +356,7 @@ class CalibrationProfilePhase extends BaseCyclePhase {
          active_bias_tags, model_id, cost_usd, judge_model_agreement
        ) VALUES ($1, $2, now(), false,
                  $3, $4, $5, $6, $7,
-                 $8::jsonb, $9::text[],
+                 $8::text::jsonb, $9::text[],
                  $10, $11,
                  $12::text[], $13, NULL, NULL)`,
       [

src/core/db-lock.ts

@@ -132,6 +132,28 @@ export function isHolderDeadLocally(
   return classifyHolderLiveness(holderPid, holderHost, ageMs, opts) === 'dead_eligible';
 }
 
+/**
+ * issue #2227: is a lock-row holder live enough to count as "running" for an
+ * observability surface (`gbrain jobs supervisor status`, `gbrain doctor`)?
+ *
+ * PID-reuse-safe by design (`pid-liveness-alone-pid-reuse`): keys on the lock's
+ * own freshness, NEVER `process.kill`. A live holder refreshes its TTL on a
+ * timer; a dead one stops, so its `ttl_expires_at` lapses and (after the steal
+ * grace) `last_refreshed_at` ages out. The primary signal is `!ttl_expired`;
+ * the heartbeat grace covers a starved-but-alive holder whose TTL briefly
+ * lapsed between refresh ticks (the #1794 thrash class), mirroring the
+ * steal-grace semantics `tryAcquireDbLock` already uses. Cross-host holders are
+ * still "running" for visibility (a supervisor exists, just elsewhere) — we
+ * report freshness, not takeover-eligibility, so host is not consulted here.
+ */
+export function isLockHolderLive(snap: LockSnapshot, ttlMinutes: number = DEFAULT_TTL_MINUTES): boolean {
+  if (!snap.ttl_expired) return true;
+  if (snap.ms_since_last_refresh !== null) {
+    return snap.ms_since_last_refresh < resolveStealGraceSeconds(ttlMinutes) * 1000;
+  }
+  return false;
+}
+
 /**
  * Try to acquire a named DB lock.
  *
@@ -707,6 +729,38 @@ export function syncLockId(sourceId: string): string {
  */
 export const SYNC_LOCK_ID = syncLockId('default');
 
+/**
+ * #1950: is `sourceId` actively holding a live (non-TTL-expired) sync lock?
+ *
+ * Centralizes the live-sync signal that `gbrain doctor` already computes inline
+ * so `gbrain sources status` (and future surfaces) read the SAME truth instead
+ * of each re-deriving it. Returns the holder when a live lock is held, else
+ * null (idle, or a stale/expired lock that's structurally available for the
+ * next acquire). Inspect failures swallow to null — a status surface should
+ * degrade to "no indicator", never crash.
+ *
+ * Honest scope: a live lock proves the holder process is heartbeating, NOT that
+ * the import is making forward progress — `withRefreshingLock` refreshes
+ * `last_refreshed_at` on its own timer regardless of import progress. So callers
+ * report "running", NOT "healthy"; a wedged-but-alive holder still reads as
+ * running here. Forward-progress stall detection lives in the sync drain loop
+ * (#1950 stall-abort); stale-lock triage lives in `gbrain doctor`.
+ */
+export async function liveSyncStatus(
+  engine: BrainEngine,
+  sourceId: string,
+): Promise<{ holder_pid: number; holder_host: string } | null> {
+  try {
+    const snap = await inspectLock(engine, syncLockId(sourceId));
+    if (snap && !snap.ttl_expired) {
+      return { holder_pid: snap.holder_pid, holder_host: snap.holder_host };
+    }
+    return null;
+  } catch {
+    return null;
+  }
+}
+
 /**
  * v0.30.1 (T4 + A4): wrap long-running work in a refreshing TTL lock.
  *

src/core/doctor-categories.ts

@@ -128,6 +128,7 @@ export const SKILL_CHECK_NAMES: ReadonlySet<string> = new Set([
  */
 export const OPS_CHECK_NAMES: ReadonlySet<string> = new Set([
   'alternative_providers',
+  'autopilot_fanout_concurrency',
   'autopilot_lock_scope',
   'batch_retry_health',
   'brainstorm_health',

src/core/eval-contradictions/runner.ts

@@ -171,11 +171,19 @@ async function generateIntraPagePairs(
   results: SearchResult[],
 ): Promise<ContradictionPair[]> {
   if (results.length === 0) return [];
-  // Unique page_ids only.
-  const pageIds = Array.from(new Set(results.map((r) => r.page_id)));
+  // Unique, FINITE page_ids only. Defensive backstop for the alias-hop bug
+  // (#2339 sibling): an alias-injected synthetic result with an undefined/NaN
+  // page_id must never reach `ANY($1::int[])` — postgres.js rejects it with
+  // UNDEFINED_VALUE and aborts the whole probe. Mirrors the hybrid.ts:63 filter.
+  const pageIds = Array.from(
+    new Set(
+      results.map((r) => r.page_id).filter((n): n is number => typeof n === 'number' && Number.isFinite(n)),
+    ),
+  );
   const takesByPage = await engine.listActiveTakesForPages(pageIds);
   const out: ContradictionPair[] = [];
   for (const r of results) {
+    if (typeof r.page_id !== 'number' || !Number.isFinite(r.page_id)) continue;
     const takes = takesByPage.get(r.page_id) ?? [];
     if (takes.length === 0) continue;
     const chunkMember = searchResultToMember(r);

src/core/minions/child-worker-supervisor.ts

@@ -61,9 +61,12 @@ export type ChildSupervisorEvent =
     }
   | {
       kind: 'health_warn';
-      reason: 'clean_restart_budget_exceeded' | 'rss_watchdog_loop';
+      reason: 'clean_restart_budget_exceeded' | 'rss_watchdog_loop' | 'crash_budget_degraded';
       count: number;
-      windowMs: number;
+      /** Present for window-scoped warnings (budget/watchdog loops). */
+      windowMs?: number;
+      /** Present for crash_budget_degraded: the soft budget that was crossed. */
+      max?: number;
     };
 
 export interface ChildWorkerSupervisorOpts {
@@ -73,8 +76,24 @@ export interface ChildWorkerSupervisorOpts {
   args: string[];
   /** Child env. Defaults to a clone of process.env. */
   env?: NodeJS.ProcessEnv;
-  /** Give up after this many consecutive code != 0 exits. */
+  /**
+   * Soft crash budget. issue #1994 (#2227 tail): crossing this NO LONGER
+   * permanently gives up. Instead the supervisor enters DEGRADED mode — it
+   * keeps respawning with capped exponential backoff (60s cap) and emits a
+   * loud `crash_budget_degraded` health_warn — so a transient DB-pooler outage
+   * that trips the counter self-heals when the DB returns (the stable-run reset
+   * clears crashCount once a respawn runs > stableRunResetMs) instead of
+   * wedging the queue until a human restart.
+   */
   maxCrashes: number;
+  /**
+   * Hard ceiling: permanently give up (fire onMaxCrashesExceeded) ONLY after
+   * this many consecutive crashes — the runaway backstop for a genuinely
+   * unrecoverable hot crash-loop that the stable-run reset never escapes.
+   * Default: maxCrashes * HARD_STOP_CRASH_MULTIPLIER. Set to 0 to disable
+   * permanent give-up entirely (retry-forever-with-backoff).
+   */
+  hardStopMaxCrashes?: number;
   /** Stable-run reset window: code != 0 after this duration resets crashCount to 1. Default 5 min. */
   stableRunResetMs?: number;
 
@@ -130,6 +149,12 @@ export interface ChildWorkerSupervisorOpts {
   _now?: () => number;
 }
 
+/** issue #1994: how many multiples of the soft crash budget before the hard
+ *  permanent-give-up backstop fires. 10× the default soft budget of 10 = 100
+ *  consecutive crashes (each within the stable-run window) before we conclude
+ *  it's a genuine code bug and stop. A transient outage recovers long before. */
+export const HARD_STOP_CRASH_MULTIPLIER = 10;
+
 const DEFAULTS = {
   stableRunResetMs: 5 * 60 * 1000,
   cleanRestartBudget: 10,
@@ -287,19 +312,50 @@ export class ChildWorkerSupervisor {
   /**
    * Run the spawn-and-respawn loop. Resolves when:
    *   1. composer.isStopping() returns true, OR
-   *   2. crashCount reaches maxCrashes (after firing onMaxCrashesExceeded).
+   *   2. crashCount reaches the HARD ceiling (after firing onMaxCrashesExceeded).
+   *
+   * issue #1994 (#2227 tail): crossing the SOFT budget (`maxCrashes`) no longer
+   * stops the supervisor. It enters degraded mode — keep respawning with capped
+   * exponential backoff (60s cap, so it's a paced retry, not a hot loop) and
+   * announce loudly — so a transient DB-pooler outage that trips the counter
+   * recovers on its own (a respawn that runs > stableRunResetMs resets
+   * crashCount to 1) instead of permanently wedging the queue. Permanent
+   * give-up fires only at the much-higher hard ceiling: the runaway backstop
+   * for a genuinely unrecoverable hot crash-loop.
    */
   async run(): Promise<void> {
-    while (!this.opts.isStopping() && this._crashCount < this.opts.maxCrashes) {
+    const hardStop = this.opts.hardStopMaxCrashes ??
+      this.opts.maxCrashes * HARD_STOP_CRASH_MULTIPLIER;
+    let degradedAnnounced = false;
+    while (!this.opts.isStopping()) {
       await this.spawnOnce();
 
       if (this.opts.isStopping()) return;
 
-      if (this._crashCount >= this.opts.maxCrashes) {
-        this.opts.onMaxCrashesExceeded(this._crashCount, this.opts.maxCrashes);
+      // Hard ceiling: permanent give-up (the runaway backstop). hardStop <= 0
+      // disables it entirely (retry-forever-with-backoff for deployments that
+      // would rather never auto-stop a recoverable supervisor).
+      if (hardStop > 0 && this._crashCount >= hardStop) {
+        this.opts.onMaxCrashesExceeded(this._crashCount, hardStop);
         return;
       }
 
+      // Soft budget crossed → degraded mode. Announce once per degradation
+      // episode; re-arm after a stable-run reset drops us back under budget.
+      if (this._crashCount >= this.opts.maxCrashes) {
+        if (!degradedAnnounced) {
+          degradedAnnounced = true;
+          this.opts.onEvent({
+            kind: 'health_warn',
+            reason: 'crash_budget_degraded',
+            count: this._crashCount,
+            max: this.opts.maxCrashes,
+          });
+        }
+      } else {
+        degradedAnnounced = false;
+      }
+
       await this.applyBackoff();
     }
   }

src/core/minions/handler-timeouts.ts

@@ -34,6 +34,9 @@ export const HANDLER_DEFAULT_TIMEOUT_MS: Readonly<Record<string, number>> = {
   subagent_aggregator: THIRTY_MIN_MS,
   'embed-backfill': THIRTY_MIN_MS,
   'autopilot-cycle': THIRTY_MIN_MS,
+  // #2194 fix #3: brain-wide maintenance (embed-all/orphans/purge/…) can run
+  // longer than a single source cycle; give it the same 30-min budget.
+  'autopilot-global-maintenance': THIRTY_MIN_MS,
 };
 
 /**

src/core/minions/handlers/subagent.ts

@@ -878,7 +878,7 @@ async function runSubagentViaGateway(args: GatewayRunArgs): Promise<SubagentResu
       const rows = await engine.executeRaw<{ gbrain_tool_use_id: string }>(
         `INSERT INTO subagent_tool_executions
            (job_id, message_idx, tool_use_id, tool_name, input, status, schema_version, ordinal, gbrain_tool_use_id, provider_id)
-         VALUES ($1, $2, $3, $4, $5::jsonb, 'pending', 2, $6, $7, $8)
+         VALUES ($1, $2, $3, $4, $5::text::jsonb, 'pending', 2, $6, $7, $8)
          ON CONFLICT (job_id, message_idx, ordinal) DO UPDATE
            SET status = subagent_tool_executions.status
          RETURNING gbrain_tool_use_id::text AS gbrain_tool_use_id`,
@@ -891,7 +891,7 @@ async function runSubagentViaGateway(args: GatewayRunArgs): Promise<SubagentResu
     onToolCallComplete: async (gbrainToolUseId, output) => {
       await engine.executeRaw(
         `UPDATE subagent_tool_executions
-           SET status = 'complete', output = $1::jsonb, ended_at = now()
+           SET status = 'complete', output = $1::text::jsonb, ended_at = now()
          WHERE gbrain_tool_use_id::text = $2`,
         [JSON.stringify(output ?? null), gbrainToolUseId],
       );
@@ -1107,7 +1107,7 @@ async function persistMessage(engine: BrainEngine, jobId: number, msg: Persisted
   await engine.executeRaw(
     `INSERT INTO subagent_messages (job_id, message_idx, role, content_blocks,
         tokens_in, tokens_out, tokens_cache_read, tokens_cache_create, model)
-     VALUES ($1, $2, $3, $4::jsonb, $5, $6, $7, $8, $9)
+     VALUES ($1, $2, $3, $4::text::jsonb, $5, $6, $7, $8, $9)
      ON CONFLICT (job_id, message_idx) DO NOTHING`,
     [
       jobId,
@@ -1131,13 +1131,15 @@ async function persistToolExecPending(
   toolName: string,
   input: unknown,
 ): Promise<void> {
-  // Serialize to JSON string for the ::jsonb cast. When `input` is already a
-  // string (e.g. pre-serialized), avoid double-encoding which produces a jsonb
-  // scalar string instead of a jsonb object — breaking `input->>'key'` lookups.
+  // Serialize to a JSON string, then bind through $5::text::jsonb. The value is
+  // ALWAYS a string here (pre-serialized input, or JSON.stringify) — binding a
+  // string to a bare $5::jsonb double-encodes it into a jsonb scalar string under
+  // postgres.js .unsafe() (#2339 class; PGLite hides it). The ::text cast makes
+  // the text→jsonb parse produce a real jsonb object.
   const jsonStr = typeof input === 'string' ? input : JSON.stringify(input);
   await engine.executeRaw(
     `INSERT INTO subagent_tool_executions (job_id, message_idx, tool_use_id, tool_name, input, status)
-     VALUES ($1, $2, $3, $4, $5::jsonb, 'pending')
+     VALUES ($1, $2, $3, $4, $5::text::jsonb, 'pending')
      ON CONFLICT (job_id, tool_use_id) DO NOTHING`,
     [jobId, messageIdx, toolUseId, toolName, jsonStr],
   );
@@ -1151,7 +1153,7 @@ async function persistToolExecComplete(
 ): Promise<void> {
   await engine.executeRaw(
     `UPDATE subagent_tool_executions
-        SET status = 'complete', output = $3::jsonb, ended_at = now()
+        SET status = 'complete', output = $3::text::jsonb, ended_at = now()
       WHERE job_id = $1 AND tool_use_id = $2`,
     [jobId, toolUseId, typeof output === 'string' ? output : JSON.stringify(output)],
   );
@@ -1170,7 +1172,7 @@ async function persistToolExecFailed(
   // rejected upfront) and "pending row exists" (tool threw mid-execute).
   await engine.executeRaw(
     `INSERT INTO subagent_tool_executions (job_id, message_idx, tool_use_id, tool_name, input, status, error, ended_at)
-     VALUES ($1, $2, $3, $4, $5::jsonb, 'failed', $6, now())
+     VALUES ($1, $2, $3, $4, $5::text::jsonb, 'failed', $6, now())
      ON CONFLICT (job_id, tool_use_id) DO UPDATE
        SET status = 'failed', error = EXCLUDED.error, ended_at = now()`,
     [jobId, messageIdx, toolUseId, toolName, typeof input === 'string' ? input : JSON.stringify(input), error],

src/core/minions/queue.ts

@@ -647,9 +647,18 @@ export class MinionQueue {
   async handleTimeouts(): Promise<MinionJob[]> {
     return this.engine.transaction(async (tx) => {
       const rows = await tx.executeRaw<Record<string, unknown>>(
+        // #1737: count the timed-out run as a spent attempt (terminal, no retry),
+        // mirroring handleWallClockTimeouts + handleStalled. handleTimeouts is the
+        // FIRST killer to fire for the long-lane handlers (timeout_ms stamped at
+        // submit), so without this the job reads `attempts: 0/N (started: N)`.
+        // Safe against double-count: the worker sweep runs handleStalled ->
+        // handleTimeouts -> handleWallClockTimeouts sequentially and awaited, and
+        // each guards on `status = 'active'`, so the first to set status='dead'
+        // excludes the row from the later sweeps.
         `UPDATE minion_jobs SET
           status = 'dead',
           error_text = 'timeout exceeded',
+          attempts_made = attempts_made + 1,
           lock_token = NULL,
           lock_until = NULL,
           finished_at = now(),

src/core/minions/supervisor.ts

@@ -30,6 +30,7 @@ import { detectTini } from './spawn-helpers.ts';
 import { resolveDefaultMaxRssMb } from './rss-default.ts';
 import {
   ChildWorkerSupervisor,
+  HARD_STOP_CRASH_MULTIPLIER,
   type ChildSupervisorEvent,
 } from './child-worker-supervisor.ts';
 import {
@@ -191,6 +192,21 @@ export function buildWorkerArgs(
  *  shutdown() drain window (issue #1801, D3). */
 const WEDGE_RESTART_GRACE_MS = 35_000;
 
+/**
+ * issue #1994: resolve the hard permanent-give-up ceiling. Default
+ * maxCrashes × HARD_STOP_CRASH_MULTIPLIER; operators override (or disable with
+ * 0 = never auto-stop) via GBRAIN_SUPERVISOR_HARD_STOP_CRASHES. A negative or
+ * non-integer override is ignored (falls back to the default).
+ */
+export function resolveHardStopMaxCrashes(maxCrashes: number): number {
+  const raw = process.env.GBRAIN_SUPERVISOR_HARD_STOP_CRASHES;
+  if (raw !== undefined && raw !== '') {
+    const n = Number(raw);
+    if (Number.isInteger(n) && n >= 0) return n;
+  }
+  return maxCrashes * HARD_STOP_CRASH_MULTIPLIER;
+}
+
 /** Calculate backoff: 1s, 2s, 4s, 8s, 16s, 32s, 60s cap. */
 export function calculateBackoffMs(crashCount: number): number {
   const base = Math.min(1000 * Math.pow(2, Math.max(crashCount, 0)), 60_000);
@@ -293,7 +309,11 @@ export const ExitCodes = {
  * pidfile path. TTL > refresh-interval × max-failures so we always exit
  * before our lock could lapse and let a second supervisor take over.
  */
-const SUPERVISOR_LOCK_TTL_MIN = 5;
+// Exported (issue #2227) so observability surfaces (`gbrain jobs supervisor
+// status`, `gbrain doctor`) compute the lock-freshness steal grace with the
+// SAME TTL the supervisor refreshes against, when detecting a live supervisor
+// via the DB lock instead of the (possibly split-$HOME) pidfile.
+export const SUPERVISOR_LOCK_TTL_MIN = 5;
 const SUPERVISOR_LOCK_REFRESH_MS = 60_000;
 const SUPERVISOR_LOCK_REFRESH_MAX_FAILURES = 3; // 3 × 60s = 180s < 5min TTL
 
@@ -825,6 +845,10 @@ export class MinionSupervisor {
       args: workerArgs,
       env,
       maxCrashes: this.opts.maxCrashes,
+      // issue #1994: hard permanent-give-up ceiling (the runaway backstop).
+      // Operators can raise/lower or disable (0 = never auto-stop) via
+      // GBRAIN_SUPERVISOR_HARD_STOP_CRASHES; default is maxCrashes × 10.
+      hardStopMaxCrashes: resolveHardStopMaxCrashes(this.opts.maxCrashes),
       _backoffFloorMs: this.opts._backoffFloorMs,
       isStopping: () => this.stopping,
       onMaxCrashesExceeded: (count, max) => {
@@ -907,6 +931,9 @@ export class MinionSupervisor {
           // ("raise --max-rss") is one glance away. Peak RSS stays in the
           // worker's own stderr line (the supervisor never sees it).
           ...(event.reason === 'rss_watchdog_loop' ? { max_rss_mb: this.opts.maxRssMb } : {}),
+          // issue #1994: degraded mode crossed the soft crash budget — surface
+          // it so doctor/status show "retrying with backoff" instead of silence.
+          ...(event.max !== undefined ? { max_crashes: event.max } : {}),
         });
         return;
     }

src/core/onboard/impact-capture.ts

@@ -120,7 +120,7 @@ export async function writeImpactLogRow(
          remediation_id, metric_name, metric_before, metric_after,
          job_id, source_id, brain_id, started_at, idempotency_key,
          applied_by, details
-       ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11::jsonb)`,
+       ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11::text::jsonb)`,
       [
         attribution.remediation_id,
         metricName,

src/core/op-checkpoint.ts

@@ -179,15 +179,22 @@ export async function recordCompleted(
   // REPLACE semantics (kept deliberately — #1794 V3). Callers like
   // extract-conversation-facts serialize a MUTABLE map through here and rely on
   // stale keys being REMOVED; an append would make them unremovable. The full
-  // set lands in the parent `completed_keys` JSONB column via a single UPSERT —
-  // exactly as before. JSON.stringify into `$3::jsonb` is correct (the text→jsonb
-  // cast yields a proper array; NOT the double-encode trap, which is the template
-  // form). Sync uses `appendCompleted` (below) instead, never this.
+  // set lands in the parent `completed_keys` JSONB column via a single UPSERT.
+  // #2339: bind through `$3::text::jsonb`, NOT `$3::jsonb`. Under postgres.js
+  // `.unsafe(sql, params)` (executeRawDirect's path) a JS string bound to a
+  // `$N::jsonb` param double-encodes — the text→jsonb cast wraps the already-JSON
+  // string into a jsonb *string scalar*, which fails the v119
+  // `op_checkpoints_completed_keys_array CHECK (jsonb_typeof = 'array')` and aborts
+  // every sync on real Postgres (PGLite parses it silently, which hid the bug).
+  // Casting through `text` first binds it as a plain text param so the text→jsonb
+  // cast parses it into a genuine jsonb array. This is the positional-param form of
+  // the CLAUDE.md double-encode trap (the grep guard only caught the template form).
+  // Sync uses `appendCompleted` (below, `unnest($3::text[])`) instead, never this.
   const sorted = [...keys].sort();
   return durableWrite(engine, key, 'write', () =>
     engine.executeRawDirect(
       `INSERT INTO op_checkpoints (op, fingerprint, completed_keys, updated_at)
-       VALUES ($1, $2, $3::jsonb, now())
+       VALUES ($1, $2, $3::text::jsonb, now())
        ON CONFLICT (op, fingerprint) DO UPDATE
          SET completed_keys = EXCLUDED.completed_keys,
              updated_at     = now()`,

src/core/operations.ts

@@ -2473,7 +2473,10 @@ const get_status_snapshot: Operation = {
     }
     const sync = await buildSyncStatusReport(ctx.engine, sources);
     const cycle = await buildCycleSnapshot(ctx.engine);
-    return { schema_version: 1 as const, sync, cycle };
+    // #1984: report the brain server's version so a thin-client `gbrain status`
+    // can surface remote_version alongside its own local CLI version.
+    const { VERSION } = await import('../version.ts');
+    return { schema_version: 1 as const, version: VERSION, sync, cycle };
   },
   scope: 'admin',
   localOnly: false,

src/core/search/hybrid.ts

@@ -647,6 +647,11 @@ export async function applyAliasHop(
     if (!page) continue;
     injectScore += 1e-6;
     out.push({
+      // #2339-sibling: include page_id. The `as SearchResult` cast hid its
+      // absence, so any consumer reading page_id off an alias-injected result got
+      // undefined — e.g. listActiveTakesForPages bound undefined/NaN into
+      // ANY($1::int[]) and crashed the contradiction probe on real Postgres.
+      page_id: page.id,
       slug: page.slug,
       title: page.title,
       type: page.type,

src/core/search/query-cache.ts

@@ -236,7 +236,7 @@ export class SemanticQueryCache {
       // the v0.40.3.0 IRON-RULE).
       await this.engine.executeRaw(
         `INSERT INTO query_cache (id, query_text, source_id, knobs_hash, embedding, results, meta, ttl_seconds, page_generations, max_generation_at_store, created_at)
-         VALUES ($1, $2, $3, $4, $5::vector, $6::jsonb, $7::jsonb, $8, $9::jsonb, $10, now())
+         VALUES ($1, $2, $3, $4, $5::vector, $6::text::jsonb, $7::text::jsonb, $8, $9::text::jsonb, $10, now())
          ON CONFLICT (id) DO UPDATE SET
            query_text = EXCLUDED.query_text,
            knobs_hash = EXCLUDED.knobs_hash,

src/core/sources-ops.ts

@@ -408,7 +408,7 @@ export async function addSource(
     try {
       await engine.executeRaw(
         `INSERT INTO sources (id, name, local_path, config)
-             VALUES ($1, $2, $3, $4::jsonb)`,
+             VALUES ($1, $2, $3, $4::text::jsonb)`,
         [opts.id, displayName, finalPath, JSON.stringify(config)],
       );
     } catch (e) {
@@ -454,7 +454,7 @@ export async function addSource(
     const displayName = opts.name ?? opts.id;
     await engine.executeRaw(
       `INSERT INTO sources (id, name, local_path, config)
-           VALUES ($1, $2, $3, $4::jsonb)`,
+           VALUES ($1, $2, $3, $4::text::jsonb)`,
       [opts.id, displayName, finalPath, JSON.stringify(config)],
     );
   }

src/core/sql-query.ts

@@ -79,15 +79,22 @@ function assertSqlValue(value: unknown): asserts value is SqlValue {
  * auth/admin surface that a focused helper preserves the contract without
  * forcing every call site to remember which positions hold JSONB.
  *
- * Why this is safe vs the v0.12.0 double-encode bug: the bug was specific
- * to postgres.js's template-tag auto-stringify path interacting with
- * sql.json() — not to positional binding through `unsafe()`. JS objects
- * passed as positional params reach the wire protocol with the correct
- * type oid (jsonb when cast in the SQL string), so there is no double-
- * encode. The CI guard (scripts/check-jsonb-pattern.sh) doesn't fire
- * because the source pattern is a method call (`executeRawJsonb(...)`),
- * not the banned literal-template-tag interpolation pattern with
- * JSON.stringify cast to jsonb.
+ * Why this is safe vs the double-encode bug: this helper binds a JS **object**
+ * (not a pre-stringified string) to each `$N::jsonb` position. postgres.js
+ * `unsafe()` and PGLite both serialize a JS object to the jsonb wire type
+ * correctly, so there is no double-encode.
+ *
+ * IMPORTANT (the #2339 distinction): positional binding is NOT universally safe.
+ * Binding `JSON.stringify(x)` (a **string**) to a `$N::jsonb` position via
+ * `unsafe()`/`executeRawDirect` DOES double-encode — the text→jsonb cast wraps
+ * the already-JSON string into a jsonb *string scalar* (PGLite hides it; real
+ * Postgres exposes it, and it broke every sync in #2339). The fixes are: pass a
+ * raw object (this helper), or cast through `$N::text::jsonb` so the string is
+ * parsed, never `$N::jsonb` + JSON.stringify. The legacy grep guard
+ * (scripts/check-jsonb-pattern.sh) only caught the template-tag form; the
+ * positional `$N::jsonb` + JSON.stringify form is caught by the AST guard
+ * scripts/check-jsonb-params.mjs. This helper's `executeRawJsonb(...)` method-call
+ * shape trips neither guard because it passes objects, which is correct.
  *
  * Usage:
  *   await executeRawJsonb(

src/core/takes-quality-eval/receipt-write.ts

@@ -32,8 +32,8 @@ export async function writeReceiptToDb(engine: BrainEngine, receipt: TakesQualit
        receipt_json, receipt_disk_path, created_at
      ) VALUES (
        $1, $2, $3, $4,
-       $5, $6, $7, $8::jsonb, $9,
-       $10::jsonb, $11, $12::timestamptz
+       $5, $6, $7, $8::text::jsonb, $9,
+       $10::text::jsonb, $11, $12::timestamptz
      )
      ON CONFLICT (receipt_sha8_corpus, receipt_sha8_prompt, receipt_sha8_models, receipt_sha8_rubric)
      DO NOTHING`,