diff --git a/packages/core/src/session.ts b/packages/core/src/session.ts index ad53940971..ba2814ea53 100644 --- a/packages/core/src/session.ts +++ b/packages/core/src/session.ts @@ -689,9 +689,7 @@ const layer = Layer.effect( metadata: input.metadata, }) if (input.resume === false) return - yield* execution - .resume(input.sessionID) - .pipe(Effect.ignore, Effect.forkIn(scope, { startImmediately: true }), Effect.asVoid) + yield* execution.wake(input.sessionID, { force: true }) }), interrupt: Effect.fn("V2Session.interrupt")((sessionID) => Effect.uninterruptible(execution.interrupt(sessionID)), diff --git a/packages/core/src/session/execution.ts b/packages/core/src/session/execution.ts index 5dbba30776..7e8c0b39d8 100644 --- a/packages/core/src/session/execution.ts +++ b/packages/core/src/session/execution.ts @@ -4,6 +4,7 @@ import { Context, Effect, Layer } from "effect" import { LayerNode } from "../effect/layer-node" import { Node } from "../effect/app-node" import { SessionRunner } from "./runner/index" +import type { SessionRunCoordinator } from "./run-coordinator" import { SessionSchema } from "./schema" export interface Interface { @@ -11,8 +12,8 @@ export interface Interface { 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. */ - readonly wake: (sessionID: SessionSchema.ID) => Effect.Effect + /** Registers newly recorded work. Repeated wakeups may coalesce. Force runs even without pending input. */ + readonly wake: (sessionID: SessionSchema.ID, options?: SessionRunCoordinator.WakeOptions) => Effect.Effect /** Interrupt active work owned by this process. Idle interruption is a no-op. */ readonly interrupt: (sessionID: SessionSchema.ID) => Effect.Effect /** Resolves once this process owns no active execution for the Session. Returns immediately when idle and never starts work. */ diff --git a/packages/core/src/session/run-coordinator.ts b/packages/core/src/session/run-coordinator.ts index 1550280164..928345db69 100644 --- a/packages/core/src/session/run-coordinator.ts +++ b/packages/core/src/session/run-coordinator.ts @@ -9,13 +9,15 @@ export interface Coordinator { /** Starts an execution while idle, or joins the active execution and returns its exit. */ readonly run: (key: Key) => Effect.Effect /** Rings the doorbell: an idle key starts an execution; an active one drains again before settling. */ - readonly wake: (key: Key) => Effect.Effect + readonly wake: (key: Key, options?: WakeOptions) => Effect.Effect /** Stops the active execution, clears its doorbell, and waits for cleanup. No-op when idle. */ readonly interrupt: (key: Key, reason?: Reason) => Effect.Effect /** Resolves once no execution is active for the key. Returns immediately when already idle and never starts work. */ readonly awaitIdle: (key: Key) => Effect.Effect } +export type WakeOptions = { readonly force?: boolean } + /** * One execution is a busy period for one key: one fiber that drains from the first wake * until the key would stay idle. `pendingWake` is the doorbell: work recorded during the @@ -27,6 +29,7 @@ type Execution = { readonly done: Deferred.Deferred owner?: Fiber.Fiber pendingWake: boolean + pendingForce: boolean stopping: boolean settling: boolean interruptionReason?: Reason @@ -63,8 +66,10 @@ export const make = (options: { Effect.suspend(() => { if (execution.stopping || !execution.pendingWake) return Effect.void execution.pendingWake = false + const force = execution.pendingForce + execution.pendingForce = false // Trampoline so drains that complete synchronously cannot grow the stack. - return Effect.yieldNow.pipe(Effect.andThen(loop(key, execution, false))) + return Effect.yieldNow.pipe(Effect.andThen(loop(key, execution, force))) }), ), ) @@ -73,6 +78,7 @@ export const make = (options: { const execution: Execution = { done: Deferred.makeUnsafe(), pendingWake: false, + pendingForce: false, stopping: false, settling: false, } @@ -100,7 +106,7 @@ export const make = (options: { // A doorbell that survives the execution loop (rung after the loop decided to end, or // during failure or interruption cleanup) starts a fresh execution for the remaining work. const settle = (key: Key, execution: Execution, exit: Exit.Exit) => { - if (execution.pendingWake) start(key, false) + if (execution.pendingWake) start(key, execution.pendingForce) else executions.delete(key) Deferred.doneUnsafe(execution.done, exit) } @@ -116,14 +122,16 @@ export const make = (options: { return restore(Deferred.await(start(key, true).done)) }) - const wake = (key: Key) => + const wake = (key: Key, options?: { readonly force?: boolean }) => Effect.sync(() => { + const force = options?.force === true const execution = executions.get(key) if (execution !== undefined) { execution.pendingWake = true + execution.pendingForce ||= force return } - start(key, false) + start(key, force) }) const interrupt = (key: Key, reason?: Reason): Effect.Effect => @@ -132,6 +140,7 @@ export const make = (options: { if (execution?.owner === undefined || execution.stopping || execution.settling) return Effect.void execution.stopping = true execution.pendingWake = false + execution.pendingForce = false execution.interruptionReason = reason return Fiber.interrupt(execution.owner) }) diff --git a/packages/core/test/session-run-coordinator.test.ts b/packages/core/test/session-run-coordinator.test.ts index c566c3f33c..e3a504418c 100644 --- a/packages/core/test/session-run-coordinator.test.ts +++ b/packages/core/test/session-run-coordinator.test.ts @@ -191,6 +191,36 @@ describe("SessionRunCoordinator", () => { ), ) + it.effect("preserves a forced wake received during active execution", () => + Effect.scoped( + Effect.gen(function* () { + const firstStarted = yield* Deferred.make() + const firstGate = yield* Deferred.make() + const secondStarted = yield* Deferred.make() + const forces: boolean[] = [] + const coordinator = yield* SessionRunCoordinator.make({ + drain: (_key, force) => + Effect.sync(() => forces.push(force)).pipe( + Effect.flatMap(() => + forces.length === 1 + ? Deferred.succeed(firstStarted, undefined).pipe(Effect.andThen(Deferred.await(firstGate))) + : Deferred.succeed(secondStarted, undefined), + ), + ), + }) + + yield* coordinator.wake("session") + yield* Deferred.await(firstStarted) + yield* coordinator.wake("session", { force: true }) + yield* Deferred.succeed(firstGate, undefined) + yield* Deferred.await(secondStarted) + yield* coordinator.awaitIdle("session") + + expect(forces).toEqual([false, true]) + }), + ), + ) + it.effect("runs again when woken during the follow-up", () => Effect.scoped( Effect.gen(function* () { diff --git a/packages/core/test/session-runner.test.ts b/packages/core/test/session-runner.test.ts index ef2d24445a..daa75c1694 100644 --- a/packages/core/test/session-runner.test.ts +++ b/packages/core/test/session-runner.test.ts @@ -71,6 +71,7 @@ const requests: LLMRequest[] = [] let response: LLMEvent[] = [] let responses: LLMEvent[][] | undefined let responseStream: Stream.Stream | undefined +let responseStreams: Stream.Stream[] | undefined let streamGate: Deferred.Deferred | undefined let streamStarted: Deferred.Deferred | undefined let streamFailure: LLMError | undefined @@ -85,6 +86,7 @@ const client = Layer.succeed( prepare: () => Effect.die("unused"), stream: ((request: LLMRequest) => { requests.push(request) + if (responseStreams) return responseStreams.shift() ?? Stream.empty if (responseStream) { const stream = responseStream responseStream = undefined @@ -416,6 +418,7 @@ const setup = Effect.gen(function* () { responses = undefined streamFailure = undefined responseStream = undefined + responseStreams = undefined streamGate = undefined streamStarted = undefined toolExecutionGate = undefined @@ -768,6 +771,34 @@ describe("SessionRunnerLLM", () => { }), ) + it.effect("runs a follow-up when synthetic context arrives during an active continuation", () => + Effect.gen(function* () { + const session = yield* setup + const secondStarted = yield* Deferred.make() + const releaseSecond = yield* Deferred.make() + responseStreams = [ + Stream.fromIterable(reply.tool("call-echo", "echo", { text: "background started" })), + Stream.unwrap( + Deferred.succeed(secondStarted, undefined).pipe( + Effect.andThen(Deferred.await(releaseSecond)), + Effect.as(Stream.fromIterable(reply.stop())), + ), + ), + Stream.fromIterable(reply.text("Handled completion", "text-completion")), + ] + yield* admit(session, "Start background work") + const running = yield* session.resume(sessionID).pipe(Effect.forkChild({ startImmediately: true })) + yield* Deferred.await(secondStarted) + + yield* session.synthetic({ sessionID, text: "Background work completed" }) + yield* Deferred.succeed(releaseSecond, undefined) + yield* Fiber.join(running) + + expect(requests).toHaveLength(3) + expect(userTexts(requests[2])).toContain("Background work completed") + }), + ) + it.effect("streams one request with registry definitions from chronological V2 user history", () => Effect.gen(function* () { const session = yield* setup