feat(sdk): expose active sessions (#33991)

packages/core/src/session.ts

@@ -155,6 +155,7 @@ export interface Interface {
   }) => Effect.Effect<void, OperationUnavailableError>
   readonly compact: (input: CompactInput) => Effect.Effect<void, NotFoundError | OperationUnavailableError>
   readonly wait: (id: SessionSchema.ID) => Effect.Effect<void, NotFoundError | OperationUnavailableError>
+  readonly active: Effect.Effect<ReadonlySet<SessionSchema.ID>>
   readonly resume: (sessionID: SessionSchema.ID) => Effect.Effect<void, NotFoundError | SessionRunner.RunError>
   readonly interrupt: (sessionID: SessionSchema.ID) => Effect.Effect<void>
   readonly revert: {
@@ -402,6 +403,7 @@ export const layer = Layer.unwrap(
               yield* result.get(sessionID)
               return yield* new OperationUnavailableError({ operation: "wait" })
             }),
+            active: execution.active,
             resume: Effect.fn("V2Session.resume")(function* (sessionID) {
               yield* result.get(sessionID)
               yield* execution.resume(sessionID)

packages/core/src/session/execution.ts

@@ -5,6 +5,8 @@ import { SessionRunner } from "./runner/index"
 import { SessionSchema } from "./schema"
 
 export interface Interface {
+  /** Snapshots active execution owned by this process. */
+  readonly active: Effect.Effect<ReadonlySet<SessionSchema.ID>>
   /** Starts execution while idle or joins the active execution. */
   readonly resume: (sessionID: SessionSchema.ID) => Effect.Effect<void, SessionRunner.RunError>
   /** Registers newly recorded work. Repeated wakeups may coalesce. */
@@ -19,5 +21,10 @@ export class Service extends Context.Service<Service, Interface>()("@opencode/v2
 /** Low-level compatibility layer for callers that only need durable Session recording. */
 export const noopLayer = Layer.succeed(
   Service,
-  Service.of({ resume: () => Effect.void, wake: () => Effect.void, interrupt: () => Effect.void }),
+  Service.of({
+    active: Effect.succeed(new Set()),
+    resume: () => Effect.void,
+    wake: () => Effect.void,
+    interrupt: () => Effect.void,
+  }),
 )

packages/core/src/session/execution/local.ts

@@ -28,6 +28,7 @@ export const layer = Layer.effect(
     })
 
     return SessionExecution.Service.of({
+      active: coordinator.active,
       interrupt: coordinator.interrupt,
       resume: coordinator.run,
       wake: coordinator.wake,

packages/core/src/session/run-coordinator.ts

@@ -4,6 +4,8 @@ import { Deferred, Effect, Exit, Fiber, FiberSet, Scope } from "effect"
 
 /** Serializes execution for each key while allowing different keys to run concurrently. */
 export interface Coordinator<Key, E> {
+  /** Snapshots keys with an execution owned by this coordinator. */
+  readonly active: Effect.Effect<ReadonlySet<Key>>
   /** Starts execution while idle or joins the active execution. */
   readonly run: (key: Key) => Effect.Effect<void, E>
   /** Registers one coalesced follow-up after newly recorded work. */
@@ -98,5 +100,5 @@ export const make = <Key, E>(options: {
         return Fiber.interrupt(entry.owner)
       })
 
-    return { run, wake, interrupt }
+    return { active: Effect.sync(() => new Set(active.keys())), run, wake, interrupt }
   })

packages/core/test/session-prompt.test.ts

@@ -22,9 +22,11 @@ import { testEffect } from "./lib/effect"
 const executionCalls: SessionV2.ID[] = []
 const interruptCalls: SessionV2.ID[] = []
 const wakeCalls: SessionV2.ID[] = []
+const activeSessions = new Set<SessionV2.ID>()
 const execution = Layer.succeed(
   SessionExecution.Service,
   SessionExecution.Service.of({
+    active: Effect.sync(() => new Set(activeSessions)),
     resume: (sessionID) =>
       Effect.sync(() => {
         executionCalls.push(sessionID)
@@ -108,6 +110,13 @@ const eventCount = (type: string) =>
   )
 
 describe("SessionV2.prompt", () => {
+  it.effect("exposes the execution registry", () =>
+    Effect.gen(function* () {
+      activeSessions.add(sessionID)
+      expect(Array.from(yield* (yield* SessionV2.Service).active)).toEqual([sessionID])
+    }).pipe(Effect.ensuring(Effect.sync(() => activeSessions.clear()))),
+  )
+
   it.effect("delegates execution continuation through SessionExecution", () =>
     Effect.gen(function* () {
       yield* setup

packages/core/test/session-run-coordinator.test.ts

@@ -66,6 +66,78 @@ describe("SessionRunCoordinator", () => {
     ),
   )
 
+  it.effect("snapshots only active executions", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const firstStarted = yield* Deferred.make<void>()
+        const secondStarted = yield* Deferred.make<void>()
+        const firstGate = yield* Deferred.make<void>()
+        const secondGate = yield* Deferred.make<void>()
+        const coordinator = yield* SessionRunCoordinator.make({
+          drain: (key: string) =>
+            Deferred.succeed(key === "first" ? firstStarted : secondStarted, undefined).pipe(
+              Effect.andThen(Deferred.await(key === "first" ? firstGate : secondGate)),
+            ),
+        })
+
+        expect(Array.from(yield* coordinator.active)).toEqual([])
+        const first = yield* coordinator.run("first").pipe(Effect.forkChild)
+        yield* Deferred.await(firstStarted)
+        expect(Array.from(yield* coordinator.active)).toEqual(["first"])
+
+        const second = yield* coordinator.run("second").pipe(Effect.forkChild)
+        yield* Deferred.await(secondStarted)
+        expect(Array.from(yield* coordinator.active)).toEqual(["first", "second"])
+
+        yield* Deferred.succeed(firstGate, undefined)
+        yield* Fiber.join(first)
+        expect(Array.from(yield* coordinator.active)).toEqual(["second"])
+        yield* Deferred.succeed(secondGate, undefined)
+        yield* Fiber.join(second)
+        expect(Array.from(yield* coordinator.active)).toEqual([])
+      }),
+    ),
+  )
+
+  it.effect("cleans active executions after failure and defect", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const failure = new Error("failed")
+        const defect = new Error("defect")
+        const coordinator = yield* SessionRunCoordinator.make({
+          drain: (key: string) => (key === "failure" ? Effect.fail(failure) : Effect.die(defect)),
+        })
+
+        const failed = yield* coordinator.run("failure").pipe(Effect.exit)
+        expect(Exit.isFailure(failed) && Cause.hasFails(failed.cause)).toBeTrue()
+        expect(Array.from(yield* coordinator.active)).toEqual([])
+
+        const died = yield* coordinator.run("defect").pipe(Effect.exit)
+        expect(Exit.isFailure(died) && Cause.hasDies(died.cause)).toBeTrue()
+        expect(Array.from(yield* coordinator.active)).toEqual([])
+      }),
+    ),
+  )
+
+  it.effect("cleans active executions when its scope closes", () =>
+    Effect.gen(function* () {
+      const started = yield* Deferred.make<void>()
+      const coordinator = yield* Effect.scoped(
+        Effect.gen(function* () {
+          const coordinator = yield* SessionRunCoordinator.make({
+            drain: () => Deferred.succeed(started, undefined).pipe(Effect.andThen(Effect.never)),
+          })
+          yield* coordinator.wake("session")
+          yield* Deferred.await(started)
+          expect(Array.from(yield* coordinator.active)).toEqual(["session"])
+          return coordinator
+        }),
+      )
+
+      expect(Array.from(yield* coordinator.active)).toEqual([])
+    }),
+  )
+
   it.effect("coalesces wakes received during active execution", () =>
     Effect.scoped(
       Effect.gen(function* () {
@@ -166,6 +238,7 @@ describe("SessionRunCoordinator", () => {
 
         const exit = yield* Fiber.await(resumed)
         expect(Exit.isFailure(exit) && Cause.hasInterruptsOnly(exit.cause)).toBeTrue()
+        expect(Array.from(yield* coordinator.active)).toEqual([])
         expect(runs).toBe(1)
       }),
     ),

packages/core/test/session-runner-recorded.test.ts

@@ -95,6 +95,7 @@ const execution = Layer.effect(
       drain: (sessionID, force) => sessionRunner.run({ sessionID, force }),
     })
     return SessionExecution.Service.of({
+      active: coordinator.active,
       resume: coordinator.run,
       wake: coordinator.wake,
       interrupt: coordinator.interrupt,

packages/core/test/session-runner.test.ts

@@ -254,6 +254,7 @@ const execution = Layer.effect(
       drain: (sessionID, force) => sessionRunner.run({ sessionID, force }),
     })
     return SessionExecution.Service.of({
+      active: coordinator.active,
       resume: coordinator.run,
       wake: coordinator.wake,
       interrupt: coordinator.interrupt,