fix(autopilot): drop maxWaiting from per-source submit (E2E found silent coalesce)

The Postgres E2E for fan-out surfaced two real bugs that the unit-stub
tests + PGLite tests couldn't catch:

1) **CRITICAL** — dispatchPerSource passed `maxWaiting: 1` to every
   per-source queue.add. maxWaiting is per-(name, queue) — since all
   per-source jobs share `name='autopilot-cycle'`, the second + third +
   Nth source's submit silently coalesced into the FIRST source's
   waiting job. Net result on a real worker: 1 job processed per tick,
   not N. The entire fan-out feature was a silent no-op past the first
   source.

   Per-source idempotency_key (`autopilot-cycle:<source_id>:<slot>`)
   already handles "two ticks for same source within slot" dedup, which
   is the only thing maxWaiting was buying us. Dropping it fixes fan-out
   without losing dedup.

   New unit-stub regression test asserts maxWaiting is NOT in the per-
   source submit opts so a future refactor that re-adds it gets caught
   in 100x faster CI (test/autopilot-fanout.test.ts).

2) **postgres-engine.ts:updateSourceConfig** — initial impl used
   sql.json() correctly but my mid-debug rewrite to executeRaw +
   positional `$1::jsonb` produced JSONB STRING shape (not OBJECT)
   because postgres-js double-encodes JS string params in unsafe mode.
   `||` between JSONB object + JSONB string yields a JSONB ARRAY,
   wiping every existing config key on update.

   Same latent bug class exists at src/commands/sources.ts:482 (gbrain
   sources federate/unfederate path); flagged for follow-up but
   out-of-scope here.

   Reverted to sql.json() inside the template tag (verified via direct
   psql round-trip: jsonb_typeof = 'object'). Updated the e2e seed
   helper to use sql.json() too — the executeRaw + JSON.stringify
   pattern was producing string-shape JSONB at SEED time which made
   the failure cascade harder to debug.

Coverage adds:
- test/e2e/list-all-sources-postgres.test.ts: 11 cases pin Postgres
  parity for listAllSources + updateSourceConfig including jsonb_typeof
  round-trip
- test/e2e/autopilot-fanout-postgres.test.ts: 6 cases end-to-end
  including 3-source fan-out producing 3 distinct rows, idempotency
  coalesce within slot, cap honored, legacy fallback path
- test/autopilot-fanout.test.ts: +1 regression guard on maxWaiting

This is the kind of bug that justifies the user's "fill test gaps then
run E2E" mandate. The unit tests + PGLite parity tests + typecheck all
passed cleanly; only the real-Postgres E2E found the coalesce.

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

src/commands/autopilot-fanout.ts

@@ -216,7 +216,12 @@ export async function dispatchPerSource(
           idempotency_key: `autopilot-cycle:${src.id}:${opts.slot}`,
           max_attempts: 2,
           timeout_ms: opts.timeoutMs,
-          maxWaiting: 1,
+          // DELIBERATELY no maxWaiting: 1 here. maxWaiting is per
+          // (name, queue), so it would coalesce all N per-source jobs
+          // sharing name='autopilot-cycle' down to ONE waiting job —
+          // killing the fan-out. The per-source idempotency_key
+          // already provides the right dedup granularity (one job per
+          // source per slot, regardless of how many ticks try).
         },
       );
       dispatched.push(src.id);

src/core/postgres-engine.ts

@@ -967,16 +967,21 @@ export class PostgresEngine implements BrainEngine {
     // shape this autopilot wave uses is flat (`last_full_cycle_at`,
     // `archive_*`, etc.) so concat is sufficient. Idempotent on re-run.
     //
-    // sql.json(patch) is the canonical safe path per feedback_postgres_jsonb_double_encode
-    // — postgres-js handles JSONB serialization (no double-encode). Matches
-    // the pattern at putPage and submitJob elsewhere in this file.
+    // MUST use sql.json(patch) inside the template tag — postgres-js's
+    // positional executeRaw + `$1::jsonb` cast DOUBLE-ENCODES the
+    // JSON.stringify'd string, producing a JSONB STRING shape instead
+    // of OBJECT. `||` between JSONB object + JSONB string yields a
+    // JSONB ARRAY (concat semantics for non-matching types), which
+    // wipes every existing config key. sql.json(...) inside the
+    // template tag is the canonical safe path — same pattern as
+    // putPage + submitJob elsewhere in this file. Empirically verified
+    // produces jsonb_typeof = 'object'.
     const sql = this.sql;
     const result = await sql`
       UPDATE sources
          SET config = COALESCE(config, '{}'::jsonb) || ${sql.json(patch as Parameters<typeof sql.json>[0])}
        WHERE id = ${sourceId}
     `;
-    // postgres-js returns count as result.count; matched-rows shape
     return (result.count ?? 0) > 0;
   }
 

test/autopilot-fanout.test.ts

@@ -281,6 +281,22 @@ describe('dispatchPerSource — integration with stubbed engine + queue', () =>
     expect(parsed.pending.length).toBe(2);
   });
 
+  test('per-source submit MUST NOT pass maxWaiting (regression — coalesces all sources to one job)', async () => {
+    // Direct unit-stub queues can't enforce maxWaiting semantics (the
+    // production MinionQueue implementation does), so this catches the
+    // regression by inspecting the submit opts at the dispatch boundary.
+    // If a future refactor re-adds maxWaiting:1 to the per-source path,
+    // the production fan-out would silently coalesce N sources to ONE
+    // waiting job per tick — killing the entire feature. The e2e test
+    // also catches this against a real queue, but this guard fires in
+    // unit tests too so the bug surfaces 100x faster.
+    const { engine, queue, added, fanoutOpts } = makeStubs([src('a'), src('b'), src('c')]);
+    await dispatchPerSource(engine, queue, fanoutOpts);
+    for (const job of added) {
+      expect(job.opts.maxWaiting).toBeUndefined();
+    }
+  });
+
   test('all-fresh tick dispatches nothing (no jobs added)', async () => {
     const NOW = Date.now();
     const recent = (id: string) =>

test/e2e/autopilot-fanout-postgres.test.ts

@@ -50,6 +50,8 @@ beforeEach(async () => {
 
 async function seedSource(id: string, opts: { local_path?: string } = {}): Promise<void> {
   const localPath = opts.local_path ?? mkdtempSync(join(tmpdir(), `gbrain-fanout-${id}-`));
+  // Direct literal `'{}'::jsonb` is fine (no parameter binding). Test
+  // explicitly resets config to {} so each test starts clean.
   await engine.executeRaw(
     `INSERT INTO sources (id, name, local_path, config, archived, created_at)
      VALUES ($1, $2, $3, '{}'::jsonb, false, NOW())

test/e2e/list-all-sources-postgres.test.ts

@@ -42,16 +42,21 @@ async function seedSource(
 ): Promise<void> {
   const localPath = opts.local_path === undefined ? `/tmp/${id}` : opts.local_path;
   const archived = opts.archived === true;
-  const config = JSON.stringify(opts.config ?? {});
-  await engine.executeRaw(
-    `INSERT INTO sources (id, name, local_path, config, archived, created_at)
-     VALUES ($1, $2, $3, $4::jsonb, $5, NOW())
-     ON CONFLICT (id) DO UPDATE
-       SET local_path = EXCLUDED.local_path,
-           config = EXCLUDED.config,
-           archived = EXCLUDED.archived`,
-    [id, id, localPath, config, archived],
-  );
+  // NOTE: do NOT use executeRaw + `JSON.stringify(config) + $N::jsonb` —
+  // postgres-js double-encodes the JS string parameter, producing a JSONB
+  // STRING shape instead of OBJECT. Use sql.json() inside the template tag.
+  // Same pattern as putPage. The pre-existing sources.ts:482 has the
+  // same latent bug; the call site there is rare (gbrain sources
+  // federate/unfederate) and out of scope for this PR.
+  const eng = engine as unknown as { sql: (...args: unknown[]) => Promise<{ count?: number }> };
+  await (eng.sql as any)`
+    INSERT INTO sources (id, name, local_path, config, archived, created_at)
+    VALUES (${id}, ${id}, ${localPath}, ${(eng.sql as any).json(opts.config ?? {})}, ${archived}, NOW())
+    ON CONFLICT (id) DO UPDATE
+      SET local_path = EXCLUDED.local_path,
+          config = EXCLUDED.config,
+          archived = EXCLUDED.archived
+  `;
 }
 
 describeIfDB('Postgres parity — listAllSources', () => {