From 68ef893818d43f248fd72028f7d51d6770cd47a8 Mon Sep 17 00:00:00 2001
From: Kit Langton <kit.langton@gmail.com>
Date: Fri, 24 Jul 2026 12:58:50 -0400
Subject: [PATCH] fix(core): start all suspended sessions promptly (#38720)

---
 .../core/src/session/execution/restart.ts     |  3 +--
 packages/core/test/session-execution.test.ts  | 27 +++++++++++++++++++
 2 files changed, 28 insertions(+), 2 deletions(-)

diff --git a/packages/core/src/session/execution/restart.ts b/packages/core/src/session/execution/restart.ts
index 985c46a556e..d7133bb7d25 100644
--- a/packages/core/src/session/execution/restart.ts
+++ b/packages/core/src/session/execution/restart.ts
@@ -40,8 +40,7 @@ export const layer = Layer.effect(
               // Drain failures are already logged and durably recorded by the execution layer.
               yield* Effect.ignore(execution.resume(sessionID))
             }),
-          // Each suspension is consumed atomically right before its drain; at most four drains run at once.
-          { concurrency: 4, discard: true },
+          { concurrency: "unbounded", discard: true },
         )
       }),
     })
diff --git a/packages/core/test/session-execution.test.ts b/packages/core/test/session-execution.test.ts
index 192b88eac87..9ff3cc0c54d 100644
--- a/packages/core/test/session-execution.test.ts
+++ b/packages/core/test/session-execution.test.ts
@@ -105,6 +105,31 @@ describe("SessionExecution lifecycle", () => {
     }),
   )
 
+  it.effect("starts every suspended execution without waiting for earlier drains to finish", () =>
+    Effect.gen(function* () {
+      const database = yield* Database.Service
+      const sessionIDs = Array.from({ length: 5 }, (_, index) => SessionV2.ID.make(`ses_resume_concurrent_${index}`))
+      yield* seedSessions(database, sessionIDs, { time_suspended: Date.now() })
+
+      const fourStarted = yield* Deferred.make<void>()
+      const started: SessionV2.ID[] = []
+      const scope = yield* Scope.make()
+      yield* Effect.addFinalizer(() => Scope.close(scope, Exit.void))
+      const context = yield* buildExecution(scope, ({ sessionID }) =>
+        Effect.sync(() => {
+          started.push(sessionID)
+          if (started.length === 4) Deferred.doneUnsafe(fourStarted, Effect.void)
+        }).pipe(Effect.andThen(Effect.never)),
+      )
+      const execution = Context.get(context, SessionExecution.Service)
+      const restart = Context.get(context, SessionRestart.Service)
+      yield* restart.resumeSuspendedSessions.pipe(Effect.forkIn(scope))
+      yield* Deferred.await(fourStarted)
+
+      expect([...(yield* execution.active)].toSorted()).toEqual(sessionIDs.toSorted())
+    }),
+  )
+
   it.effect("resumes each suspended Session at most once", () =>
     Effect.gen(function* () {
       const database = yield* Database.Service
@@ -115,9 +140,11 @@ describe("SessionExecution lifecycle", () => {
       const drained: string[] = []
       const scope = yield* Scope.make()
       const context = yield* buildExecution(scope, ({ sessionID }) => Effect.sync(() => void drained.push(sessionID)))
+      const execution = Context.get(context, SessionExecution.Service)
       const restart = Context.get(context, SessionRestart.Service)
 
       yield* restart.resumeSuspendedSessions
+      yield* Effect.forEach([first, second], execution.awaitIdle, { discard: true })
       expect(drained.toSorted()).toEqual([first, second])
       expect(yield* suspensions(database)).toEqual({ [first]: false, [second]: false })