merge: #2074 fix(postgres-engine): atomic JSONB merge in updateSourceConfig

Reviewed: single-statement config || patch eliminates the read-then-write
lost-update race. Touches updateSourceConfig (disjoint from the deleteFactsForPage
region the authored #1928 fix changed).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

src/core/postgres-engine.ts

@@ -1276,11 +1276,28 @@ export class PostgresEngine implements BrainEngine {
   }
 
   async updateSourceConfig(sourceId: string, patch: Record<string, unknown>): Promise<boolean> {
-    // v0.38: atomic JSONB merge. `||` is the Postgres concat operator —
-    // for jsonb, right-side keys overwrite left-side; nested object keys
-    // are NOT deep-merged (use jsonb_set for nested paths). The patch
-    // shape this autopilot wave uses is flat (`last_full_cycle_at`,
-    // `archive_*`, etc.) so concat is sufficient. Idempotent on re-run.
+    // Atomic single-statement merge. The previous read-then-write form dropped
+    // concurrent updates: two callers patching different keys could both read
+    // the same old config and the later `SET config = ...` clobbered the
+    // earlier patch. These keys are written by background cycle/autopilot
+    // paths, so the merge must happen inside the UPDATE (parity with
+    // pglite-engine.updateSourceConfig, which already uses JSONB `||`).
+    //
+    // The CASE normalizes historical bad shapes inline (so `config` is re-read
+    // against the row-locked latest version — a CTE/subquery snapshot would
+    // reintroduce the lost-update race under READ COMMITTED): older code paths
+    // could store config as a JSONB string (double-encoded) or as a JSONB array
+    // of patch objects. We coerce those to a flat object before the `||` merge
+    // so doctor and source routing keep getting flat keys.
+    //
+    // String branch guard: a JSONB string whose inner text is NOT itself valid
+    // JSON (one of the historical bad shapes this path repairs) would make the
+    // bare `::jsonb` cast raise `invalid input syntax for type json`, failing
+    // the whole UPDATE. Postgres has no `try_cast`, so we gate the cast with
+    // the SQL `IS JSON` predicate (Postgres 16+): parseable inner text is
+    // double-encoded config and gets parsed; unparseable text falls back to `{}`.
+    // The guard keeps the merge a single atomic statement (no extra round-trip,
+    // no lost-update race).
     //
     // MUST use sql.json(patch) inside the template tag — postgres-js's
     // positional executeRaw + `$1::jsonb` cast DOUBLE-ENCODES the
@@ -1294,7 +1311,25 @@ export class PostgresEngine implements BrainEngine {
     const sql = this.sql;
     const result = await sql`
       UPDATE sources
-         SET config = COALESCE(config, '{}'::jsonb) || ${sql.json(patch as Parameters<typeof sql.json>[0])}
+         SET config =
+           CASE
+             WHEN jsonb_typeof(config) = 'object' THEN config
+             WHEN jsonb_typeof(config) = 'string'
+               THEN CASE
+                 WHEN (config #>> '{}') IS JSON
+                   THEN COALESCE(NULLIF((config #>> '{}'), '')::jsonb, '{}'::jsonb)
+                 ELSE '{}'::jsonb
+               END
+             WHEN jsonb_typeof(config) = 'array'
+               THEN COALESCE(
+                 (SELECT jsonb_object_agg(kv.key, kv.value)
+                    FROM jsonb_array_elements(config) elem,
+                         jsonb_each(elem) kv),
+                 '{}'::jsonb
+               )
+             ELSE '{}'::jsonb
+           END
+           || ${sql.json(patch as Parameters<typeof sql.json>[0])}
        WHERE id = ${sourceId}
     `;
     return (result.count ?? 0) > 0;

test/list-all-sources.test.ts

@@ -135,4 +135,51 @@ describe('engine.updateSourceConfig', () => {
     const all = await engine.listAllSources();
     expect(all.find(s => s.id === 'delta')!.config.last_full_cycle_at).toBe('2026-05-22T11:00:00.000Z');
   });
+
+  // IS JSON guard: the postgres-engine atomic merge gates its `::jsonb` cast
+  // behind the SQL `IS JSON` predicate so a historical bad row whose config is
+  // a JSONB string of NON-JSON text normalizes to `{}` instead of raising
+  // `invalid input syntax for type json` and aborting the UPDATE.
+  //
+  // We exercise the SQL CASE expression directly via executeRaw against PGLite
+  // (which ships Postgres 17 + IS JSON parity) rather than calling
+  // PostgresEngine.updateSourceConfig (which requires a live Postgres pool).
+  test('IS-JSON guard: non-JSON string config normalizes to {} on merge', async () => {
+    const patch = { merged_key: 'v' };
+
+    const guarded = (configExpr: string) => `
+      SELECT (
+        CASE
+          WHEN jsonb_typeof(${configExpr}) = 'object' THEN ${configExpr}
+          WHEN jsonb_typeof(${configExpr}) = 'string'
+            THEN CASE
+              WHEN (${configExpr} #>> '{}') IS JSON
+                THEN COALESCE(NULLIF((${configExpr} #>> '{}'), '')::jsonb, '{}'::jsonb)
+              ELSE '{}'::jsonb
+            END
+          WHEN jsonb_typeof(${configExpr}) = 'array'
+            THEN COALESCE(
+              (SELECT jsonb_object_agg(kv.key, kv.value)
+                 FROM jsonb_array_elements(${configExpr}) elem,
+                      jsonb_each(elem) kv),
+              '{}'::jsonb
+            )
+          ELSE '{}'::jsonb
+        END || $1::jsonb
+      ) AS result`;
+
+    // 1. JSONB string holding NON-JSON text → normalizes to {} then merges patch.
+    const bad = await engine.executeRaw<{ result: Record<string, unknown> }>(
+      guarded(`to_jsonb('garbage text'::text)`),
+      [JSON.stringify(patch)],
+    );
+    expect(bad[0].result).toEqual({ merged_key: 'v' });
+
+    // 2. JSONB string holding double-encoded valid JSON object → parsed + merged.
+    const good = await engine.executeRaw<{ result: Record<string, unknown> }>(
+      guarded(`to_jsonb('{"x":1}'::text)`),
+      [JSON.stringify(patch)],
+    );
+    expect(good[0].result).toEqual({ x: 1, merged_key: 'v' });
+  });
 });