test(desktop): synchronize the presence and Stop tests on real signals

The overlapping-stream test waits for both presence entries, then for the
leaving one's removal, instead of sleeping; the Stop race releases its held
transaction in a finally, so a failed poll cannot leave the run row locked.

Author: waleedlatif1

apps/sim/lib/desktop/application/executor.integration.ts

@@ -30,10 +30,9 @@ import {
 } from '@sim/db/schema'
 import { createDeferred } from '@sim/testing/helpers/deferred'
 import { featureFlagsMock, featureFlagsMockFns } from '@sim/testing/mocks/feature-flags.mock'
-import { sleep } from '@sim/utils/helpers'
 import { generateId } from '@sim/utils/id'
 import { eq, inArray, sql } from 'drizzle-orm'
-import { closeRedisConnection } from '@/lib/core/config/redis'
+import { closeRedisConnection, getRedisClient } from '@/lib/core/config/redis'
 import {
   claimDesktopTool,
   completeDesktopTool,
@@ -569,17 +568,20 @@ describe.runIf(Boolean(redisUrl))('desktop background executor protocol', () =>
         principal: desktop.principal,
         input: { deviceId: desktop.deviceId, toolCallId, executionToken, status: 'success' },
       })
-      /** Release Stop only once the result is queued behind its lock on the run row. */
-      await expect
-        .poll(async () => {
-          const [waiter] = await db.execute<{ pid: number }>(sql`
-            SELECT pid FROM pg_stat_activity WHERE wait_event_type = 'Lock'
-              AND ${stopPid}::int = ANY(pg_blocking_pids(pid)) LIMIT 1`)
-          return Boolean(waiter)
-        })
-        .toBe(true)
-      releaseStop.resolve()
-      await stopping
+      try {
+        /** Release Stop only once the result is queued behind its lock on the run row. */
+        await expect
+          .poll(async () => {
+            const [waiter] = await db.execute<{ pid: number }>(sql`
+              SELECT pid FROM pg_stat_activity WHERE wait_event_type = 'Lock'
+                AND ${stopPid}::int = ANY(pg_blocking_pids(pid)) LIMIT 1`)
+            return Boolean(waiter)
+          })
+          .toBe(true)
+      } finally {
+        releaseStop.resolve()
+        await stopping
+      }
 
       await expect(completing).resolves.toEqual({ outcome: 'superseded', status: 'cancelled' })
       expect((await row(toolCallId)).status).toBe('cancelled')
@@ -737,10 +739,11 @@ describe.runIf(Boolean(redisUrl))('desktop background executor protocol', () =>
       const closeStaying = (await open()).subscribe(() => {})
       await expect.poll(() => isDesktopPresent(desktop.deviceId)).toBe(true)
       const closeLeaving = (await open()).subscribe(() => {})
+      const streams = () => getRedisClient()?.zcard(`desktop:presence:${desktop.deviceId}`)
       try {
-        await sleep(200)
+        await expect.poll(streams).toBe(2)
         closeLeaving()
-        await sleep(200)
+        await expect.poll(streams).toBe(1)
         expect(await isDesktopPresent(desktop.deviceId)).toBe(true)
       } finally {
         closeStaying()