test(backfill): the two cutoff guards pin reachability, not text order (#1149)

backend/__tests__/service/threading.backfillTransaction.test.js

@@ -0,0 +1,198 @@
+/**
+ * The backfill's UPDATE and its ledger INSERT are ONE transaction — executed.
+ *
+ * Tier 1 on purpose. The guard for this property in
+ * `__tests__/unit/models/threadingCutoffRecord.test.js` is a token scan over a
+ * slice of the script source, and @sprint-review measured what that misses:
+ * gate the ledger INSERT on a never-true condition and every anchor stays put
+ * (BEGIN, the UPDATE, the INSERT, COMMIT, ROLLBACK all still present, still in
+ * order), both unit suites report 38/38 green, and the run leaves every chain
+ * rooted with no cutoff recorded. That is verbatim the harm the transaction
+ * guard's own comment names — so the guard is blind to the one failure it was
+ * written for.
+ *
+ * It cannot be fixed at Tier 0: pg-mem rejects `WITH RECURSIVE` outright, which
+ * `threading.derivation.test.js` already records. The backfill's CTE needs a
+ * real server, so the atomicity property does too.
+ *
+ * Two directions, because "one transaction" is two claims:
+ *   forward  — a successful run leaves BOTH the rooted rows and the ledger row
+ *   backward — a run whose INSERT throws leaves NEITHER
+ * A test of only the forward half passes against a script with no transaction
+ * at all.
+ */
+const { Pool } = require('pg');
+
+const RUN = process.env.INTEGRATION_TEST === 'true';
+const d = RUN ? describe : describe.skip;
+
+// `--apply` is read at module load (`const APPLY = process.argv.includes(...)`),
+// so it has to be in place before the require below, not before the call.
+if (!process.argv.includes('--apply')) process.argv.push('--apply');
+// eslint-disable-next-line global-require
+const { main, MIGRATION_NAME } = require('../../scripts/backfill-thread-root-id');
+
+const POD = 'aaaaaaaaaaaaaaaaaaaaaa02';
+const USER = 'bbbbbbbbbbbbbbbbbbbbbb02';
+
+let pool;
+
+const connect = () => new Pool({
+  host: process.env.PG_HOST,
+  port: Number(process.env.PG_PORT || 5432),
+  database: process.env.PG_DATABASE,
+  user: process.env.PG_USER,
+  password: process.env.PG_PASSWORD,
+  ssl: false,
+});
+
+const reset = async () => {
+  await pool.query('DELETE FROM migration_records WHERE name = $1', [MIGRATION_NAME]);
+  await pool.query('DELETE FROM messages WHERE pod_id = $1', [POD]);
+  await pool.query(
+    `INSERT INTO users (_id, username) VALUES ($1,'tester') ON CONFLICT (_id) DO NOTHING`, [USER],
+  );
+  await pool.query(
+    `INSERT INTO pods (id, name, type, created_by) VALUES ($1,'P','chat',$2)
+     ON CONFLICT (id) DO NOTHING`, [POD, USER],
+  );
+  const ins = async (replyTo, rootId, minutesAgo) => {
+    const { rows } = await pool.query(
+      `INSERT INTO messages (pod_id, user_id, content, reply_to_message_id, thread_root_id, created_at)
+       VALUES ($1,$2,'m',$3,$4, NOW() - ($5 || ' minutes')::interval) RETURNING id`,
+      [POD, USER, replyTo, rootId, String(minutesAgo)],
+    );
+    return rows[0].id;
+  };
+  // One already-rooted reply so CUTOFF_SQL takes the from_rooted branch —
+  // without it the script needs --derivation-live and refuses to record.
+  const liveRoot = await ins(null, null, 50);
+  await ins(liveRoot, liveRoot, 40);
+  // And an un-rooted chain for the backfill to actually repair.
+  const oldRoot = await ins(null, null, 30);
+  const a = await ins(oldRoot, null, 20);
+  await ins(a, null, 10);
+  return { oldRoot };
+};
+
+const unrootedCount = async () => {
+  const { rows } = await pool.query(
+    `SELECT count(*)::int AS n FROM messages
+      WHERE pod_id = $1 AND reply_to_message_id IS NOT NULL AND thread_root_id IS NULL`, [POD],
+  );
+  return rows[0].n;
+};
+
+const ledgerRow = async () => {
+  const { rows } = await pool.query(
+    'SELECT details FROM migration_records WHERE name = $1', [MIGRATION_NAME],
+  );
+  return rows[0] || null;
+};
+
+d('the UPDATE and the ledger INSERT commit together or not at all', () => {
+  beforeAll(() => { pool = connect(); });
+  afterAll(async () => {
+    if (!pool) return;
+    await pool.query('DELETE FROM messages WHERE pod_id = $1', [POD]);
+    await pool.query('DELETE FROM migration_records WHERE name = $1', [MIGRATION_NAME]);
+    await pool.end();
+  });
+
+  beforeEach(async () => {
+    await reset();
+    jest.spyOn(console, 'log').mockImplementation(() => {});
+    jest.spyOn(console, 'error').mockImplementation(() => {});
+  });
+  afterEach(() => jest.restoreAllMocks());
+
+  it('FORWARD: a successful run leaves both the rooted rows and the ledger row', async () => {
+    expect(await unrootedCount()).toBeGreaterThan(0);   // CONTROL: there was work
+    await main(pool);
+    expect(await unrootedCount()).toBe(0);
+    const row = await ledgerRow();
+    expect(row).not.toBeNull();
+    // The mutant @sprint-review measured gates exactly this INSERT off. Every
+    // text anchor survives it; this assertion does not.
+    expect(row.details.rowsUpdated).toBeGreaterThan(0);
+    expect(row.details).toHaveProperty('threadingCutoff');
+  });
+
+  it('BACKWARD: an INSERT that throws rolls the UPDATE back', async () => {
+    // Why this case is not redundant with FORWARD, measured rather than argued.
+    // @sprint-review reproduced the gated-INSERT mutant and found all three arms
+    // redden, reading that as BACKWARD earning its place. It doesn't: three arms
+    // failing on ONE mutant is co-sensitivity, not discrimination — FORWARD alone
+    // would have caught it. The question is whether any mutant reddens BACKWARD
+    // and leaves the rest green. Two do, on real pg16:
+    //
+    //   remove `BEGIN`            → FORWARD ✓, BACKWARD ✕ | Tier 0 ✕ (textually)
+    //   UPDATE via `pool.query`   → FORWARD ✓, BACKWARD ✕ | Tier 0 38/38 GREEN
+    //
+    // The second is the one that matters, and it is the likelier refactor of the
+    // two: the transaction still opens, the ledger INSERT still sits inside it,
+    // BEGIN still precedes UPDATE precedes INSERT precedes COMMIT in the source —
+    // so every ordering assertion in threadingCutoffRecord.test.js holds while the
+    // UPDATE autocommits on a second connection and survives the ROLLBACK. A
+    // source scan can see that a transaction is written; only this case can see
+    // that the write is inside it.
+    const before = await unrootedCount();
+    expect(before).toBeGreaterThan(0);
+
+    // Wrap the pool so the ledger INSERT — and only it — fails. Everything
+    // else, including BEGIN/ROLLBACK, goes to the real server, so this
+    // exercises the script's own transaction rather than a simulated one.
+    //
+    // The patch is UNDONE on release. A pg client is pooled: assigning to
+    // `client.query` mutates the object the pool hands out again, so without
+    // this the poison outlives the test and every later checkout rejects its
+    // ledger INSERT. That cost the next two cases in this file before it was
+    // found, and it failed as a plausible product bug (no ledger row) rather
+    // than as a broken fixture.
+    const failingPool = {
+      query: (...args) => pool.query(...args),
+      connect: async () => {
+        const client = await pool.connect();
+        const realQuery = client.query.bind(client);
+        const realRelease = client.release.bind(client);
+        client.query = (...args) => {
+          const sql = String(args[0]?.text || args[0]);
+          if (/INSERT INTO migration_records/i.test(sql)) {
+            return Promise.reject(new Error('simulated ledger write failure'));
+          }
+          return realQuery(...args);
+        };
+        client.release = (...args) => {
+          delete client.query;
+          delete client.release;
+          return realRelease(...args);
+        };
+        return client;
+      },
+    };
+
+    // NOT a rejection: main()'s outer catch logs and sets `process.exitCode`,
+    // so a failed backfill surfaces as a non-zero exit, never as a thrown
+    // promise. Asserting on the exit code is asserting the contract an
+    // operator (and any CI wrapper) actually observes.
+    process.exitCode = 0;
+    await main(failingPool);
+    expect(process.exitCode).toBe(1);
+    process.exitCode = 0;
+
+    // Neither half landed: the rows are still un-rooted and there is no record.
+    expect(await unrootedCount()).toBe(before);
+    expect(await ledgerRow()).toBeNull();
+  });
+
+  it('and a second run reports the recorded cutoff without re-measuring', async () => {
+    await main(pool);
+    const first = await ledgerRow();
+    console.log.mockClear();
+    await main(pool);
+    const second = await ledgerRow();
+    expect(second.details).toEqual(first.details);
+    expect(console.log.mock.calls.map((c) => String(c[0])).join('\n'))
+      .toContain('the boundary is not re-measured');
+  });
+});

backend/__tests__/unit/models/threadingCutoffLedgerFirst.test.js

@@ -0,0 +1,99 @@
+/**
+ * A second run reports the recorded cutoff and STOPS — executed, not read.
+ *
+ * @sprint-review 57397 established the ordering; #1149 pinned it; and
+ * @sprint-review then re-checked the guards themselves and found both were
+ * pure `indexOf`/`lastIndexOf` over the script SOURCE, so the mutants they
+ * were written against ship green. A position comparison cannot see a
+ * `return`: delete the early return and `ledgerRead < populationRead` still
+ * holds, while the script re-measures a boundary that is no longer
+ * measurable — the exact #1148 failure.
+ *
+ * Text guards remain in threadingCutoffRecord.test.js and are useful for the
+ * structural half (rule 18). This file covers the behavioural half the only
+ * way it can be covered: by running the thing.
+ *
+ * The instrument is a counting pool. "Stopped" is not observable from the
+ * database — ON CONFLICT DO NOTHING means a re-measuring run leaves the ledger
+ * row byte-identical, so asserting on stored state cannot distinguish the two.
+ * What distinguishes them is WHICH QUERIES ISSUE, so that is what we record.
+ */
+const { newDb } = require('pg-mem');
+const { applyTable } = require('../../utils/schemaTable');
+
+const mockDb = newDb();
+const basePool = new (mockDb.adapters.createPg().Pool)();
+
+const issued = [];
+const countingPool = {
+  query: (...args) => {
+    issued.push(String(args[0]?.text || args[0]));
+    return basePool.query(...args);
+  },
+};
+
+const { main, MIGRATION_NAME } = require('../../../scripts/backfill-thread-root-id');
+
+const CUTOFF = '2026-08-01T00:00:00.000Z';
+
+beforeAll(async () => {
+  await applyTable(basePool, 'migration_records');
+});
+
+beforeEach(() => {
+  issued.length = 0;
+  jest.spyOn(console, 'log').mockImplementation(() => {});
+  jest.spyOn(console, 'error').mockImplementation(() => {});
+});
+
+afterEach(() => jest.restoreAllMocks());
+
+describe('the ledger row short-circuits the run', () => {
+  beforeAll(async () => {
+    await basePool.query(
+      `INSERT INTO migration_records (name, details)
+       VALUES ($1, $2::jsonb) ON CONFLICT (name) DO NOTHING`,
+      [MIGRATION_NAME, JSON.stringify({ threadingCutoff: CUTOFF, rowsUpdated: 7 })],
+    );
+  });
+
+  it('issues the ledger read and NOTHING else', async () => {
+    await main(countingPool);
+    expect(issued).toHaveLength(1);
+    expect(issued[0]).toMatch(/FROM migration_records WHERE name = \$1/);
+  });
+
+  it('never counts the population, which is the query that would lie', async () => {
+    // After a backfill every pre-threading reply carries a root, so the
+    // population read and the MIN below it return a plausible wrong answer
+    // rather than an obviously broken one. Not reaching them is the property.
+    await main(countingPool);
+    expect(issued.join('\n')).not.toMatch(/needs_root/);
+    expect(issued.join('\n')).not.toMatch(/UPDATE messages/);
+  });
+
+  it('reports the recorded value rather than a recomputed one', async () => {
+    await main(countingPool);
+    const said = console.log.mock.calls.map((c) => String(c[0])).join('\n');
+    expect(said).toContain(MIGRATION_NAME);
+    expect(said).toContain(CUTOFF);
+    expect(said).toContain('the boundary is not re-measured');
+  });
+
+  it('CONTROL: with no ledger row the run does NOT stop there', async () => {
+    // Without this the assertions above are equally consistent with a main()
+    // that issues one query and returns under every condition — including a
+    // stubbed-out one. Proves the short-circuit is the ledger row's doing.
+    await basePool.query('DELETE FROM migration_records WHERE name = $1', [MIGRATION_NAME]);
+    issued.length = 0;
+    await main(countingPool).catch(() => {});
+    expect(issued.length).toBeGreaterThan(1);
+    expect(issued.join('\n')).toMatch(/needs_root/);
+    // Restore for any later describe.
+    await basePool.query(
+      `INSERT INTO migration_records (name, details)
+       VALUES ($1, $2::jsonb) ON CONFLICT (name) DO NOTHING`,
+      [MIGRATION_NAME, JSON.stringify({ threadingCutoff: CUTOFF, rowsUpdated: 7 })],
+    );
+  });
+});

backend/__tests__/unit/models/threadingCutoffRecord.test.js

@@ -90,6 +90,22 @@ describe('the cutoff is measured before it becomes unmeasurable', () => {
     const populationRead = SCRIPT.indexOf('count(*)::int AS needs_root');
     expect(ledgerRead).toBeGreaterThan(-1);
     expect(ledgerRead).toBeLessThan(populationRead);
+    // #1149: reading the ledger first is not the property — STOPPING is, and a
+    // position comparison cannot see a `return`. Deleting the early return in
+    // the script's `if (ledger)` branch leaves both indices unchanged and the
+    // script re-measures anyway, which is the exact failure #1148 removed. Pin
+    // the thing that stops it.
+    //
+    // Cited by symbol, not by line. This comment said `:209` and the export of
+    // `main` in this same PR moved it to `:221` — a pointer that rotted inside
+    // the branch that wrote it, caught by @sprint-review reproducing the two
+    // mutation rows. A line number in a comment is a claim about a file's
+    // length, which is the one property every commit is entitled to change.
+    const reportedAndStopped = SCRIPT.slice(
+      SCRIPT.indexOf('already recorded at'),
+      populationRead,
+    );
+    expect(reportedAndStopped).toMatch(/\breturn\b/);
     expect(SCRIPT).toMatch(/already recorded at \$\{ledger\.applied_at\}/);
     expect(SCRIPT).toMatch(/the boundary is not re-measured/);
   });
@@ -106,6 +122,38 @@ describe('the cutoff is measured before it becomes unmeasurable', () => {
     expect(begin).toBeLessThan(update);
     expect(update).toBeLessThan(insert);
     expect(insert).toBeLessThan(commit);
+    // #1149: order is not reachability. A bare `return` between the UPDATE and
+    // the INSERT keeps all four anchors in place and makes the ledger write
+    // unreachable — the mutant that stayed green here. Assert no control-flow
+    // break separates them.
+    expect(SCRIPT.slice(update, insert)).not.toMatch(/\breturn\b|\bthrow\b|process\.exit/);
+    // MEASURED, not predicted (@sprint-review, 2026-08-25): this test is blind
+    // to THE defect it was written for. Gate the ledger INSERT on a never-true
+    // condition and every anchor above stays in place, this file reports 38/38
+    // green, and the run leaves every chain rooted with no cutoff recorded —
+    // which is word for word the harm named four lines up. Not "blind to a
+    // class of mutants": blind to the one.
+    //
+    // The class is the generalisation, and it is the weaker sentence, so it
+    // goes second. A token scan catches mutants that INSERT a control-flow
+    // keyword and misses every mutant that removes reachability without one.
+    // It is false-red-prone in the other direction too: extract a helper
+    // between these two anchors and its `return` fails this.
+    //
+    // The anchors themselves are sound and were checked — `UPDATE messages m`
+    // is unique, and `lastIndexOf` is the correct pick for the INSERT because
+    // the zero-eligible-edges branch above contains a second
+    // `INSERT INTO migration_records` that `indexOf` would grab instead. So
+    // this guard is as good as a source scan can be here; the limit is the
+    // instrument, not the anchors, which is exactly why the property moved to
+    // a tier that runs the code.
+    //
+    // Both halves are now executed elsewhere, and this stays as the cheap
+    // structural check only:
+    //   ledger-first  -> threadingCutoffLedgerFirst.test.js (Tier 0, pg-mem)
+    //   atomicity     -> service/threading.backfillTransaction.test.js (Tier 1)
+    // Atomicity had to go to Tier 1 because pg-mem cannot parse the backfill's
+    // `WITH RECURSIVE`, which threading.derivation.test.js already records.
     expect(SCRIPT).toMatch(/await client\.query\('ROLLBACK'\)/);
     expect(SCRIPT).toMatch(/client\.release\(\)/);
   });

backend/__tests__/unit/services/agentEventService.lifecycle.test.js

@@ -286,7 +286,8 @@ describe('garbageCollect: the requeue predicate and its cap', () => {
     expect(expire.filter.attempts).toEqual({ $gte: 3 });
   });
 
-  // #993: the requeue at :650 and the pending delete at :698 ran in the same
+  // #993: the requeue (`requeueResult = await AgentEvent.updateMany`) and the
+  // pending delete (`deleteMany({ status: 'pending' ... })`) ran in the same
   // Promise.all against different fields — the requeue sets `status` but not
   // `createdAt`, so an event it had just rescued walked into a `createdAt`-keyed
   // delete carrying its original age. 38 events were destroyed in one measured