From ef5c9f4931adf1d92677f372bc8d3b6a91058260 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Fri, 26 Jun 2026 05:33:35 +0200 Subject: [PATCH] feat(sdk): expose active sessions (#33991) --- CONTEXT.md | 1 + .../client/src/generated-effect/client.ts | 141 +++++++++--------- packages/client/src/generated/client.ts | 12 ++ packages/client/src/generated/types.ts | 2 + packages/client/test/effect.test.ts | 9 +- packages/client/test/promise.test.ts | 6 +- packages/core/src/session.ts | 2 + packages/core/src/session/execution.ts | 9 +- packages/core/src/session/execution/local.ts | 1 + packages/core/src/session/run-coordinator.ts | 4 +- packages/core/test/session-prompt.test.ts | 9 ++ .../core/test/session-run-coordinator.test.ts | 73 +++++++++ .../core/test/session-runner-recorded.test.ts | 1 + packages/core/test/session-runner.test.ts | 1 + .../test/server/httpapi-exercise/index.ts | 1 + packages/protocol/src/groups/session.ts | 16 ++ packages/sdk-next/test/embedded.test.ts | 2 + packages/server/src/handlers/session.ts | 10 ++ specs/v2/session.md | 7 + 19 files changed, 236 insertions(+), 71 deletions(-) diff --git a/CONTEXT.md b/CONTEXT.md index 762a1ba7c2..5e5955d344 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -174,6 +174,7 @@ _Avoid_: Response envelope - `sessions.messages(...)` returns a **Page** and uses the same cursor discipline as `sessions.list(...)`: the initial request supplies `sessionID`, ordering, and page size; continuation supplies `sessionID` plus only an opaque branded message cursor carrying ordering, page size, direction, and message anchor. Using a cursor with another Session is invalid. - `sessions.message({ sessionID, messageID })` is a required resource lookup. An unknown Session fails with `SessionNotFoundError`; a known Session with an absent or differently owned message fails with `MessageNotFoundError` without disclosing cross-Session ownership. Absence is not represented as `undefined` across the public HTTP boundary. - `sessions.interrupt({ sessionID })` first verifies that the durable Session exists, failing with `SessionNotFoundError` otherwise. For a known Session, interruption is idempotent: idle, already-settled, or locally unowned execution is a no-op. +- `sessions.active()` snapshots the current process's foreground Session drain registry as a record of Session IDs to `{ type: "running" }`. Missing IDs are inactive; background subagents and tasks do not make their parent Session active, and process restart clears the registry. - `sessions.context({ sessionID })` preserves the existing message-only operation. It returns projected conversational messages selected as Session context; it does not include or represent the complete provider request context, whose baseline system context and other contributions remain separate. - **Open question**: Should a future, separately named operation expose the complete provider request context, including baseline system context, selected source contributions, and context-epoch metadata? - `sessions.prompt(...)` exposes `resume?: boolean`. Omitting it preserves durable admission followed by an advisory execution wake; `resume: false` requests durable admit-only behavior. diff --git a/packages/client/src/generated-effect/client.ts b/packages/client/src/generated-effect/client.ts index 5d4a58464b..172eb4ca22 100644 --- a/packages/client/src/generated-effect/client.ts +++ b/packages/client/src/generated-effect/client.ts @@ -55,43 +55,49 @@ const Endpoint0_1 = (raw: RawClient["server.session"]) => (input?: Endpoint0_1In Effect.map((value) => value.data), ) -type Endpoint0_2Request = Parameters[0] -type Endpoint0_2Input = { readonly sessionID: Endpoint0_2Request["params"]["sessionID"] } -const Endpoint0_2 = (raw: RawClient["server.session"]) => (input: Endpoint0_2Input) => +const Endpoint0_2 = (raw: RawClient["server.session"]) => () => + raw["session.active"]({}).pipe( + Effect.mapError(mapClientError), + Effect.map((value) => value.data), + ) + +type Endpoint0_3Request = Parameters[0] +type Endpoint0_3Input = { readonly sessionID: Endpoint0_3Request["params"]["sessionID"] } +const Endpoint0_3 = (raw: RawClient["server.session"]) => (input: Endpoint0_3Input) => raw["session.get"]({ params: { sessionID: input.sessionID } }).pipe( Effect.mapError(mapClientError), Effect.map((value) => value.data), ) -type Endpoint0_3Request = Parameters[0] -type Endpoint0_3Input = { - readonly sessionID: Endpoint0_3Request["params"]["sessionID"] - readonly agent: Endpoint0_3Request["payload"]["agent"] +type Endpoint0_4Request = Parameters[0] +type Endpoint0_4Input = { + readonly sessionID: Endpoint0_4Request["params"]["sessionID"] + readonly agent: Endpoint0_4Request["payload"]["agent"] } -const Endpoint0_3 = (raw: RawClient["server.session"]) => (input: Endpoint0_3Input) => +const Endpoint0_4 = (raw: RawClient["server.session"]) => (input: Endpoint0_4Input) => raw["session.switchAgent"]({ params: { sessionID: input.sessionID }, payload: { agent: input.agent } }).pipe( Effect.mapError(mapClientError), ) -type Endpoint0_4Request = Parameters[0] -type Endpoint0_4Input = { - readonly sessionID: Endpoint0_4Request["params"]["sessionID"] - readonly model: Endpoint0_4Request["payload"]["model"] +type Endpoint0_5Request = Parameters[0] +type Endpoint0_5Input = { + readonly sessionID: Endpoint0_5Request["params"]["sessionID"] + readonly model: Endpoint0_5Request["payload"]["model"] } -const Endpoint0_4 = (raw: RawClient["server.session"]) => (input: Endpoint0_4Input) => +const Endpoint0_5 = (raw: RawClient["server.session"]) => (input: Endpoint0_5Input) => raw["session.switchModel"]({ params: { sessionID: input.sessionID }, payload: { model: input.model } }).pipe( Effect.mapError(mapClientError), ) -type Endpoint0_5Request = Parameters[0] -type Endpoint0_5Input = { - readonly sessionID: Endpoint0_5Request["params"]["sessionID"] - readonly id?: Endpoint0_5Request["payload"]["id"] - readonly prompt: Endpoint0_5Request["payload"]["prompt"] - readonly delivery?: Endpoint0_5Request["payload"]["delivery"] - readonly resume?: Endpoint0_5Request["payload"]["resume"] +type Endpoint0_6Request = Parameters[0] +type Endpoint0_6Input = { + readonly sessionID: Endpoint0_6Request["params"]["sessionID"] + readonly id?: Endpoint0_6Request["payload"]["id"] + readonly prompt: Endpoint0_6Request["payload"]["prompt"] + readonly delivery?: Endpoint0_6Request["payload"]["delivery"] + readonly resume?: Endpoint0_6Request["payload"]["resume"] } -const Endpoint0_5 = (raw: RawClient["server.session"]) => (input: Endpoint0_5Input) => +const Endpoint0_6 = (raw: RawClient["server.session"]) => (input: Endpoint0_6Input) => raw["session.prompt"]({ params: { sessionID: input.sessionID }, payload: { id: input.id, prompt: input.prompt, delivery: input.delivery, resume: input.resume }, @@ -100,23 +106,23 @@ const Endpoint0_5 = (raw: RawClient["server.session"]) => (input: Endpoint0_5Inp Effect.map((value) => value.data), ) -type Endpoint0_6Request = Parameters[0] -type Endpoint0_6Input = { readonly sessionID: Endpoint0_6Request["params"]["sessionID"] } -const Endpoint0_6 = (raw: RawClient["server.session"]) => (input: Endpoint0_6Input) => - raw["session.compact"]({ params: { sessionID: input.sessionID } }).pipe(Effect.mapError(mapClientError)) - -type Endpoint0_7Request = Parameters[0] +type Endpoint0_7Request = Parameters[0] type Endpoint0_7Input = { readonly sessionID: Endpoint0_7Request["params"]["sessionID"] } const Endpoint0_7 = (raw: RawClient["server.session"]) => (input: Endpoint0_7Input) => + raw["session.compact"]({ params: { sessionID: input.sessionID } }).pipe(Effect.mapError(mapClientError)) + +type Endpoint0_8Request = Parameters[0] +type Endpoint0_8Input = { readonly sessionID: Endpoint0_8Request["params"]["sessionID"] } +const Endpoint0_8 = (raw: RawClient["server.session"]) => (input: Endpoint0_8Input) => raw["session.wait"]({ params: { sessionID: input.sessionID } }).pipe(Effect.mapError(mapClientError)) -type Endpoint0_8Request = Parameters[0] -type Endpoint0_8Input = { - readonly sessionID: Endpoint0_8Request["params"]["sessionID"] - readonly messageID: Endpoint0_8Request["payload"]["messageID"] - readonly files?: Endpoint0_8Request["payload"]["files"] +type Endpoint0_9Request = Parameters[0] +type Endpoint0_9Input = { + readonly sessionID: Endpoint0_9Request["params"]["sessionID"] + readonly messageID: Endpoint0_9Request["payload"]["messageID"] + readonly files?: Endpoint0_9Request["payload"]["files"] } -const Endpoint0_8 = (raw: RawClient["server.session"]) => (input: Endpoint0_8Input) => +const Endpoint0_9 = (raw: RawClient["server.session"]) => (input: Endpoint0_9Input) => raw["session.revert.stage"]({ params: { sessionID: input.sessionID }, payload: { messageID: input.messageID, files: input.files }, @@ -125,30 +131,30 @@ const Endpoint0_8 = (raw: RawClient["server.session"]) => (input: Endpoint0_8Inp Effect.map((value) => value.data), ) -type Endpoint0_9Request = Parameters[0] -type Endpoint0_9Input = { readonly sessionID: Endpoint0_9Request["params"]["sessionID"] } -const Endpoint0_9 = (raw: RawClient["server.session"]) => (input: Endpoint0_9Input) => - raw["session.revert.clear"]({ params: { sessionID: input.sessionID } }).pipe(Effect.mapError(mapClientError)) - -type Endpoint0_10Request = Parameters[0] +type Endpoint0_10Request = Parameters[0] type Endpoint0_10Input = { readonly sessionID: Endpoint0_10Request["params"]["sessionID"] } const Endpoint0_10 = (raw: RawClient["server.session"]) => (input: Endpoint0_10Input) => - raw["session.revert.commit"]({ params: { sessionID: input.sessionID } }).pipe(Effect.mapError(mapClientError)) + raw["session.revert.clear"]({ params: { sessionID: input.sessionID } }).pipe(Effect.mapError(mapClientError)) -type Endpoint0_11Request = Parameters[0] +type Endpoint0_11Request = Parameters[0] type Endpoint0_11Input = { readonly sessionID: Endpoint0_11Request["params"]["sessionID"] } const Endpoint0_11 = (raw: RawClient["server.session"]) => (input: Endpoint0_11Input) => + raw["session.revert.commit"]({ params: { sessionID: input.sessionID } }).pipe(Effect.mapError(mapClientError)) + +type Endpoint0_12Request = Parameters[0] +type Endpoint0_12Input = { readonly sessionID: Endpoint0_12Request["params"]["sessionID"] } +const Endpoint0_12 = (raw: RawClient["server.session"]) => (input: Endpoint0_12Input) => raw["session.context"]({ params: { sessionID: input.sessionID } }).pipe( Effect.mapError(mapClientError), Effect.map((value) => value.data), ) -type Endpoint0_12Request = Parameters[0] -type Endpoint0_12Input = { - readonly sessionID: Endpoint0_12Request["params"]["sessionID"] - readonly after?: Endpoint0_12Request["query"]["after"] +type Endpoint0_13Request = Parameters[0] +type Endpoint0_13Input = { + readonly sessionID: Endpoint0_13Request["params"]["sessionID"] + readonly after?: Endpoint0_13Request["query"]["after"] } -const Endpoint0_12 = (raw: RawClient["server.session"]) => (input: Endpoint0_12Input) => +const Endpoint0_13 = (raw: RawClient["server.session"]) => (input: Endpoint0_13Input) => Stream.unwrap( raw["session.events"]({ params: { sessionID: input.sessionID }, query: { after: input.after } }).pipe( Effect.mapError(mapClientError), @@ -156,17 +162,17 @@ const Endpoint0_12 = (raw: RawClient["server.session"]) => (input: Endpoint0_12I ), ) -type Endpoint0_13Request = Parameters[0] -type Endpoint0_13Input = { readonly sessionID: Endpoint0_13Request["params"]["sessionID"] } -const Endpoint0_13 = (raw: RawClient["server.session"]) => (input: Endpoint0_13Input) => +type Endpoint0_14Request = Parameters[0] +type Endpoint0_14Input = { readonly sessionID: Endpoint0_14Request["params"]["sessionID"] } +const Endpoint0_14 = (raw: RawClient["server.session"]) => (input: Endpoint0_14Input) => raw["session.interrupt"]({ params: { sessionID: input.sessionID } }).pipe(Effect.mapError(mapClientError)) -type Endpoint0_14Request = Parameters[0] -type Endpoint0_14Input = { - readonly sessionID: Endpoint0_14Request["params"]["sessionID"] - readonly messageID: Endpoint0_14Request["params"]["messageID"] +type Endpoint0_15Request = Parameters[0] +type Endpoint0_15Input = { + readonly sessionID: Endpoint0_15Request["params"]["sessionID"] + readonly messageID: Endpoint0_15Request["params"]["messageID"] } -const Endpoint0_14 = (raw: RawClient["server.session"]) => (input: Endpoint0_14Input) => +const Endpoint0_15 = (raw: RawClient["server.session"]) => (input: Endpoint0_15Input) => raw["session.message"]({ params: { sessionID: input.sessionID, messageID: input.messageID } }).pipe( Effect.mapError(mapClientError), Effect.map((value) => value.data), @@ -175,19 +181,20 @@ const Endpoint0_14 = (raw: RawClient["server.session"]) => (input: Endpoint0_14I const adaptGroup0 = (raw: RawClient["server.session"]) => ({ list: Endpoint0_0(raw), create: Endpoint0_1(raw), - get: Endpoint0_2(raw), - switchAgent: Endpoint0_3(raw), - switchModel: Endpoint0_4(raw), - prompt: Endpoint0_5(raw), - compact: Endpoint0_6(raw), - wait: Endpoint0_7(raw), - stage: Endpoint0_8(raw), - clear: Endpoint0_9(raw), - commit: Endpoint0_10(raw), - context: Endpoint0_11(raw), - events: Endpoint0_12(raw), - interrupt: Endpoint0_13(raw), - message: Endpoint0_14(raw), + active: Endpoint0_2(raw), + get: Endpoint0_3(raw), + switchAgent: Endpoint0_4(raw), + switchModel: Endpoint0_5(raw), + prompt: Endpoint0_6(raw), + compact: Endpoint0_7(raw), + wait: Endpoint0_8(raw), + stage: Endpoint0_9(raw), + clear: Endpoint0_10(raw), + commit: Endpoint0_11(raw), + context: Endpoint0_12(raw), + events: Endpoint0_13(raw), + interrupt: Endpoint0_14(raw), + message: Endpoint0_15(raw), }) const adaptClient = (raw: RawClient) => ({ sessions: adaptGroup0(raw["server.session"]) }) diff --git a/packages/client/src/generated/client.ts b/packages/client/src/generated/client.ts index b1aa2a2405..2d3cd92ee0 100644 --- a/packages/client/src/generated/client.ts +++ b/packages/client/src/generated/client.ts @@ -3,6 +3,7 @@ import type { SessionsListOutput, SessionsCreateInput, SessionsCreateOutput, + SessionsActiveOutput, SessionsGetInput, SessionsGetOutput, SessionsSwitchAgentInput, @@ -198,6 +199,17 @@ export function make(options: ClientOptions) { }, requestOptions, ).then((value) => value.data), + active: (requestOptions?: RequestOptions) => + request<{ readonly data: SessionsActiveOutput }>( + { + method: "GET", + path: `/api/session/active`, + successStatus: 200, + declaredStatuses: [401, 400], + empty: false, + }, + requestOptions, + ).then((value) => value.data), get: (input: SessionsGetInput, requestOptions?: RequestOptions) => request<{ readonly data: SessionsGetOutput }>( { diff --git a/packages/client/src/generated/types.ts b/packages/client/src/generated/types.ts index 858756e2a4..f426a50f87 100644 --- a/packages/client/src/generated/types.ts +++ b/packages/client/src/generated/types.ts @@ -243,6 +243,8 @@ export type SessionsCreateOutput = { } }["data"] +export type SessionsActiveOutput = { readonly data: { readonly [x: string]: { readonly type: "running" } } }["data"] + export type SessionsGetInput = { readonly sessionID: { readonly sessionID: string }["sessionID"] } export type SessionsGetOutput = { diff --git a/packages/client/test/effect.test.ts b/packages/client/test/effect.test.ts index f9fbf1b96f..6ca73c22f4 100644 --- a/packages/client/test/effect.test.ts +++ b/packages/client/test/effect.test.ts @@ -37,6 +37,11 @@ test("session methods retain decoded Effect inputs and outputs", async () => { if (url.includes("/message/")) { return Effect.succeed(HttpClientResponse.fromWeb(request, Response.json({ data: modelSwitchedMessage }))) } + if (url.endsWith("/api/session/active")) { + return Effect.succeed( + HttpClientResponse.fromWeb(request, Response.json({ data: { ses_test: { type: "running" } } })), + ) + } if (request.method === "POST" && url.endsWith("/api/session")) { return Effect.succeed(HttpClientResponse.fromWeb(request, Response.json(session))) } @@ -50,6 +55,7 @@ test("session methods retain decoded Effect inputs and outputs", async () => { const result = await Effect.gen(function* () { const client = yield* OpenCode.make({ baseUrl: "http://localhost:3000" }) const page = yield* client.sessions.list({ limit: 10 }) + const active = yield* client.sessions.active() const created = yield* client.sessions.create({ location: Location.Ref.make({ directory: AbsolutePath.make("/tmp/project") }), }) @@ -74,10 +80,11 @@ test("session methods retain decoded Effect inputs and outputs", async () => { sessionID: Session.ID.make("ses_test"), messageID: SessionMessage.ID.make("msg_model"), }) - return { page, created, admitted, context, events, message } + return { page, active, created, admitted, context, events, message } }).pipe(Effect.provideService(HttpClient.HttpClient, httpClient), Effect.runPromise) expect(DateTime.toEpochMillis(result.page.data[0].time.created)).toBe(1_717_171_717_000) + expect(result.active).toEqual({ ses_test: { type: "running" } }) expect(Object.getPrototypeOf(result.page.data[0])).toBe(Object.prototype) expect(Object.getPrototypeOf(result.created)).toBe(Object.prototype) expect(result.created.id).toBe("ses_test") diff --git a/packages/client/test/promise.test.ts b/packages/client/test/promise.test.ts index 16a475b443..952a3fc0d7 100644 --- a/packages/client/test/promise.test.ts +++ b/packages/client/test/promise.test.ts @@ -32,6 +32,7 @@ test("session methods use the public HTTP contract", async () => { if (url.includes("/prompt")) return Response.json(admission) if (url.includes("/context")) return Response.json({ data: [] }) if (url.includes("/message/")) return Response.json({ data: modelSwitchedMessage }) + if (url.endsWith("/api/session/active")) return Response.json({ data: { ses_test: { type: "running" } } }) if (init?.method === "POST" && url.endsWith("/api/session")) return Response.json(session) if (init?.method === "POST") return new Response(null, { status: 204 }) return Response.json({ data: [session.data], cursor: { next: "next" } }) @@ -39,6 +40,7 @@ test("session methods use the public HTTP contract", async () => { }) const page = await client.sessions.list({ limit: "10", order: "desc" }) + const active = await client.sessions.active() const created = await client.sessions.create({ location: { directory: "/tmp/project" } }) await client.sessions.switchAgent({ sessionID: "ses_test", agent: "build" }) await client.sessions.switchModel({ @@ -59,6 +61,7 @@ test("session methods use the public HTTP contract", async () => { const message = await client.sessions.message({ sessionID: "ses_test", messageID: "msg_model" }) expect(page.cursor.next).toBe("next") + expect(active).toEqual({ ses_test: { type: "running" } }) expect(created.id).toBe("ses_test") expect(admitted.id).toBe("msg_test") expect(context).toEqual([]) @@ -66,6 +69,7 @@ test("session methods use the public HTTP contract", async () => { expect(message).toEqual(modelSwitchedMessage) expect(requests.map((request) => [request.init?.method, request.url])).toEqual([ ["GET", "http://localhost:3000/api/session?limit=10&order=desc"], + ["GET", "http://localhost:3000/api/session/active"], ["POST", "http://localhost:3000/api/session"], ["POST", "http://localhost:3000/api/session/ses_test/agent"], ["POST", "http://localhost:3000/api/session/ses_test/model"], @@ -77,7 +81,7 @@ test("session methods use the public HTTP contract", async () => { ["POST", "http://localhost:3000/api/session/ses_test/interrupt"], ["GET", "http://localhost:3000/api/session/ses_test/message/msg_model"], ]) - const body = requests[4]?.init?.body + const body = requests.find((request) => request.url.endsWith("/api/session/ses_test/prompt"))?.init?.body if (typeof body !== "string") throw new Error("Expected JSON request body") expect(JSON.parse(body)).toEqual({ prompt: { text: "Hello" }, diff --git a/packages/core/src/session.ts b/packages/core/src/session.ts index 81b96ab661..97745b41bf 100644 --- a/packages/core/src/session.ts +++ b/packages/core/src/session.ts @@ -155,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: { @@ -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) diff --git a/packages/core/src/session/execution.ts b/packages/core/src/session/execution.ts index a08912e956..f92929e6ee 100644 --- a/packages/core/src/session/execution.ts +++ b/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> /** Starts execution while idle or joins the active execution. */ readonly resume: (sessionID: SessionSchema.ID) => Effect.Effect /** Registers newly recorded work. Repeated wakeups may coalesce. */ @@ -19,5 +21,10 @@ export class Service extends Context.Service()("@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, + }), ) diff --git a/packages/core/src/session/execution/local.ts b/packages/core/src/session/execution/local.ts index 7e0e3ca700..bd1be74539 100644 --- a/packages/core/src/session/execution/local.ts +++ b/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, diff --git a/packages/core/src/session/run-coordinator.ts b/packages/core/src/session/run-coordinator.ts index cfab42300e..2f89aff9e3 100644 --- a/packages/core/src/session/run-coordinator.ts +++ b/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 { + /** Snapshots keys with an execution owned by this coordinator. */ + readonly active: Effect.Effect> /** Starts execution while idle or joins the active execution. */ readonly run: (key: Key) => Effect.Effect /** Registers one coalesced follow-up after newly recorded work. */ @@ -98,5 +100,5 @@ export const make = (options: { return Fiber.interrupt(entry.owner) }) - return { run, wake, interrupt } + return { active: Effect.sync(() => new Set(active.keys())), run, wake, interrupt } }) diff --git a/packages/core/test/session-prompt.test.ts b/packages/core/test/session-prompt.test.ts index af30c4d877..9ff0e22490 100644 --- a/packages/core/test/session-prompt.test.ts +++ b/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() 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 diff --git a/packages/core/test/session-run-coordinator.test.ts b/packages/core/test/session-run-coordinator.test.ts index 909ba64870..dfbeda664c 100644 --- a/packages/core/test/session-run-coordinator.test.ts +++ b/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() + const secondStarted = yield* Deferred.make() + const firstGate = yield* Deferred.make() + const secondGate = yield* Deferred.make() + 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() + 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) }), ), diff --git a/packages/core/test/session-runner-recorded.test.ts b/packages/core/test/session-runner-recorded.test.ts index 077ee48823..da8699fbd2 100644 --- a/packages/core/test/session-runner-recorded.test.ts +++ b/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, diff --git a/packages/core/test/session-runner.test.ts b/packages/core/test/session-runner.test.ts index 6df50ac575..c9594e8244 100644 --- a/packages/core/test/session-runner.test.ts +++ b/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, diff --git a/packages/opencode/test/server/httpapi-exercise/index.ts b/packages/opencode/test/server/httpapi-exercise/index.ts index 692a63defd..d3c8981a64 100644 --- a/packages/opencode/test/server/httpapi-exercise/index.ts +++ b/packages/opencode/test/server/httpapi-exercise/index.ts @@ -964,6 +964,7 @@ const scenarios: Scenario[] = [ headers: ctx.headers(), })) .status(400, undefined, "none"), + http.protected.get("/api/session/active", "v2.session.active").json(200, data(object), "none"), http.protected .post("/api/session", "v2.session.create") .at((ctx) => ({ diff --git a/packages/protocol/src/groups/session.ts b/packages/protocol/src/groups/session.ts index 5e799091a9..f1f908e6ac 100644 --- a/packages/protocol/src/groups/session.ts +++ b/packages/protocol/src/groups/session.ts @@ -73,6 +73,10 @@ export const SessionsCursor = Schema.String.pipe( ) export type SessionsCursor = typeof SessionsCursor.Type +const SessionActive = Schema.Struct({ + type: Schema.Literal("running"), +}).annotate({ identifier: "SessionActive" }) + const SessionsQueryCursor = SessionsCursor.annotate({ description: "Opaque pagination cursor returned as cursor.previous or cursor.next in the previous response.", }) @@ -124,6 +128,18 @@ export const makeSessionGroup = (sessionLo }), ), ) + .add( + HttpApiEndpoint.get("session.active", "/api/session/active", { + success: Schema.Struct({ data: Schema.Record(Session.ID, SessionActive) }), + }).annotateMerge( + OpenApi.annotations({ + identifier: "v2.session.active", + summary: "List active sessions", + description: + "Retrieve foreground Session drains currently owned by this OpenCode process. Sessions absent from the result are inactive.", + }), + ), + ) .add( HttpApiEndpoint.get("session.get", "/api/session/:sessionID", { params: { sessionID: Session.ID }, diff --git a/packages/sdk-next/test/embedded.test.ts b/packages/sdk-next/test/embedded.test.ts index 5ba505cbca..dbd7ed9378 100644 --- a/packages/sdk-next/test/embedded.test.ts +++ b/packages/sdk-next/test/embedded.test.ts @@ -33,6 +33,7 @@ test("embedded client uses the real router and handlers", async () => { yield* opencode.sessions.switchModel({ sessionID, model }) const selected = yield* opencode.sessions.get({ sessionID }) const page = yield* opencode.sessions.list({ directory: AbsolutePath.make(directory) }) + const active = yield* opencode.sessions.active() const admitted = yield* opencode.sessions.prompt({ sessionID, prompt: Prompt.make({ text: "Do not run" }), @@ -81,6 +82,7 @@ test("embedded client uses the real router and handlers", async () => { expect(selected.model?.id).toBe(model.id) expect(selected.model?.providerID).toBe(model.providerID) expect(page.data.some((session) => session.id === sessionID)).toBe(true) + expect(active).toEqual({}) expect(admitted.sessionID).toBe(sessionID) expect(prompted.type).toBe("session.next.prompted") expect(wakeContext).toContainEqual(expect.objectContaining({ id: wake.id, type: "user" })) diff --git a/packages/server/src/handlers/session.ts b/packages/server/src/handlers/session.ts index e3b1270b94..602673e1ff 100644 --- a/packages/server/src/handlers/session.ts +++ b/packages/server/src/handlers/session.ts @@ -76,6 +76,16 @@ export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handl } }), ) + .handle( + "session.active", + Effect.fn(function* () { + return { + data: Object.fromEntries( + Array.from(yield* session.active, (sessionID) => [sessionID, { type: "running" as const }]), + ), + } + }), + ) .handle( "session.get", Effect.fn(function* (ctx) { diff --git a/specs/v2/session.md b/specs/v2/session.md index 6b16dddee1..54daa69713 100644 --- a/specs/v2/session.md +++ b/specs/v2/session.md @@ -25,6 +25,11 @@ sessions.interrupt(sessionID) -> clears a coalesced follow-up wake already registered with this coordinator -> preserves durable inbox rows for a later wake or resume -> idle or missing Session is a no-op + +sessions.active() + -> snapshots foreground Session drains owned by this process + -> returns only active Session IDs with { type: "running" } + -> absence means inactive; activity is not durable across process restarts ``` `session_input` is the durable admission inbox. `PromptAdmitted` records and projects accepted input so pending queue state can be replayed, replicated, and observed by clients. Admitted inputs remain outside model-visible Session history until the serialized runner publishes `Prompted`. Its projector atomically writes the visible user message and marks the inbox row promoted in the same event transaction. The V1-to-V2 shadow bridge publishes the same `Prompted` event for already-visible V1 prompts. @@ -161,6 +166,8 @@ Post-crash continuation recovery is intentionally deferred. A wake does not infe A process-global `SessionRunCoordinator` serializes execution for each local Session while allowing different Sessions to run concurrently. Resumes join active execution, overlapping wakes coalesce into one follow-up, and interruption stops current process-local execution without deleting durable inbox work. The runner enters the Session's current Location when execution starts and fences each new provider turn against that Location. +The coordinator's active registry is also the source for `sessions.active()`. It represents only foreground Session drains owned by the current process; background subagents and tasks do not add parent Sessions to this registry. The snapshot is runtime state and is empty after a process restart. + Inbox promotion coalesces pending steers in durable admission order. Once continuation would otherwise end, it promotes one queued input at a time in FIFO order. Add explicit inbox backlog and steering-batch limits before exposing broad multi-caller admission or untrusted queue growth. Eager local-tool execution is intentionally unbounded in the current local slice. This minimizes tool latency but does not increase SQLite settlement throughput: Session-event publication remains serialized per provider turn. Before broadening exposure, revisit per-turn call limits, output truncation, and operational backpressure using observed workloads. The `session.next.*` event schemas remain experimental and unshipped; databases created by earlier experimental builds are disposable rather than compatibility targets.