fix(tests): close three integration-test races that flaked CI (#9964)

* fix(tests): wait for the job lane's first leg text to persist before killing its process

The takeover test aborted the turn as soon as the fake worker had written the
first leg, before the dying process had recorded that text in its stream
buffer. The re-attach then carried receivedTextChars 0 and the occurrence's
summary lost the first leg. Wait for the text event in the buffer first, as
the unattended-turn suite already does.

* fix(tests): handle the cancelled token-sealing run's rejection before polling for its backend

The operator-cancel test started migrate() and only attached a handler after
its pg_cancel_backend polling loop. When the cancellation reached the run's
connection before the loop broke, Node saw an unhandled rejection and Vitest
failed the whole db shard even though every test passed. Attach the rejection
assertion when the run starts and await it after the loop.

* fix(tests): handle the refused provider expansion before awaiting the concurrent grant

The concurrent-grant test started ensureWorkspaceAccountsGroup and awaited
the grant transaction before attaching a handler to the refusal. When the
refusal resolved first, Node reported an unhandled rejection and Vitest failed
the shard although the test passed. Attach the rejection assertion when the
call starts and await it after the grant commits.

* chore(ci): log lock waits and slow statements in the integration Postgres

Author: waleedlatif1

.github/workflows/checks.yml

@@ -149,6 +149,10 @@ jobs:
         working-directory: packages/db
         run: |
           bun -e 'import postgres from "postgres"; const sql = postgres(process.env.DATABASE_URL); for (const extension of ["vector", "btree_gin", "pg_trgm"]) await sql`CREATE EXTENSION IF NOT EXISTS ${sql(extension)}`; await sql.end()'
+          # A test that hangs on a lock leaves only its timeout behind; these name the waiter, the
+          # lock, and the blocking statement in the Postgres log. Service containers take no -c flags,
+          # and ALTER SYSTEM runs outside a transaction, so each setting is its own statement.
+          bun -e 'import postgres from "postgres"; const sql = postgres(process.env.DATABASE_URL); for (const setting of ["log_lock_waits = on", "deadlock_timeout = 1000", "log_min_duration_statement = 20000"]) await sql.unsafe(`ALTER SYSTEM SET ${setting}`); await sql`SELECT pg_reload_conf()`; await sql.end()'
           bun run db:${{ matrix.provision }}
 
       - name: Verify migration replay is a no-op

apps/sim/lib/knowledge/__integration__/search-source-setup.integration.ts

@@ -492,14 +492,17 @@ describe('Search source identity and concurrent creation', () => {
         .where(eq(resourcePolicy.resourceId, searchGroupId))
     })
     await locked
-    const append = ensureWorkspaceAccountsGroup(ids.workspaceId, ids.aliceId, {
-      provider: 'jira',
-      label: 'Jira',
-      required: false,
-    })
+    /** Handled from the start: the refusal can land before the grant's commit resolves here. */
+    const refused = expect(
+      ensureWorkspaceAccountsGroup(ids.workspaceId, ids.aliceId, {
+        provider: 'jira',
+        label: 'Jira',
+        required: false,
+      })
+    ).rejects.toThrow('has workflow access')
     releaseGrant()
     await grant
-    await expect(append).rejects.toThrow('has workflow access')
+    await refused
     const existing = await ensureWorkspaceAccountsGroup(ids.workspaceId, ids.aliceId, {
       provider: 'google-drive',
       label: 'Google Drive',

apps/sim/lib/mothership/jobs/lane.integration.ts

@@ -165,6 +165,7 @@ import { jobOccurrenceId, MIN_JOB_CADENCE_MS } from '@/lib/mothership/jobs/occur
 import { JOB_DISPATCH_GRACE_MS, reconcileJobRuns } from '@/lib/mothership/jobs/reconcile'
 import { takeOverAbandonedRuns } from '@/lib/mothership/request/application/abandoned-runs'
 import { startUnattendedTurn } from '@/lib/mothership/request/lifecycle/unattended'
+import { readEvents } from '@/lib/mothership/request/session/buffer'
 import {
   chatStreamLockKey,
   StreamControllerSupersededError,
@@ -506,6 +507,14 @@ describe.runIf(Boolean(redisUrl))('the scheduled job lane', () => {
     ])
     await runJobLane(new Date())
     await waitFor(async () => worker.open.has(occurrenceId), 'the first leg')
+    /** The worker has written the text, but the process must persist it before it dies. */
+    await waitFor(
+      async () =>
+        (await readEvents(occurrenceId, '0')).some(
+          (event) => event.type === 'text' && event.payload.text === 'Started triage. '
+        ),
+      'the first leg text to persist before the process dies'
+    )
 
     controllers.get(occurrenceId)?.abort(new StreamControllerSupersededError())
     worker.open.get(occurrenceId)?.response.socket?.destroy()

packages/db/script-migrations/0037_seal_account_tokens.integration.ts

@@ -358,7 +358,8 @@ describe('sealing stored account tokens in PostgreSQL', () => {
       await insertAccount('cancelled-access', 'cancelled-refresh')
       await onAccountUpdate('PERFORM pg_sleep(20)')
 
-      const run = migrate()
+      /** Handled from the start: the cancel can fail the run before the polling loop sees it. */
+      const failed = expect(migrate()).rejects.toThrow('canceling statement due to user request')
       for (let attempt = 0; attempt < 100; attempt++) {
         const [cancelled] = await admin<{ cancelled: boolean }[]>`
           SELECT pg_cancel_backend(pid) AS cancelled FROM pg_stat_activity
@@ -368,7 +369,7 @@ describe('sealing stored account tokens in PostgreSQL', () => {
         await sleep(50)
       }
 
-      await expect(run).rejects.toThrow('canceling statement due to user request')
+      await failed
       expect(await isRecorded()).toBe(false)
       expect((await storedRows())[0].access_token).toBe('cancelled-access')
     }, 30_000)