fix(core): resume background completions

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)),

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<ReadonlySet<SessionSchema.ID>>
   /** Starts execution while idle or joins the active execution. */
   readonly resume: (sessionID: SessionSchema.ID) => Effect.Effect<void, SessionRunner.RunError>
-  /** Registers newly recorded work. Repeated wakeups may coalesce. */
-  readonly wake: (sessionID: SessionSchema.ID) => Effect.Effect<void>
+  /** Registers newly recorded work. Repeated wakeups may coalesce. Force runs even without pending input. */
+  readonly wake: (sessionID: SessionSchema.ID, options?: SessionRunCoordinator.WakeOptions) => Effect.Effect<void>
   /** Interrupt active work owned by this process. Idle interruption is a no-op. */
   readonly interrupt: (sessionID: SessionSchema.ID) => Effect.Effect<void>
   /** Resolves once this process owns no active execution for the Session. Returns immediately when idle and never starts work. */

packages/core/src/session/run-coordinator.ts

@@ -9,13 +9,15 @@ export interface Coordinator<Key, E, Reason = never> {
   /** Starts an execution while idle, or joins the active execution and returns its exit. */
   readonly run: (key: Key) => Effect.Effect<void, E>
   /** Rings the doorbell: an idle key starts an execution; an active one drains again before settling. */
-  readonly wake: (key: Key) => Effect.Effect<void>
+  readonly wake: (key: Key, options?: WakeOptions) => Effect.Effect<void>
   /** Stops the active execution, clears its doorbell, and waits for cleanup. No-op when idle. */
   readonly interrupt: (key: Key, reason?: Reason) => Effect.Effect<void>
   /** Resolves once no execution is active for the key. Returns immediately when already idle and never starts work. */
   readonly awaitIdle: (key: Key) => Effect.Effect<void>
 }
 
+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<E, Reason> = {
   readonly done: Deferred.Deferred<void, E>
   owner?: Fiber.Fiber<void>
   pendingWake: boolean
+  pendingForce: boolean
   stopping: boolean
   settling: boolean
   interruptionReason?: Reason
@@ -63,8 +66,10 @@ export const make = <Key, E, Reason = never>(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 = <Key, E, Reason = never>(options: {
       const execution: Execution<E, Reason> = {
         done: Deferred.makeUnsafe<void, E>(),
         pendingWake: false,
+        pendingForce: false,
         stopping: false,
         settling: false,
       }
@@ -100,7 +106,7 @@ export const make = <Key, E, Reason = never>(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<E, Reason>, exit: Exit.Exit<void, E>) => {
-      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 = <Key, E, Reason = never>(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<void> =>
@@ -132,6 +140,7 @@ export const make = <Key, E, Reason = never>(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)
       })

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<void>()
+        const firstGate = yield* Deferred.make<void>()
+        const secondStarted = yield* Deferred.make<void>()
+        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* () {

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<LLMEvent, LLMError> | undefined
+let responseStreams: Stream.Stream<LLMEvent, LLMError>[] | undefined
 let streamGate: Deferred.Deferred<void> | undefined
 let streamStarted: Deferred.Deferred<void> | 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<void>()
+      const releaseSecond = yield* Deferred.make<void>()
+      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