diff --git a/packages/core/src/session.ts b/packages/core/src/session.ts index c7436b7cc7..90b50869e8 100644 --- a/packages/core/src/session.ts +++ b/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> readonly list: (input?: ListInput) => Effect.Effect readonly create: (input: CreateInput) => Effect.Effect readonly get: (sessionID: SessionSchema.ID) => Effect.Effect @@ -156,6 +155,7 @@ export interface Interface { }) => Effect.Effect readonly compact: (input: CompactInput) => Effect.Effect readonly wait: (id: SessionSchema.ID) => Effect.Effect + readonly active: Effect.Effect> readonly resume: (sessionID: SessionSchema.ID) => Effect.Effect readonly interrupt: (sessionID: SessionSchema.ID) => Effect.Effect 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) diff --git a/packages/core/test/session-prompt.test.ts b/packages/core/test/session-prompt.test.ts index 7f54a43a65..3e921b8a15 100644 --- a/packages/core/test/session-prompt.test.ts +++ b/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() 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 diff --git a/packages/core/test/session-run-coordinator.test.ts b/packages/core/test/session-run-coordinator.test.ts index a44dd0731d..67b0bba2bf 100644 --- a/packages/core/test/session-run-coordinator.test.ts +++ b/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() + 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) }), ), diff --git a/packages/protocol/src/groups/session.ts b/packages/protocol/src/groups/session.ts index 649a2d780c..f623d1bdb8 100644 --- a/packages/protocol/src/groups/session.ts +++ b/packages/protocol/src/groups/session.ts @@ -73,10 +73,9 @@ export const SessionsCursor = Schema.String.pipe( ) export type SessionsCursor = typeof SessionsCursor.Type -export const SessionActive = Schema.Struct({ +const SessionActive = Schema.Struct({ type: Schema.Literal("running"), }).annotate({ identifier: "SessionActive" }) -export interface SessionActive extends Schema.Schema.Type {} const SessionsQueryCursor = SessionsCursor.annotate({ description: "Opaque pagination cursor returned as cursor.previous or cursor.next in the previous response.", diff --git a/packages/sdk/openapi.json b/packages/sdk/openapi.json index 7377e63f9a..4e206d9114 100644 --- a/packages/sdk/openapi.json +++ b/packages/sdk/openapi.json @@ -10187,6 +10187,66 @@ ] } }, + "/api/session/active": { + "get": { + "tags": ["sessions"], + "operationId": "v2.session.active", + "parameters": [], + "security": [], + "responses": { + "200": { + "description": "Success", + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "data": { + "type": "object", + "patternProperties": { + "^ses": { + "$ref": "#/components/schemas/SessionActive" + } + } + } + }, + "required": ["data"], + "additionalProperties": false + } + } + } + }, + "400": { + "description": "InvalidRequestError", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/InvalidRequestError" + } + } + } + }, + "401": { + "description": "UnauthorizedError", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/UnauthorizedError" + } + } + } + } + }, + "description": "Retrieve foreground Session drains currently owned by this OpenCode process. Sessions absent from the result are inactive.", + "summary": "List active sessions", + "x-codeSamples": [ + { + "lang": "js", + "source": "import { createOpencodeClient } from \"@opencode-ai/sdk\n\nconst client = createOpencodeClient()\nawait client.v2.session.active({\n ...\n})" + } + ] + } + }, "/api/session/{sessionID}": { "get": { "tags": ["sessions"], @@ -23445,6 +23505,17 @@ "required": ["_tag", "message"], "additionalProperties": false }, + "SessionActive": { + "type": "object", + "properties": { + "type": { + "type": "string", + "enum": ["running"] + } + }, + "required": ["type"], + "additionalProperties": false + }, "SessionNotFoundError": { "type": "object", "properties": {