From 7c829c13056eb708d4946523116b1a0cf4bee6b3 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 12 Nov 2026 16:03:18 -0800 Subject: [PATCH 1/4] 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. --- apps/sim/lib/mothership/jobs/lane.integration.ts | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/apps/sim/lib/mothership/jobs/lane.integration.ts b/apps/sim/lib/mothership/jobs/lane.integration.ts index 90e3691fcd6..3f48510b313 100644 --- a/apps/sim/lib/mothership/jobs/lane.integration.ts +++ b/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() From 3810acc81e69aef2f029a3c8eda8334e8dbe558d Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 12 Nov 2026 16:20:06 -0800 Subject: [PATCH 2/4] 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. --- .../0037_seal_account_tokens.integration.ts | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/packages/db/script-migrations/0037_seal_account_tokens.integration.ts b/packages/db/script-migrations/0037_seal_account_tokens.integration.ts index 1a20acdff36..e0b112d1522 100644 --- a/packages/db/script-migrations/0037_seal_account_tokens.integration.ts +++ b/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) From 2be3fb36bb86d77a58f42b75e8def6b7213be1f4 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 12 Nov 2026 16:21:19 -0800 Subject: [PATCH 3/4] 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. --- .../search-source-setup.integration.ts | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/apps/sim/lib/knowledge/__integration__/search-source-setup.integration.ts b/apps/sim/lib/knowledge/__integration__/search-source-setup.integration.ts index 85f98cabd07..9ad0ac3616d 100644 --- a/apps/sim/lib/knowledge/__integration__/search-source-setup.integration.ts +++ b/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', From f6442fb48b3a911e5de7c9cfd1e20bf9743d4e11 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 12 Nov 2026 16:28:45 -0800 Subject: [PATCH 4/4] chore(ci): log lock waits and slow statements in the integration Postgres --- .github/workflows/checks.yml | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/.github/workflows/checks.yml b/.github/workflows/checks.yml index e0d1d77f3b8..c938b1fd884 100644 --- a/.github/workflows/checks.yml +++ b/.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