diff --git a/packages/core/src/session.ts b/packages/core/src/session.ts index ebe91c5ccba..48c7625f9fe 100644 --- a/packages/core/src/session.ts +++ b/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, { diff --git a/packages/core/src/session/execution.ts b/packages/core/src/session/execution.ts index 8a201d1853e..6ea2fc0779c 100644 --- a/packages/core/src/session/execution.ts +++ b/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({ - 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 { + 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({ + 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( diff --git a/packages/core/src/session/runner/index.ts b/packages/core/src/session/runner/index.ts index 6922c287786..e744f9bb75c 100644 --- a/packages/core/src/session/runner/index.ts +++ b/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 + readonly continuation?: Continuation + }) => Effect.Effect } export class Service extends Context.Service()("@opencode/SessionRunner") {} diff --git a/packages/core/src/session/runner/llm.ts b/packages/core/src/session/runner/llm.ts index 1ab5bc84ff5..cc971dcd565 100644 --- a/packages/core/src/session/runner/llm.ts +++ b/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 }], diff --git a/packages/core/test/session-execution.test.ts b/packages/core/test/session-execution.test.ts index 090bbb54322..d851d081efc 100644 --- a/packages/core/test/session-execution.test.ts +++ b/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[0], + ) => Effect.Effect, 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( diff --git a/packages/core/test/session-move.test.ts b/packages/core/test/session-move.test.ts index 03b0238a33f..4dc77547387 100644 --- a/packages/core/test/session-move.test.ts +++ b/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" }]) }), ), ), diff --git a/packages/core/test/session-runner-recorded.test.ts b/packages/core/test/session-runner-recorded.test.ts index 6e68fc7d9c6..74613a505fb 100644 --- a/packages/core/test/session-runner-recorded.test.ts +++ b/packages/core/test/session-runner-recorded.test.ts @@ -129,7 +129,7 @@ const execution = (llmClient: Layer.Layer) => Effect.gen(function* () { const sessionRunner = yield* SessionRunner.Service const coordinator = yield* SessionRunCoordinator.make({ - drain: (sessionID, force) => sessionRunner.drain({ sessionID, force }), + drain: (sessionID, force) => sessionRunner.drain({ sessionID, force }).pipe(Effect.asVoid), }) return SessionExecution.Service.of({ active: coordinator.active, diff --git a/packages/core/test/session-runner.test.ts b/packages/core/test/session-runner.test.ts index 033056eb640..d0247df45f9 100644 --- a/packages/core/test/session-runner.test.ts +++ b/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 { + 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({ - 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