fix(core): continue sessions after location move (#42138)

packages/core/src/session.ts

@@ -748,7 +748,7 @@ const layer = Layer.effect(
             projectID: project.id,
             subpath: RelativePath.make(path.relative(project.directory, directory).replaceAll("\\", "/")),
           },
-          delivery: input.delivery ?? "queue",
+          delivery: input.delivery ?? "steer",
         })
         const inboxID = SessionMessage.ID.create()
         yield* SessionInbox.admit(db, bus, {

packages/core/src/session/execution.ts

@@ -67,16 +67,17 @@ export const layer = Layer.effect(
     const releaseOnCommit = (sessionID: SessionSchema.ID) => ({
       commit: () => store.release(sessionID),
     })
-    const coordinator = yield* SessionRunCoordinator.make<SessionSchema.ID, SessionRunner.RunError, InterruptReason>({
-      started: (sessionID) =>
-        reportLifecycle(
-          sessionID,
-          bus.publish(SessionEvent.Execution.Started, { sessionID }, claimOnCommit(sessionID)),
-        ),
-      drain: Effect.fnUntraced(function* (sessionID: SessionSchema.ID, force) {
+    function drain(
+      sessionID: SessionSchema.ID,
+      force: boolean,
+      continuation?: SessionRunner.Continuation,
+    ): Effect.Effect<void, SessionRunner.RunError> {
+      return Effect.gen(function* () {
         const session = yield* store.get(sessionID)
         if (!session) return yield* Effect.die(new Error(`Session not found: ${sessionID}`))
-        return yield* SessionRunner.Service.use((runner) => runner.drain({ sessionID, force })).pipe(
+        const result = yield* SessionRunner.Service.use((runner) =>
+          runner.drain({ sessionID, force, continuation }),
+        ).pipe(
           Effect.provide(locations.get(session.location)),
           Effect.tapCause((cause) =>
             Cause.hasInterruptsOnly(cause)
@@ -84,7 +85,17 @@ export const layer = Layer.effect(
               : Effect.logError("Failed to drain Session", cause).pipe(Effect.annotateLogs({ sessionID })),
           ),
         )
-      }),
+        if (result.type === "complete") return
+        return yield* drain(sessionID, false, result.continuation)
+      })
+    }
+    const coordinator = yield* SessionRunCoordinator.make<SessionSchema.ID, SessionRunner.RunError, InterruptReason>({
+      started: (sessionID) =>
+        reportLifecycle(
+          sessionID,
+          bus.publish(SessionEvent.Execution.Started, { sessionID }, claimOnCommit(sessionID)),
+        ),
+      drain: (sessionID, force) => drain(sessionID, force),
       // One terminal observation per busy period, covering every coalesced drain.
       settled: (sessionID, exit, reason) =>
         reportLifecycle(

packages/core/src/session/runner/index.ts

@@ -16,13 +16,20 @@ export type RunError =
   | UserInterruptedError
   | Instructions.InitializationBlocked
 
+export type Continuation = { readonly step: number }
+
+export type DrainResult =
+  | { readonly type: "complete" }
+  | { readonly type: "moved"; readonly continuation?: Continuation }
+
 /** Runs one local continuation from already-recorded Session history. */
 export interface Interface {
-  /** Drains eligible durable work. Explicit runs make one model call even when no work is eligible. */
+  /** Drains eligible durable work, returning transient state when execution must continue at a new Location. */
   readonly drain: (input: {
     readonly sessionID: SessionSchema.ID
     readonly force: boolean
-  }) => Effect.Effect<void, RunError>
+    readonly continuation?: Continuation
+  }) => Effect.Effect<DrainResult, RunError>
 }
 
 export class Service extends Context.Service<Service, Interface>()("@opencode/SessionRunner") {}

packages/core/src/session/runner/llm.ts

@@ -23,7 +23,7 @@ import { SessionMessage } from "../message.js"
 import { SessionSchema } from "../schema.js"
 import { SessionStore } from "../store.js"
 import { SessionTitle } from "../title.js"
-import { Service } from "./index.js"
+import { Service, type Continuation } from "./index.js"
 import { createLLMEventPublisher, type StepRecord } from "./publish-llm-event.js"
 import { Snapshot } from "../../snapshot.js"
 import { makeLocationNode } from "@opencode-ai/util/effect/app-node"
@@ -124,19 +124,25 @@ const layer = Layer.effect(
     const drain = Effect.fn("SessionRunner.drain")(function* (input: {
       readonly sessionID: SessionSchema.ID
       readonly force: boolean
+      readonly continuation?: Continuation
     }) {
       let force = input.force
-      if (!force && !(yield* SessionInbox.has(db, input.sessionID, "any"))) return
+      let continuation = input.continuation
+      if (!force && !continuation && !(yield* SessionInbox.has(db, input.sessionID, "any")))
+        return { type: "complete" as const }
       yield* settleStaleToolCalls(input.sessionID)
       while (true) {
         if (yield* runPendingCompaction(input.sessionID)) {
           force = false
           continue
         }
-        if (yield* runPendingMove(input.sessionID)) return
-        if (!force && !(yield* SessionInbox.has(db, input.sessionID, "input"))) return
-        if (yield* runSteps(input.sessionID)) return
+        if (yield* runPendingMove(input.sessionID, "input")) return { type: "moved" as const }
+        if (!force && !continuation && !(yield* SessionInbox.has(db, input.sessionID, "input")))
+          return { type: "complete" as const }
+        const result = yield* runSteps(input.sessionID, continuation)
+        if (result.type === "moved") return result
         force = false
+        continuation = undefined
       }
     })
 
@@ -144,15 +150,21 @@ const layer = Layer.effect(
      * Runs logical steps until no tool result or newly admitted steer requires another
      * model call. Queued inputs remain pending until the current model work reaches idle.
      */
-    const runSteps = Effect.fn("SessionRunner.runSteps")(function* (sessionID: SessionSchema.ID) {
+    const runSteps = Effect.fn("SessionRunner.runSteps")(function* (
+      sessionID: SessionSchema.ID,
+      continuation?: Continuation,
+    ) {
       // Fresh work may promote queued input; later steps absorb steers only.
-      let promotable: SessionInbox.Promotable = "input"
-      let step = 1
+      let promotable: SessionInbox.Promotable = continuation ? "steer" : "input"
+      let step = continuation?.step ?? 1
+      let next = continuation
       while (true) {
         if (yield* runPendingCompaction(sessionID)) continue
-        if (yield* runPendingMove(sessionID)) return true
+        if (yield* runPendingMove(sessionID, "steer")) return { type: "moved" as const, continuation: next }
         const result = yield* runStep(sessionID, promotable, step)
-        if (!result.needsContinuation && !(yield* SessionInbox.has(db, sessionID, "steer"))) return false
+        next = result.needsContinuation ? { step: result.step + 1 } : undefined
+        if (!result.needsContinuation && !(yield* SessionInbox.has(db, sessionID, "steer")))
+          return { type: "complete" as const }
         promotable = "steer"
         step = result.step + 1
       }
@@ -530,12 +542,16 @@ const layer = Layer.effect(
       )
     })
 
-    const runPendingMove = Effect.fn("SessionRunner.runPendingMove")(function* (sessionID: SessionSchema.ID) {
+    const runPendingMove = Effect.fn("SessionRunner.runPendingMove")(function* (
+      sessionID: SessionSchema.ID,
+      promotable: SessionInbox.Promotable,
+    ) {
       return yield* SessionInbox.serialized(
         sessionID,
         Effect.gen(function* () {
           const pending =
-            (yield* SessionInbox.nextSteer(db, sessionID)) ?? (yield* SessionInbox.nextQueued(db, sessionID))
+            (yield* SessionInbox.nextSteer(db, sessionID)) ??
+            (promotable === "input" ? yield* SessionInbox.nextQueued(db, sessionID) : undefined)
           if (pending?.type !== "move") return false
           yield* bus.publishAll([
             [SessionEvent.InboxDelivered, { sessionID, inboxID: pending.id }],

packages/core/test/session-execution.test.ts

@@ -346,14 +346,19 @@ function attempts(database: Database.Service["Service"], sessionID: Session.ID)
 /** Builds the local execution layer plus the restart actions against the test harness services. */
 function buildExecution(
   scope: Scope.Closeable,
-  drain: SessionRunner.Interface["drain"],
+  drain: (
+    input: Parameters<SessionRunner.Interface["drain"]>[0],
+  ) => Effect.Effect<void, SessionRunner.RunError>,
   options?: SessionRestart.Options,
 ) {
   return Effect.gen(function* () {
     const database = yield* Database.Service
     const bus = yield* Bus.Service
     const store = yield* SessionStore.Service
-    const runner = Layer.succeed(SessionRunner.Service, SessionRunner.Service.of({ drain }))
+    const runner = Layer.succeed(
+      SessionRunner.Service,
+      SessionRunner.Service.of({ drain: (input) => drain(input).pipe(Effect.as({ type: "complete" as const })) }),
+    )
     const locations = Layer.effect(
       LocationServiceMap.Service,
       LayerMap.make(

packages/core/test/session-move.test.ts

@@ -55,7 +55,7 @@ describe("Session.move", () => {
           expect(yield* session.inbox(created.id)).toMatchObject([
             {
               type: "move",
-              delivery: "queue",
+              delivery: "steer",
               payload: {
                 location: { directory: destination },
                 projectID: Project.ID.global,
@@ -69,8 +69,8 @@ describe("Session.move", () => {
           const steered = yield* session.create({
             location: Location.Ref.make({ directory: AbsolutePath.make(path.join(tmp.path, "other")) }),
           })
-          yield* session.move({ sessionID: steered.id, directory: destination, delivery: "steer" })
-          expect(yield* session.inbox(steered.id)).toMatchObject([{ type: "move", delivery: "steer" }])
+          yield* session.move({ sessionID: steered.id, directory: destination, delivery: "queue" })
+          expect(yield* session.inbox(steered.id)).toMatchObject([{ type: "move", delivery: "queue" }])
         }),
       ),
     ),

packages/core/test/session-runner-recorded.test.ts

@@ -129,7 +129,7 @@ const execution = (llmClient: Layer.Layer<typeof LLMClient.Service>) =>
     Effect.gen(function* () {
       const sessionRunner = yield* SessionRunner.Service
       const coordinator = yield* SessionRunCoordinator.make<Session.ID, SessionRunner.RunError>({
-        drain: (sessionID, force) => sessionRunner.drain({ sessionID, force }),
+        drain: (sessionID, force) => sessionRunner.drain({ sessionID, force }).pipe(Effect.asVoid),
       })
       return SessionExecution.Service.of({
         active: coordinator.active,

packages/core/test/session-runner.test.ts

@@ -387,8 +387,19 @@ const execution = Layer.effect(
   SessionExecution.Service,
   Effect.gen(function* () {
     const sessionRunner = yield* SessionRunner.Service
+    function drain(
+      sessionID: Session.ID,
+      force: boolean,
+      continuation?: SessionRunner.Continuation,
+    ): Effect.Effect<void, SessionRunner.RunError> {
+      return sessionRunner.drain({ sessionID, force, continuation }).pipe(
+        Effect.flatMap((result) =>
+          result.type === "complete" ? Effect.void : drain(sessionID, false, result.continuation),
+        ),
+      )
+    }
     const coordinator = yield* SessionRunCoordinator.make<Session.ID, SessionRunner.RunError>({
-      drain: (sessionID, force) => sessionRunner.drain({ sessionID, force }),
+      drain: (sessionID, force) => drain(sessionID, force),
     })
     return SessionExecution.Service.of({
       active: coordinator.active,
@@ -1261,6 +1272,41 @@ describe("SessionRunnerLLM", () => {
     }),
   )
 
+  it.effect("preserves a tool continuation across a steered move", () =>
+    Effect.gen(function* () {
+      const session = yield* setup
+      const bus = yield* Bus.Service
+      const { db } = yield* Database.Service
+      yield* admit(session, "Echo before moving")
+      yield* TestLLM.push(
+        TestLLM.tool("call-move", "echo", { text: "moving" }),
+        TestLLM.text("Done", "text-after-move"),
+      )
+      const tools = yield* blockTools()
+      const run = yield* session.resume(sessionID).pipe(Effect.forkChild)
+      yield* tools.started
+      yield* SessionInbox.admit(db, bus, {
+        id: SessionMessage.ID.create(),
+        sessionID,
+        item: {
+          type: "move",
+          payload: {
+            location: Location.Ref.make({ directory: AbsolutePath.make("/project") }),
+            projectID: Project.ID.global,
+          },
+          delivery: "steer",
+        },
+      })
+
+      yield* tools.release
+      yield* Fiber.join(run)
+
+      expect(requests).toHaveLength(2)
+      expect(requests.map(messageRoles).at(1)?.slice(0, 3)).toEqual(["user", "assistant", "tool"])
+      expect(yield* session.inbox(sessionID)).toEqual([])
+    }),
+  )
+
   it.effect("seeds a fork with the parent's newest instruction values", () =>
     Effect.gen(function* () {
       const session = yield* setup