From d3ace428d6f5f749dc88975d63687d22759e0865 Mon Sep 17 00:00:00 2001 From: Kit Langton <7587245+kitlangton@users.noreply.github.com> Date: Sun, 28 Jun 2026 19:47:17 +0000 Subject: [PATCH] fix(core): share session runtime coordination --- packages/core/src/plugin/host.ts | 1 + packages/core/src/session.ts | 6 ++- .../core/src/session/runtime-coordinator.ts | 47 +++++++++++++++++++ packages/core/src/session/runtime.ts | 30 +++++++----- packages/plugin/src/v2/effect/session.ts | 16 +++---- 5 files changed, 80 insertions(+), 20 deletions(-) create mode 100644 packages/core/src/session/runtime-coordinator.ts diff --git a/packages/core/src/plugin/host.ts b/packages/core/src/plugin/host.ts index e52a27a09c..9c9c5691d4 100644 --- a/packages/core/src/plugin/host.ts +++ b/packages/core/src/plugin/host.ts @@ -217,6 +217,7 @@ export const make = Effect.fn("PluginHost.make")(function* (plugin: PluginV2.Int session.create( input.parentID ? { + id: input.id ? SessionV2.ID.make(input.id) : undefined, parentID: SessionV2.ID.make(input.parentID), title: input.title, agent: input.agent ? AgentV2.ID.make(input.agent) : undefined, diff --git a/packages/core/src/session.ts b/packages/core/src/session.ts index c63ffbf5a2..399cf89a82 100644 --- a/packages/core/src/session.ts +++ b/packages/core/src/session.ts @@ -25,6 +25,7 @@ import { ProjectTable } from "./project/sql" import path from "path" import { fromRow } from "./session/info" import { SessionRunner } from "./session/runner/index" +import { SessionRuntimeCoordinator } from "./session/runtime-coordinator" import { SessionStore } from "./session/store" import { makeGlobalNode } from "./effect/app-node" import { MessageDecodeError } from "./session/error" @@ -192,6 +193,7 @@ export const layer = Layer.effect( const db = database.db const events = yield* EventV2.Service const projects = yield* ProjectV2.Service + const runtimeCoordinator = yield* SessionRuntimeCoordinator.Service const store = yield* SessionStore.Service const decodeMessage = Schema.decodeUnknownEffect(SessionMessage.Message) const isDurableSessionEvent = Schema.is(SessionEvent.Durable) @@ -437,7 +439,7 @@ export const layer = Layer.effect( yield* result.get(sessionID) return yield* Effect.die("SessionV2.wait moved to SessionRuntime.Service") }), - active: Effect.succeed(new Set()), + active: runtimeCoordinator.active, resume: Effect.fn("V2Session.resume")(function* (sessionID) { yield* result.get(sessionID) return yield* Effect.die("SessionV2.resume moved to SessionRuntime.Service") @@ -469,6 +471,7 @@ export const defaultLayer = layer.pipe( Layer.provide(EventV2.defaultLayer), Layer.provide(Database.defaultLayer), Layer.provide(ProjectV2.defaultLayer), + Layer.provide(SessionRuntimeCoordinator.defaultLayer), Layer.orDie, ) @@ -494,6 +497,7 @@ export const node = makeGlobalNode({ Database.node, EventV2.node, ProjectV2.node, + SessionRuntimeCoordinator.node, SessionStore.node, SessionProjector.node, ], diff --git a/packages/core/src/session/runtime-coordinator.ts b/packages/core/src/session/runtime-coordinator.ts new file mode 100644 index 0000000000..0828b4c650 --- /dev/null +++ b/packages/core/src/session/runtime-coordinator.ts @@ -0,0 +1,47 @@ +export * as SessionRuntimeCoordinator from "./runtime-coordinator" + +import { Context, Effect, Layer } from "effect" +import { makeGlobalNode } from "../effect/app-node" +import { SessionRunner } from "./runner" +import { SessionRunCoordinator } from "./run-coordinator" +import { SessionSchema } from "./schema" + +type Drain = (force: boolean) => Effect.Effect + +export interface Interface { + readonly active: Effect.Effect> + readonly run: (sessionID: SessionSchema.ID, drain: Drain) => Effect.Effect + readonly wake: (sessionID: SessionSchema.ID, drain: Drain) => Effect.Effect + readonly interrupt: (sessionID: SessionSchema.ID) => Effect.Effect + readonly wait: (sessionID: SessionSchema.ID) => Effect.Effect +} + +export class Service extends Context.Service()("@opencode/v2/SessionRuntimeCoordinator") {} + +export const layer = Layer.effect( + Service, + Effect.gen(function* () { + const drains = new Map() + const coordinator = yield* SessionRunCoordinator.make({ + drain: Effect.fnUntraced(function* (sessionID, force) { + const drain = drains.get(sessionID) + if (!drain) return yield* Effect.die(`No SessionRuntime drain registered for ${sessionID}`) + return yield* drain(force) + }), + }) + + return Service.of({ + active: coordinator.active, + run: (sessionID, drain) => + Effect.sync(() => drains.set(sessionID, drain)).pipe(Effect.andThen(coordinator.run(sessionID))), + wake: (sessionID, drain) => + Effect.sync(() => drains.set(sessionID, drain)).pipe(Effect.andThen(coordinator.wake(sessionID))), + interrupt: coordinator.interrupt, + wait: coordinator.awaitIdle, + }) + }), +) + +export const defaultLayer = layer + +export const node = makeGlobalNode({ service: Service, layer, deps: [] }) diff --git a/packages/core/src/session/runtime.ts b/packages/core/src/session/runtime.ts index 1f1e60088c..f026331603 100644 --- a/packages/core/src/session/runtime.ts +++ b/packages/core/src/session/runtime.ts @@ -11,7 +11,7 @@ import { SessionInput } from "./input" import { SessionRevert } from "./revert" import { SessionRunner } from "./runner" import * as SessionRunnerLLM from "./runner/llm" -import { SessionRunCoordinator } from "./run-coordinator" +import { SessionRuntimeCoordinator } from "./runtime-coordinator" import { SessionSchema } from "./schema" import { Snapshot } from "../snapshot" import { FSUtil } from "../fs-util" @@ -59,6 +59,8 @@ export const layer = Layer.effect( const location = yield* Location.Service const sessions = yield* SessionV2.Service const runner = yield* SessionRunner.Service + const snapshot = yield* Snapshot.Service + const coordinator = yield* SessionRuntimeCoordinator.Service const local = Effect.fn("SessionRuntime.local")(function* (sessionID: SessionSchema.ID) { const session = yield* sessions.get(sessionID) @@ -67,12 +69,11 @@ export const layer = Layer.effect( return session }) - const coordinator = yield* SessionRunCoordinator.make({ - drain: Effect.fnUntraced(function* (sessionID: SessionSchema.ID, force) { + const drain = (sessionID: SessionSchema.ID) => + Effect.fnUntraced(function* (force: boolean) { yield* local(sessionID).pipe(Effect.orDie) return yield* runner.run({ sessionID, force }) - }), - }) + }) return Service.of({ prompt: Effect.fn("SessionRuntime.prompt")((input) => @@ -97,21 +98,24 @@ export const layer = Layer.effect( ) if (!SessionInput.equivalent(admitted, expected)) return yield* new PromptConflictError({ sessionID: input.sessionID, messageID }) - if (input.resume !== false) yield* coordinator.wake(admitted.sessionID) + if (input.resume !== false) yield* coordinator.wake(admitted.sessionID, drain(admitted.sessionID)) return admitted }), ), ), wait: Effect.fn("SessionRuntime.wait")(function* (sessionID) { yield* local(sessionID) - yield* coordinator.awaitIdle(sessionID) + yield* coordinator.wait(sessionID) }), active: coordinator.active, resume: Effect.fn("SessionRuntime.resume")(function* (sessionID) { yield* local(sessionID) - yield* coordinator.run(sessionID) + yield* coordinator.run(sessionID, drain(sessionID)) + }), + interrupt: Effect.fn("SessionRuntime.interrupt")(function* (sessionID) { + yield* local(sessionID) + yield* Effect.uninterruptible(coordinator.interrupt(sessionID)) }), - interrupt: Effect.fn("SessionRuntime.interrupt")((sessionID) => Effect.uninterruptible(coordinator.interrupt(sessionID))), revert: { stage: Effect.fn("SessionRuntime.revert.stage")(function* (input) { const session = yield* local(input.sessionID) @@ -119,12 +123,16 @@ export const layer = Layer.effect( return yield* SessionRevert.stage({ session, messageID: input.messageID, files: input.files }).pipe( Effect.provideService(Database.Service, database), Effect.provideService(EventV2.Service, events), + Effect.provideService(Snapshot.Service, snapshot), ) }), clear: Effect.fn("SessionRuntime.revert.clear")(function* (sessionID) { const session = yield* local(sessionID) if ((yield* coordinator.active).has(sessionID)) return yield* new BusyError({ sessionID }) - return yield* SessionRevert.clear(session).pipe(Effect.provideService(EventV2.Service, events)) + return yield* SessionRevert.clear(session).pipe( + Effect.provideService(EventV2.Service, events), + Effect.provideService(Snapshot.Service, snapshot), + ) }), commit: Effect.fn("SessionRuntime.revert.commit")(function* (sessionID) { const session = yield* local(sessionID) @@ -153,5 +161,5 @@ const resolvePrompt = (input: PromptInput.Prompt) => export const node = makeLocationNode({ service: Service, layer, - deps: [Database.node, EventV2.node, Location.node, SessionV2.node, SessionRunnerLLM.node, Snapshot.node], + deps: [Database.node, EventV2.node, Location.node, SessionV2.node, SessionRunnerLLM.node, Snapshot.node, SessionRuntimeCoordinator.node], }) diff --git a/packages/plugin/src/v2/effect/session.ts b/packages/plugin/src/v2/effect/session.ts index 6bc628f918..97ae1b38d8 100644 --- a/packages/plugin/src/v2/effect/session.ts +++ b/packages/plugin/src/v2/effect/session.ts @@ -8,23 +8,23 @@ export interface SessionDomain { readonly title?: string readonly agent?: string readonly model?: SessionV2Info["model"] - }) => Effect.Effect - readonly get: (sessionID: string) => Effect.Effect + }) => Effect.Effect + readonly get: (sessionID: string) => Effect.Effect readonly messages: (input: { readonly sessionID: string readonly limit?: number readonly order?: "asc" | "desc" readonly cursor?: { readonly id: string; readonly direction: "previous" | "next" } - }) => Effect.Effect> - readonly context: (sessionID: string) => Effect.Effect> + }) => Effect.Effect, unknown> + readonly context: (sessionID: string) => Effect.Effect, unknown> readonly prompt: (input: { readonly id?: string readonly sessionID: string readonly prompt: PromptInput readonly delivery?: "steer" | "queue" readonly resume?: boolean - }) => Effect.Effect - readonly resume: (sessionID: string) => Effect.Effect - readonly wait: (sessionID: string) => Effect.Effect - readonly interrupt: (sessionID: string) => Effect.Effect + }) => Effect.Effect + readonly resume: (sessionID: string) => Effect.Effect + readonly wait: (sessionID: string) => Effect.Effect + readonly interrupt: (sessionID: string) => Effect.Effect }