Merge remote-tracking branch 'origin/dev' into session-event-stream

# ------------------------ >8 ------------------------
# Do not modify or remove the line above.
# Everything below it will be ignored.
#
# Conflicts:
#	packages/core/src/session/execution.ts
#	packages/core/src/session/execution/local.ts
#	packages/core/src/session/run-coordinator.ts
#	packages/core/test/session-prompt.test.ts
#	packages/core/test/session-run-coordinator.test.ts
#	packages/core/test/session-runner-recorded.test.ts
#	packages/core/test/session-runner.test.ts
#	packages/protocol/src/groups/session.ts

packages/core/src/session.ts

@@ -106,7 +106,6 @@ export type MessageNotFoundError = SessionRevert.MessageNotFoundError
 export type Error = NotFoundError | MessageDecodeError | OperationUnavailableError | PromptConflictError
 
 export interface Interface {
-  readonly active: Effect.Effect<ReadonlySet<SessionSchema.ID>>
   readonly list: (input?: ListInput) => Effect.Effect<SessionSchema.Info[]>
   readonly create: (input: CreateInput) => Effect.Effect<SessionSchema.Info>
   readonly get: (sessionID: SessionSchema.ID) => Effect.Effect<SessionSchema.Info, NotFoundError>
@@ -156,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: {
@@ -202,7 +202,6 @@ export const layer = Layer.unwrap(
             )
 
           const result = Service.of({
-            active: execution.active,
             create: Effect.fn("V2Session.create")(function* (input) {
               const sessionID = input.id ?? SessionSchema.ID.create()
               const recorded = yield* store.get(sessionID)
@@ -432,6 +431,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/test/session-prompt.test.ts

@@ -22,6 +22,7 @@ import { testEffect } from "./lib/effect"
 const executionCalls: SessionV2.ID[] = []
 const interruptCalls: SessionV2.ID[] = []
 const wakeCalls: SessionV2.ID[] = []
+const activeSessions = new Set<SessionV2.ID>()
 let activityObserver: ((active: boolean) => void) | undefined
 let activityState = false
 const setActivity = (active: boolean) =>
@@ -32,7 +33,7 @@ const setActivity = (active: boolean) =>
 const execution = Layer.succeed(
   SessionExecution.Service,
   SessionExecution.Service.of({
-    active: Effect.succeed(new Set()),
+    active: Effect.sync(() => new Set(activeSessions)),
     activity: () =>
       Effect.sync(() => {
         activityState = false
@@ -128,6 +129,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

@@ -120,6 +120,44 @@ describe("SessionRunCoordinator", () => {
     ),
   )
 
+  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* () {
@@ -220,6 +258,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)
       }),
     ),