Run async execs in background

Author: richiemcilroy

apps/api/src/routes/execs.test.ts

@@ -332,7 +332,7 @@ describe.skipIf(!RUN_API_INTEGRATION_TESTS)('GET /v1/sandboxes/:id/exec/:execId
     expect(body.ended_at).toBeDefined()
   })
 
-  test('returns queued exec from async execution', async () => {
+  test('returns completed exec from async execution (background fiber finishes)', async () => {
     const env = createTestEnv()
     const sandboxId = await createRunningSandbox(env)
 
@@ -347,6 +347,9 @@ describe.skipIf(!RUN_API_INTEGRATION_TESTS)('GET /v1/sandboxes/:id/exec/:execId
         )
         const execBody = (yield* execRes.json) as { exec_id: string }
 
+        // Allow the daemon fiber to complete
+        yield* Effect.yieldNow()
+
         const getRes = yield* client.execute(
           HttpClientRequest.get(
             `/v1/sandboxes/${sandboxId}/exec/${execBody.exec_id}`,
@@ -359,8 +362,9 @@ describe.skipIf(!RUN_API_INTEGRATION_TESTS)('GET /v1/sandboxes/:id/exec/:execId
 
     expect(result.status).toBe(200)
     const body = result.body as Record<string, unknown>
-    expect(body.status).toBe('queued')
-    expect(body.exit_code).toBeNull()
+    // Background fiber runs the exec and transitions to done
+    expect(body.status).toBe('done')
+    expect(body.exit_code).toBe(0)
   })
 
   test('returns 404 for unknown exec', async () => {
@@ -474,13 +478,16 @@ describe.skipIf(!RUN_API_INTEGRATION_TESTS)('GET /v1/sandboxes/:id/execs — lis
             HttpClientRequest.bodyUnsafeJson({ cmd: ['echo', 'sync'] }),
           ),
         )
-        // Async exec → queued
+        // Async exec → also completes to done (background fiber runs immediately in tests)
         yield* client.execute(
           HttpClientRequest.post(`/v1/sandboxes/${sandboxId}/exec`).pipe(
             HttpClientRequest.bodyUnsafeJson({ cmd: ['sleep', '10'], wait: false }),
           ),
         )
 
+        // Allow the daemon fiber to complete
+        yield* Effect.yieldNow()
+
         const response = yield* client.execute(
           HttpClientRequest.get(`/v1/sandboxes/${sandboxId}/execs?status=done`),
         )
@@ -491,8 +498,10 @@ describe.skipIf(!RUN_API_INTEGRATION_TESTS)('GET /v1/sandboxes/:id/execs — lis
 
     expect(result.status).toBe(200)
     const body = result.body as { execs: Array<{ status: string }> }
-    expect(body.execs.length).toBe(1)
+    // Both execs complete to done (background fiber finishes immediately with in-memory services)
+    expect(body.execs.length).toBe(2)
     expect(body.execs[0].status).toBe('done')
+    expect(body.execs[1].status).toBe('done')
   })
 
   test('returns 404 for unknown sandbox', async () => {
@@ -710,7 +719,7 @@ describe.skipIf(!RUN_API_INTEGRATION_TESTS)('Exec status transitions', () => {
     expect(result.ended_at).toBeDefined()
   })
 
-  test('async exec stays queued until processed', async () => {
+  test('async exec transitions to done after background fiber completes', async () => {
     const env = createTestEnv()
     const sandboxId = await createRunningSandbox(env)
 
@@ -725,6 +734,9 @@ describe.skipIf(!RUN_API_INTEGRATION_TESTS)('Exec status transitions', () => {
         )
         const execBody = (yield* execRes.json) as { exec_id: string }
 
+        // Allow the daemon fiber to complete
+        yield* Effect.yieldNow()
+
         const getRes = yield* client.execute(
           HttpClientRequest.get(
             `/v1/sandboxes/${sandboxId}/exec/${execBody.exec_id}`,
@@ -735,10 +747,10 @@ describe.skipIf(!RUN_API_INTEGRATION_TESTS)('Exec status transitions', () => {
       }),
     )
 
-    expect(result.status).toBe('queued')
-    expect(result.exit_code).toBeNull()
-    expect(result.started_at).toBeNull()
-    expect(result.ended_at).toBeNull()
+    expect(result.status).toBe('done')
+    expect(result.exit_code).toBe(0)
+    expect(result.started_at).toBeDefined()
+    expect(result.ended_at).toBeDefined()
   })
 })
 

apps/api/src/routes/execs.ts

@@ -1,5 +1,5 @@
 import { HttpRouter, HttpServerRequest, HttpServerResponse } from '@effect/platform'
-import { Effect } from 'effect'
+import { Cause, Effect } from 'effect'
 import {
   generateUUIDv7,
   idToBytes,
@@ -68,6 +68,92 @@ function rowToExec(row: ExecRow): Exec {
   }
 }
 
+// -- Background async exec ----------------------------------------------------
+
+const runAsyncExec = (
+  execId: Uint8Array,
+  execIdStr: string,
+  sandboxIdBytes: Uint8Array,
+  cmdArray: string[],
+  cwd: string,
+  env: Record<string, string>,
+  timeoutSeconds: number,
+) =>
+  Effect.gen(function* () {
+    const execRepo = yield* ExecRepo
+    const nodeClient = yield* NodeClient
+    const redis = yield* RedisService
+
+    // Transition to running
+    yield* execRepo.updateStatus(execId, 'running', { startedAt: new Date() })
+
+    const result = yield* nodeClient.exec({
+      sandboxId: sandboxIdBytes,
+      execId: execIdStr,
+      cmd: cmdArray,
+      cwd,
+      env,
+      timeoutSeconds,
+    })
+
+    // Push events to Redis for SSE consumption
+    let eventSeq = 0
+    const now = new Date().toISOString()
+
+    if (result.stdout) {
+      eventSeq++
+      yield* redis.pushExecEvent(
+        execIdStr,
+        { seq: eventSeq, ts: now, data: { seq: eventSeq, t: 'stdout', data: result.stdout } },
+        EVENT_TTL_SECONDS,
+      )
+    }
+    if (result.stderr) {
+      eventSeq++
+      yield* redis.pushExecEvent(
+        execIdStr,
+        { seq: eventSeq, ts: now, data: { seq: eventSeq, t: 'stderr', data: result.stderr } },
+        EVENT_TTL_SECONDS,
+      )
+    }
+    eventSeq++
+    yield* redis.pushExecEvent(
+      execIdStr,
+      {
+        seq: eventSeq,
+        ts: now,
+        data: {
+          seq: eventSeq,
+          t: 'exit',
+          code: result.exitCode,
+          duration_ms: result.durationMs,
+          resource_usage: {
+            cpu_ms: result.cpuMs,
+            peak_memory_bytes: result.peakMemoryBytes,
+          },
+        },
+      },
+      EVENT_TTL_SECONDS,
+    )
+
+    // Update exec row to done
+    yield* execRepo.updateStatus(execId, 'done', {
+      exitCode: result.exitCode,
+      cpuMs: result.cpuMs,
+      peakMemoryBytes: result.peakMemoryBytes,
+      durationMs: result.durationMs,
+      endedAt: new Date(),
+    })
+  }).pipe(
+    Effect.catchAllCause((cause) =>
+      Effect.gen(function* () {
+        yield* Effect.logError(`Async exec failed for ${execIdStr}: ${Cause.pretty(cause)}`)
+        const execRepo = yield* ExecRepo
+        yield* execRepo.updateStatus(execId, 'failed', { endedAt: new Date() })
+      }).pipe(Effect.catchAll(() => Effect.void)),
+    ),
+  )
+
 // -- Execute command ----------------------------------------------------------
 
 const execCommand = Effect.gen(function* () {
@@ -185,8 +271,20 @@ const execCommand = Effect.gen(function* () {
 
   const execIdStr = bytesToId(EXEC_PREFIX, execId)
 
-  // Async mode: return immediately
+  // Async mode: fire background execution and return immediately
   if (!wait) {
+    yield* Effect.forkDaemon(
+      runAsyncExec(
+        execId,
+        execIdStr,
+        sandboxIdBytes,
+        cmdArray,
+        body.cwd ?? DEFAULT_CWD,
+        body.env ?? {},
+        timeoutSeconds,
+      ),
+    )
+
     const response: ExecAsyncResponse = {
       exec_id: execIdStr,
       status: 'queued',