From 984e62502bf948612438f253f4bf0cdc5dc39da8 Mon Sep 17 00:00:00 2001 From: Dax Raad Date: Wed, 12 Aug 2026 09:25:13 -0400 Subject: [PATCH] feat(core): defer active session moves --- packages/client/src/effect/api/api.ts | 10 ++ .../client/src/promise/generated/types.ts | 30 +++- packages/core/src/session.ts | 24 ++-- packages/core/src/session/execution.ts | 41 ++++-- .../core/src/session/execution/restart.ts | 13 +- packages/core/src/session/message-updater.ts | 1 + packages/core/src/session/pending.ts | 130 ++++++++++++++++-- packages/core/src/session/projector.ts | 15 ++ packages/core/src/session/run-coordinator.ts | 27 +++- packages/core/src/session/runner/llm.ts | 4 + packages/core/src/session/sql.ts | 6 +- packages/core/test/session-execution.test.ts | 107 +++++++++++++- packages/core/test/session-move.test.ts | 37 ++--- .../core/test/session-run-coordinator.test.ts | 34 ++++- packages/protocol/src/groups/session.ts | 2 +- packages/schema/src/session-event.ts | 12 ++ packages/schema/src/session-pending.ts | 22 ++- packages/schema/test/contract-hygiene.test.ts | 10 ++ packages/schema/test/event-manifest.test.ts | 1 + 19 files changed, 456 insertions(+), 70 deletions(-) diff --git a/packages/client/src/effect/api/api.ts b/packages/client/src/effect/api/api.ts index 43c52806a3..96de9eddcd 100644 --- a/packages/client/src/effect/api/api.ts +++ b/packages/client/src/effect/api/api.ts @@ -358,6 +358,15 @@ export type Endpoint5_31Output = readonly previous?: Model.Ref | undefined } } + | { + readonly id: Event.ID + readonly created: DateTime.Utc + readonly metadata?: { readonly [x: string]: unknown } | undefined + readonly type: "session.move.admitted" + readonly durable: { readonly aggregateID: string; readonly seq: Event.Seq; readonly version: Event.Version } + readonly location?: Location.Ref | undefined + readonly data: { readonly sessionID: Session.ID; readonly move: SessionPending.MoveData } + } | { readonly id: Event.ID readonly created: DateTime.Utc @@ -367,6 +376,7 @@ export type Endpoint5_31Output = readonly location?: Location.Ref | undefined readonly data: { readonly sessionID: Session.ID + readonly moveID?: Event.ID | undefined readonly location: Location.Ref readonly projectID?: Project.ID | undefined readonly subpath?: RelativePath | undefined diff --git a/packages/client/src/promise/generated/types.ts b/packages/client/src/promise/generated/types.ts index fffe17e413..5358c41289 100644 --- a/packages/client/src/promise/generated/types.ts +++ b/packages/client/src/promise/generated/types.ts @@ -420,6 +420,8 @@ export type SessionMessageLocationSwitched = { previous?: { location: LocationRef; projectID?: string; subpath?: string } } +export type SessionPendingMoveData = { location: LocationRef; projectID: string; subpath?: string } + export type SessionCreated = { id: string created: number @@ -468,7 +470,7 @@ export type SessionMoved = { type: "session.moved" durable: { aggregateID: string; seq: number; version: 1 } location?: LocationRef - data: { sessionID: string; location: LocationRef; projectID?: string; subpath?: string } + data: { sessionID: string; moveID?: string; location: LocationRef; projectID?: string; subpath?: string } } export type SessionRenamed = { @@ -1527,6 +1529,24 @@ export type VcsInfo = { branch: VcsBranch } export type PermissionRuleset = Array +export type SessionPendingMove = { + id: string + sessionID: string + timeCreated: number + type: "move" + data: SessionPendingMoveData +} + +export type SessionMoveAdmitted = { + id: string + created: number + metadata?: { [x: string]: any } + type: "session.move.admitted" + durable: { aggregateID: string; seq: number; version: 1 } + location?: LocationRef + data: { sessionID: string; move: SessionPendingMoveData } +} + export type SessionInfo = { id: string parentID?: string @@ -1914,7 +1934,11 @@ export type FormFields = [FormField, ...Array] export type FormFields3 = [FormField1, ...Array] -export type SessionPendingInfo = SessionPendingUser | SessionPendingSynthetic | SessionPendingCompaction +export type SessionPendingInfo = + | SessionPendingUser + | SessionPendingSynthetic + | SessionPendingCompaction + | SessionPendingMove export type SessionPendingMessage = SessionPendingUserMessage | SessionPendingSyntheticMessage @@ -1983,6 +2007,7 @@ export type SessionEventDurable = | SessionCreated | SessionAgentSelected | SessionModelSelected + | SessionMoveAdmitted | SessionMoved | SessionRenamed | SessionDeleted @@ -2046,6 +2071,7 @@ export type V2Event = | SessionCreated | SessionAgentSelected | SessionModelSelected + | SessionMoveAdmitted | SessionMoved | SessionRenamed | SessionUsageUpdated diff --git a/packages/core/src/session.ts b/packages/core/src/session.ts index bae924f9f4..bf0fcf4caf 100644 --- a/packages/core/src/session.ts +++ b/packages/core/src/session.ts @@ -185,8 +185,8 @@ export interface Interface { ) => Effect.Effect /** * Durable admitted session work not yet visible in projected history, - * ordered by admission. Includes unpromoted user and synthetic inputs and - * unhandled compaction barriers. + * ordered by admission. Includes unpromoted user and synthetic inputs, + * unhandled compaction barriers, and deferred moves. */ readonly pending: (sessionID: SessionSchema.ID) => Effect.Effect readonly cancelPending: (input: PendingInputRef) => Effect.Effect @@ -738,23 +738,21 @@ const layer = Layer.effect( const info = yield* fs.stat(directory).pipe(Effect.catch(() => Effect.succeed(undefined))) if (!info) return yield* new DestinationNotFoundError({ directory }) if (info.type !== "Directory") return yield* new DestinationNotDirectoryError({ directory }) - if (current.location.directory === directory && current.location.workspaceID === input.workspaceID) return + const pending = yield* SessionPending.move(db, input.sessionID) + if (!pending && current.location.directory === directory && current.location.workspaceID === input.workspaceID) + return const project = yield* projects.resolve(directory) yield* persistProject(project) - if ((yield* execution.active).has(input.sessionID)) { - yield* execution.interrupt(input.sessionID) - yield* execution.awaitIdle(input.sessionID) - } - yield* bus.publish( - SessionEvent.Moved, - { - sessionID: input.sessionID, + yield* SessionPending.admitMove(db, bus, { + sessionID: input.sessionID, + source: current.location, + data: { location: Location.Ref.make({ directory, workspaceID: input.workspaceID }), projectID: project.id, subpath: RelativePath.make(path.relative(project.directory, directory).replaceAll("\\", "/")), }, - { location: current.location }, - ) + }) + yield* execution.wake(input.sessionID) }), compact: Effect.fn("Session.compact")(function* (input) { yield* result.get(input.sessionID) diff --git a/packages/core/src/session/execution.ts b/packages/core/src/session/execution.ts index 8c60a894b8..ce50ad4caa 100644 --- a/packages/core/src/session/execution.ts +++ b/packages/core/src/session/execution.ts @@ -11,6 +11,8 @@ import { SessionSchema } from "./schema.js" import { SessionStore } from "./store.js" import { toSessionError } from "./to-session-error.js" import { UserInterruptedError } from "./error.js" +import { Database } from "../database/database.js" +import { SessionPending } from "./pending.js" export interface Interface { /** Snapshots active execution owned by this process. */ @@ -45,6 +47,7 @@ export const layer = Layer.effect( const store = yield* SessionStore.Service const locations = yield* LocationServiceMap.Service const bus = yield* Bus.Service + const db = (yield* Database.Service).db const reportLifecycle = (sessionID: SessionSchema.ID, effect: Effect.Effect) => effect.pipe( Effect.tapCause((cause) => @@ -54,7 +57,6 @@ export const layer = Layer.effect( Effect.annotateLogs({ sessionID }), ), ), - Effect.asVoid, ) // Write-ahead claim: starting records the durable intent that a turn is in flight, in the same // transaction as the started event. Terminals release it — except shutdown interruption, which @@ -72,7 +74,7 @@ export const layer = Layer.effect( reportLifecycle( sessionID, bus.publish(SessionEvent.Execution.Started, { sessionID }, claimOnCommit(sessionID)), - ), + ).pipe(Effect.asVoid), drain: Effect.fnUntraced(function* (sessionID: SessionSchema.ID, force) { const session = yield* store.get(sessionID) if (!session) return yield* Effect.die(new Error(`Session not found: ${sessionID}`)) @@ -91,11 +93,9 @@ export const layer = Layer.effect( sessionID, Effect.gen(function* () { const outcome = terminal(exit, reason) - if (outcome.type === "succeeded") { + if (outcome.type === "succeeded") yield* bus.publish(SessionEvent.Execution.Succeeded, { sessionID }, releaseOnCommit(sessionID)) - return - } - if (outcome.type === "interrupted") { + if (outcome.type === "interrupted") // A user cancel (or a superseding execution) releases the claim: the turn must not // resurrect at the next boot. Shutdown interruption keeps it for restart continuity. yield* bus.publish( @@ -103,16 +103,27 @@ export const layer = Layer.effect( { sessionID, reason: outcome.reason }, outcome.reason === "shutdown" ? undefined : releaseOnCommit(sessionID), ) - return - } + if (outcome.type === "failed") + yield* bus.publish( + SessionEvent.Execution.Failed, + { + sessionID, + error: outcome.error, + }, + releaseOnCommit(sessionID), + ) + + if (outcome.type === "interrupted" && outcome.reason === "shutdown") return false + const pending = yield* SessionPending.move(db, sessionID) + if (!pending) return false + const session = yield* store.get(sessionID) + if (!session) return yield* Effect.die(new Error(`Session not found: ${sessionID}`)) yield* bus.publish( - SessionEvent.Execution.Failed, - { - sessionID, - error: outcome.error, - }, - releaseOnCommit(sessionID), + SessionEvent.Moved, + { sessionID, moveID: pending.id, ...pending.data }, + { location: session.location }, ) + return yield* SessionPending.has(db, sessionID, "any") }), ), }) @@ -130,7 +141,7 @@ export const layer = Layer.effect( export const node = makeGlobalNode({ service: Service, layer, - deps: [SessionStore.node, LocationServiceMap.node, Bus.node], + deps: [SessionStore.node, LocationServiceMap.node, Bus.node, Database.node], }) /** Low-level compatibility layer for callers that only need durable Session recording. */ diff --git a/packages/core/src/session/execution/restart.ts b/packages/core/src/session/execution/restart.ts index 7413d69038..c82349e8d8 100644 --- a/packages/core/src/session/execution/restart.ts +++ b/packages/core/src/session/execution/restart.ts @@ -7,6 +7,8 @@ import { SessionEvent } from "../event.js" import { SessionExecution } from "../execution.js" import { SessionSchema } from "../schema.js" import { SessionStore } from "../store.js" +import { Database } from "../../database/database.js" +import { SessionPending } from "../pending.js" const CONTINUE_AFTER_SERVER_RESTART = "The server restarted while you were working. Continue from where you left off without repeating completed work." @@ -62,6 +64,7 @@ export const layer = (options?: Options) => const store = yield* SessionStore.Service const execution = yield* SessionExecution.Service const bus = yield* Bus.Service + const db = (yield* Database.Service).db const scope = yield* Effect.scope const maxAttempts = options?.maxAttempts ?? DEFAULT_MAX_ATTEMPTS @@ -103,6 +106,14 @@ export const layer = (options?: Options) => // them would only inject a stray continuation into a live turn. const orphaned = (yield* store.listSuspended()).filter((sessionID) => !active.has(sessionID)) yield* Effect.forEach(orphaned, resumeOne, { concurrency: "unbounded", discard: true }) + const claimed = new Set(orphaned) + yield* Effect.forEach( + (yield* SessionPending.moveSessions(db)).filter( + (sessionID) => !active.has(sessionID) && !claimed.has(sessionID), + ), + execution.wake, + { concurrency: "unbounded", discard: true }, + ) }), }) }), @@ -111,5 +122,5 @@ export const layer = (options?: Options) => export const node = makeGlobalNode({ service: Service, layer: layer(), - deps: [SessionStore.node, SessionExecution.node, Bus.node], + deps: [SessionStore.node, SessionExecution.node, Bus.node, Database.node], }) diff --git a/packages/core/src/session/message-updater.ts b/packages/core/src/session/message-updater.ts index 29617feb07..e9efa8d05a 100644 --- a/packages/core/src/session/message-updater.ts +++ b/packages/core/src/session/message-updater.ts @@ -106,6 +106,7 @@ export function update(adapter: Adapter, event: SessionEvent.DurableEvent) { ) }) }, + "session.move.admitted": () => Effect.void, "session.renamed": () => Effect.void, "session.deleted": () => Effect.void, "session.forked": () => Effect.void, diff --git a/packages/core/src/session/pending.ts b/packages/core/src/session/pending.ts index d491e13c2a..b4030bb552 100644 --- a/packages/core/src/session/pending.ts +++ b/packages/core/src/session/pending.ts @@ -7,6 +7,8 @@ import { Delivery, Info, Message, + Move, + MoveData, Synthetic, SyntheticData, User, @@ -19,10 +21,11 @@ import { SessionEvent } from "./event.js" import { SessionMessage } from "./message.js" import { SessionSchema } from "./schema.js" import { SessionMessageTable, SessionPendingTable } from "./sql.js" +import { Event } from "@opencode-ai/schema/event" type DatabaseService = Database.Interface["db"] -export { Compaction, Delivery, Info, Message, Synthetic, SyntheticData, User, UserData } +export { Compaction, Delivery, Info, Message, Move, MoveData, Synthetic, SyntheticData, User, UserData } /** * Which pending input `promote` may consume: "steer" promotes steers only (a step @@ -35,6 +38,8 @@ const decodeUser = Schema.decodeUnknownSync(UserData) const encodeUser = Schema.encodeSync(UserData) const decodeSynthetic = Schema.decodeUnknownSync(SyntheticData) const encodeSynthetic = Schema.encodeSync(SyntheticData) +const decodeMove = Schema.decodeUnknownSync(MoveData) +const encodeMove = Schema.encodeSync(MoveData) const decodeMessage = Schema.decodeUnknownSync(SessionMessage.Info) const inboxLocks = KeyedMutex.makeUnsafe() type PendingRef = { readonly id: SessionMessage.ID; readonly sessionID: SessionSchema.ID } @@ -42,21 +47,24 @@ type PendingRef = { readonly id: SessionMessage.ID; readonly sessionID: SessionS export class LifecycleConflict extends Schema.TaggedErrorClass()( "SessionPending.LifecycleConflict", { - id: SessionMessage.ID, + id: Schema.Union([SessionMessage.ID, Event.ID]), }, ) {} const fromRow = (row: typeof SessionPendingTable.$inferSelect): Info => { const base = { - id: SessionMessage.ID.make(row.id), sessionID: SessionSchema.ID.make(row.session_id), timeCreated: DateTime.makeUnsafe(row.time_created), } - if (row.type === "compaction") return Compaction.make({ ...base, type: "compaction" }) - if (!row.delivery) throw new LifecycleConflict({ id: base.id }) + if (row.type === "move") + return Move.make({ ...base, id: Event.ID.make(row.id), type: "move", data: decodeMove(row.data) }) + const id = SessionMessage.ID.make(row.id) + if (row.type === "compaction") return Compaction.make({ ...base, id, type: "compaction" }) + if (!row.delivery) throw new LifecycleConflict({ id }) if (row.type === "user") return User.make({ ...base, + id, type: "user", data: decodeUser(row.data), delivery: row.delivery, @@ -64,11 +72,12 @@ const fromRow = (row: typeof SessionPendingTable.$inferSelect): Info => { if (row.type === "synthetic") return Synthetic.make({ ...base, + id, type: "synthetic", data: decodeSynthetic(row.data), delivery: row.delivery, }) - throw new LifecycleConflict({ id: base.id }) + throw new LifecycleConflict({ id }) } export const find = Effect.fn("SessionPending.find")(function* (db: DatabaseService, id: SessionMessage.ID) { @@ -98,6 +107,44 @@ export const compaction = Effect.fn("SessionPending.compaction")(function* ( return entry.type === "compaction" ? entry : undefined }) +export const move = Effect.fn("SessionPending.move")(function* (db: DatabaseService, sessionID: SessionSchema.ID) { + const row = yield* db + .select() + .from(SessionPendingTable) + .where(and(eq(SessionPendingTable.session_id, sessionID), eq(SessionPendingTable.type, "move"))) + .orderBy(asc(SessionPendingTable.admitted_seq)) + .limit(1) + .get() + .pipe(Effect.orDie) + if (!row) return + const entry = fromRow(row) + return entry.type === "move" ? entry : undefined +}) + +export const admitMove = Effect.fn("SessionPending.admitMove")(function* ( + db: DatabaseService, + bus: Bus.Interface, + input: { readonly sessionID: SessionSchema.ID; readonly data: MoveData; readonly source: MoveData["location"] }, +) { + return yield* inboxLocks.withLock(input.sessionID)( + Effect.gen(function* () { + const pending = yield* move(db, input.sessionID) + if (pending && JSON.stringify(encodeMove(pending.data)) === JSON.stringify(encodeMove(input.data))) return pending + const event = yield* bus.publish( + SessionEvent.MoveAdmitted, + { + sessionID: input.sessionID, + move: input.data, + }, + { location: input.source }, + ) + const stored = yield* move(db, input.sessionID) + if (stored) return stored + return yield* Effect.die(new LifecycleConflict({ id: event.id })) + }), + ) +}) + const promotedFromMessage = Effect.fn("SessionPending.promotedFromMessage")(function* ( db: DatabaseService, sessionID: SessionSchema.ID, @@ -288,6 +335,35 @@ export const projectCompactionAdmitted = Effect.fn("SessionPending.projectCompac return yield* Effect.die(new LifecycleConflict({ id: input.id })) }) +export const projectMoveAdmitted = Effect.fn("SessionPending.projectMoveAdmitted")(function* ( + db: DatabaseService, + input: { + readonly admittedSeq: number + readonly id: Event.ID + readonly sessionID: SessionSchema.ID + readonly data: MoveData + readonly timeCreated: DateTime.Utc + }, +) { + yield* db + .delete(SessionPendingTable) + .where(and(eq(SessionPendingTable.session_id, input.sessionID), eq(SessionPendingTable.type, "move"))) + .run() + .pipe(Effect.orDie) + yield* db + .insert(SessionPendingTable) + .values({ + id: input.id, + session_id: input.sessionID, + type: "move", + data: input.data, + admitted_seq: input.admittedSeq, + time_created: DateTime.toEpochMillis(input.timeCreated), + }) + .run() + .pipe(Effect.orDie) +}) + /** * Consume one pending row at promotion. The row's content feeds the projected * message insert inside the same event transaction; the deleted row is what @@ -297,7 +373,8 @@ export const projectPromoted = Effect.fn("SessionPending.projectPromoted")(funct db: DatabaseService, input: PendingRef, ) { - if (yield* compaction(db, input.sessionID)) return yield* Effect.die(new LifecycleConflict({ id: input.id })) + if ((yield* compaction(db, input.sessionID)) || (yield* move(db, input.sessionID))) + return yield* Effect.die(new LifecycleConflict({ id: input.id })) const deleted = yield* db .delete(SessionPendingTable) .where(and(eq(SessionPendingTable.id, input.id), eq(SessionPendingTable.session_id, input.sessionID))) @@ -306,7 +383,8 @@ export const projectPromoted = Effect.fn("SessionPending.projectPromoted")(funct .pipe(Effect.orDie) if (!deleted) return yield* Effect.die(new LifecycleConflict({ id: input.id })) const stored = fromRow(deleted) - if (stored.type === "compaction") return yield* Effect.die(new LifecycleConflict({ id: input.id })) + if (stored.type === "compaction" || stored.type === "move") + return yield* Effect.die(new LifecycleConflict({ id: input.id })) return stored }) @@ -374,6 +452,33 @@ export const settleCompaction = Effect.fn("SessionPending.settleCompaction")(fun return undefined }) +export const settleMove = Effect.fn("SessionPending.settleMove")(function* ( + db: DatabaseService, + input: { readonly sessionID: SessionSchema.ID; readonly id: Event.ID }, +) { + yield* db + .delete(SessionPendingTable) + .where( + and( + eq(SessionPendingTable.id, input.id), + eq(SessionPendingTable.session_id, input.sessionID), + eq(SessionPendingTable.type, "move"), + ), + ) + .run() + .pipe(Effect.orDie) +}) + +export const moveSessions = Effect.fn("SessionPending.moveSessions")(function* (db: DatabaseService) { + const rows = yield* db + .select({ sessionID: SessionPendingTable.session_id }) + .from(SessionPendingTable) + .where(eq(SessionPendingTable.type, "move")) + .all() + .pipe(Effect.orDie) + return [...new Set(rows.map((row) => row.sessionID))] +}) + export const list = Effect.fn("SessionPending.list")(function* (db: DatabaseService, sessionID: SessionSchema.ID) { const rows = yield* db .select() @@ -397,7 +502,7 @@ export const has = Effect.fn("SessionPending.has")(function* ( sessionID: SessionSchema.ID, scope: Scope, ) { - if (scope !== "any" && (yield* compaction(db, sessionID))) return false + if (scope !== "any" && ((yield* compaction(db, sessionID)) || (yield* move(db, sessionID)))) return false const row = yield* db .select({ id: SessionPendingTable.id }) .from(SessionPendingTable) @@ -473,12 +578,13 @@ const publish = Effect.fn("SessionPending.publish")(function* ( sessionID: SessionSchema.ID, rows: ReadonlyArray, ) { - if (yield* compaction(db, sessionID)) return 0 + if ((yield* compaction(db, sessionID)) || (yield* move(db, sessionID))) return 0 yield* Effect.forEach( rows, (row) => { const entry = fromRow(row) - if (entry.type === "compaction") return Effect.die(new LifecycleConflict({ id: entry.id })) + if (entry.type === "compaction" || entry.type === "move") + return Effect.die(new LifecycleConflict({ id: entry.id })) return bus .publish(SessionEvent.InputPromoted, { sessionID, @@ -512,7 +618,7 @@ export const promote = Effect.fn("SessionPending.promote")(function* ( ) { return yield* inboxLocks.withLock(sessionID)( Effect.gen(function* () { - if (yield* compaction(db, sessionID)) return 0 + if ((yield* compaction(db, sessionID)) || (yield* move(db, sessionID))) return 0 const steers = yield* db .select() .from(SessionPendingTable) diff --git a/packages/core/src/session/projector.ts b/packages/core/src/session/projector.ts index dc9b25ad8b..e0994b0402 100644 --- a/packages/core/src/session/projector.ts +++ b/packages/core/src/session/projector.ts @@ -433,6 +433,8 @@ const layer = Layer.effectDiscard( .run() .pipe(Effect.orDie) yield* InstructionState.reset(db, event.data.sessionID) + if (event.data.moveID) + yield* SessionPending.settleMove(db, { sessionID: event.data.sessionID, id: event.data.moveID }) }), ) yield* bus.project(SessionEvent.Deleted, (event) => @@ -522,6 +524,19 @@ const layer = Layer.effectDiscard( .pipe(Effect.orDie) }), ) + yield* bus.project(SessionEvent.MoveAdmitted, (event) => + Effect.gen(function* () { + if (event.durable === undefined) + return yield* Effect.die(new Error("Durable Session event is missing aggregate sequence")) + yield* SessionPending.projectMoveAdmitted(db, { + admittedSeq: event.durable.seq, + id: event.id, + sessionID: event.data.sessionID, + data: event.data.move, + timeCreated: event.created, + }) + }), + ) yield* bus.project(SessionEvent.InputCancelled, (event) => SessionPending.projectCancelled(db, { id: event.data.inputID, diff --git a/packages/core/src/session/run-coordinator.ts b/packages/core/src/session/run-coordinator.ts index d29ba0d9dd..df2f1824b1 100644 --- a/packages/core/src/session/run-coordinator.ts +++ b/packages/core/src/session/run-coordinator.ts @@ -1,6 +1,6 @@ export * as SessionRunCoordinator from "./run-coordinator.js" -import { Deferred, Effect, Exit, Fiber, FiberSet, Scope } from "effect" +import { Cause, Deferred, Effect, Exit, Fiber, FiberSet, Scope } from "effect" /** Serializes execution for each key while allowing different keys to run concurrently. */ export interface Coordinator { @@ -50,11 +50,17 @@ export const make = (options: { * Runs in the execution fiber for every exit, including interruption, after the final * drain and before the execution settles (waiters resolve after it completes). */ - readonly settled?: (key: Key, exit: Exit.Exit, reason?: Reason) => Effect.Effect + readonly settled?: (key: Key, exit: Exit.Exit, reason?: Reason) => Effect.Effect }): Effect.Effect, never, Scope.Scope> => Effect.gen(function* () { const executions = new Map>() const fork = yield* FiberSet.makeRuntime() + const closing = { value: false } + yield* Effect.addFinalizer(() => + Effect.sync(() => { + closing.value = true + }), + ) const loop = (key: Key, execution: Execution, force: boolean): Effect.Effect => Effect.suspend(() => options.drain(key, force)).pipe( @@ -85,7 +91,22 @@ export const make = (options: { Effect.onExit((exit) => Effect.sync(() => { execution.owner = undefined - }).pipe(Effect.andThen(options.settled?.(key, exit, execution.interruptionReason) ?? Effect.void)), + if (closing.value && Exit.isFailure(exit) && Cause.hasInterrupts(exit.cause)) { + execution.stopping = true + execution.pendingWake = false + } + }).pipe( + Effect.andThen(options.settled?.(key, exit, execution.interruptionReason) ?? Effect.void), + Effect.map(Boolean), + Effect.tap((restart) => + restart && !execution.stopping + ? Effect.sync(() => { + execution.pendingWake = true + }) + : Effect.void, + ), + Effect.asVoid, + ), ), Effect.onExit((exit) => Effect.sync(() => settle(key, execution, exit))), Effect.exit, diff --git a/packages/core/src/session/runner/llm.ts b/packages/core/src/session/runner/llm.ts index 6b5bfebead..989b82e1f3 100644 --- a/packages/core/src/session/runner/llm.ts +++ b/packages/core/src/session/runner/llm.ts @@ -143,6 +143,7 @@ const layer = Layer.effect( let promotable: SessionPending.Promotable = "input" let step = 1 while (true) { + if (yield* SessionPending.move(db, sessionID)) return const result = yield* runStep(sessionID, promotable, step) if (step === 1) yield* startTitle(sessionID) yield* runPendingCompaction(sessionID) @@ -236,6 +237,8 @@ const layer = Layer.effect( // a blocked first step leaves pending inputs untouched. yield* InstructionState.prepare(db, bus, selected.instructions, selected.session.id) const promoted = promotable ? yield* SessionPending.promote(db, bus, selected.session.id, promotable) : 0 + if (promotable && promoted === 0 && (yield* SessionPending.move(db, sessionID))) + return CallOutcome.Completed({ needsContinuation: false, step }) // Promoted input opens a fresh step allowance. const currentStep = promoted > 0 ? 1 : step const loaded = yield* context.load(selected) @@ -482,6 +485,7 @@ const layer = Layer.effect( const runPendingCompaction = Effect.fn("SessionRunner.runPendingCompaction")(function* ( sessionID: SessionSchema.ID, ) { + if (yield* SessionPending.move(db, sessionID)) return const pending = yield* SessionPending.compaction(db, sessionID) if (!pending) return const session = yield* getSession(sessionID) diff --git a/packages/core/src/session/sql.ts b/packages/core/src/session/sql.ts index 8db29460e6..246dbf5256 100644 --- a/packages/core/src/session/sql.ts +++ b/packages/core/src/session/sql.ts @@ -96,13 +96,15 @@ export const SessionMessageTable = sqliteTable( export const SessionPendingTable = sqliteTable( "session_pending", { - id: text().$type().primaryKey(), + id: text().$type().primaryKey(), session_id: text() .$type() .notNull() .references(() => SessionTable.id, { onDelete: "cascade" }), type: text().$type().notNull(), - data: text({ mode: "json" }).$type>().notNull(), + data: text({ mode: "json" }) + .$type>() + .notNull(), delivery: text().$type(), admitted_seq: integer().notNull(), time_created: integer() diff --git a/packages/core/test/session-execution.test.ts b/packages/core/test/session-execution.test.ts index e6a6d498d5..7818095779 100644 --- a/packages/core/test/session-execution.test.ts +++ b/packages/core/test/session-execution.test.ts @@ -8,7 +8,7 @@ import { LocationServiceMap } from "@opencode-ai/core/location-service-map" import type { LocationServices } from "@opencode-ai/core/location-services" import { Project } from "@opencode-ai/core/project" import { ProjectTable } from "@opencode-ai/core/project/sql" -import { AbsolutePath } from "@opencode-ai/core/schema" +import { AbsolutePath, RelativePath } from "@opencode-ai/core/schema" import { Session } from "@opencode-ai/core/session" import { SessionExecution } from "@opencode-ai/core/session/execution" import { SessionRestart } from "@opencode-ai/core/session/execution/restart" @@ -17,11 +17,16 @@ import { SessionEvent } from "@opencode-ai/core/session/event" import { SessionRunner } from "@opencode-ai/core/session/runner" import { SessionTable } from "@opencode-ai/core/session/sql" import { SessionStore } from "@opencode-ai/core/session/store" +import { SessionProjector } from "@opencode-ai/core/session/projector" +import { SessionPending } from "@opencode-ai/core/session/pending" +import { Location } from "@opencode-ai/core/location" import { Context, Deferred, Effect, Exit, Fiber, Layer, LayerMap, Scope } from "effect" import { eq } from "drizzle-orm" import { testEffect } from "./lib/effect" -const it = testEffect(AppNodeBuilder.build(LayerNode.group([Database.node, Bus.node, SessionStore.node]))) +const it = testEffect( + AppNodeBuilder.build(LayerNode.group([Database.node, Bus.node, SessionProjector.node, SessionStore.node])), +) describe("SessionExecution lifecycle", () => { test("classifies success and typed failure terminals", () => { @@ -133,6 +138,104 @@ describe("SessionExecution lifecycle", () => { }), ) + it.effect("applies a deferred move only after the active execution settles", () => + Effect.gen(function* () { + const database = yield* Database.Service + const bus = yield* Bus.Service + const store = yield* SessionStore.Service + const sessionID = Session.ID.make("ses_deferred_move") + yield* seedSessions(database, [sessionID]) + + const draining = yield* Deferred.make() + const release = yield* Deferred.make() + const scope = yield* Scope.make() + yield* Effect.addFinalizer(() => Scope.close(scope, Exit.void)) + const context = yield* buildExecution(scope, () => + Deferred.succeed(draining, undefined).pipe(Effect.andThen(Deferred.await(release))), + ) + const execution = Context.get(context, SessionExecution.Service) + yield* execution.resume(sessionID).pipe(Effect.forkIn(scope)) + yield* Deferred.await(draining) + + yield* bus.publish(SessionEvent.MoveAdmitted, { + sessionID, + move: { + location: Location.Ref.make({ directory: AbsolutePath.make("/destination") }), + projectID: Project.ID.global, + subpath: RelativePath.make(""), + }, + }) + expect((yield* store.get(sessionID))?.location.directory).toBe(AbsolutePath.make("/project")) + expect((yield* SessionPending.move(database.db, sessionID))?.data.location.directory).toBe( + AbsolutePath.make("/destination"), + ) + + yield* Deferred.succeed(release, undefined) + yield* execution.awaitIdle(sessionID) + + expect((yield* store.get(sessionID))?.location.directory).toBe(AbsolutePath.make("/destination")) + expect(yield* SessionPending.move(database.db, sessionID)).toBeUndefined() + }), + ) + + it.effect("settling one move preserves a newer admitted destination", () => + Effect.gen(function* () { + const database = yield* Database.Service + const bus = yield* Bus.Service + const store = yield* SessionStore.Service + const sessionID = Session.ID.make("ses_move_replacement") + yield* seedSessions(database, [sessionID]) + const first = { + location: Location.Ref.make({ directory: AbsolutePath.make("/first") }), + projectID: Project.ID.global, + subpath: RelativePath.make("first"), + } + const second = { + location: Location.Ref.make({ directory: AbsolutePath.make("/second") }), + projectID: Project.ID.global, + subpath: RelativePath.make("second"), + } + + const admittedFirst = yield* bus.publish(SessionEvent.MoveAdmitted, { sessionID, move: first }) + const admittedSecond = yield* bus.publish(SessionEvent.MoveAdmitted, { sessionID, move: second }) + yield* bus.publish(SessionEvent.Moved, { sessionID, moveID: admittedFirst.id, ...first }) + + expect((yield* store.get(sessionID))?.location.directory).toBe(first.location.directory) + expect((yield* SessionPending.move(database.db, sessionID))?.id).toBe(admittedSecond.id) + + yield* bus.publish(SessionEvent.Moved, { sessionID, moveID: admittedSecond.id, ...second }) + expect((yield* store.get(sessionID))?.location.directory).toBe(second.location.directory) + expect(yield* SessionPending.move(database.db, sessionID)).toBeUndefined() + }), + ) + + it.effect("recovers an unclaimed deferred move on startup", () => + Effect.gen(function* () { + const database = yield* Database.Service + const bus = yield* Bus.Service + const store = yield* SessionStore.Service + const sessionID = Session.ID.make("ses_move_recovery") + yield* seedSessions(database, [sessionID]) + yield* bus.publish(SessionEvent.MoveAdmitted, { + sessionID, + move: { + location: Location.Ref.make({ directory: AbsolutePath.make("/recovered") }), + projectID: Project.ID.global, + }, + }) + + const scope = yield* Scope.make() + yield* Effect.addFinalizer(() => Scope.close(scope, Exit.void)) + const context = yield* buildExecution(scope, () => Effect.void) + const execution = Context.get(context, SessionExecution.Service) + yield* Context.get(context, SessionRestart.Service).resumeSuspendedSessions + yield* execution.awaitIdle(sessionID) + + expect((yield* store.get(sessionID))?.location.directory).toBe(AbsolutePath.make("/recovered")) + expect(yield* SessionPending.move(database.db, sessionID)).toBeUndefined() + }), + ) + it.effect("starts every claimed execution without waiting for earlier drains to finish", () => Effect.gen(function* () { const database = yield* Database.Service diff --git a/packages/core/test/session-move.test.ts b/packages/core/test/session-move.test.ts index 9b2798697c..23fe49aa76 100644 --- a/packages/core/test/session-move.test.ts +++ b/packages/core/test/session-move.test.ts @@ -1,5 +1,6 @@ import { describe, expect } from "bun:test" import path from "path" +import fs from "fs/promises" import { Effect, Layer } from "effect" import { Bus } from "@opencode-ai/core/bus" import { Database } from "@opencode-ai/core/database/database" @@ -34,7 +35,7 @@ const it = testEffect( ) describe("Session.move", () => { - it.effect("moves a session whose source directory no longer exists", () => + it.effect("durably admits a move when the source directory no longer exists", () => Effect.acquireRelease( Effect.promise(() => tmpdir()), (tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()), @@ -49,24 +50,26 @@ describe("Session.move", () => { yield* session.move({ sessionID: created.id, directory: destination }) - expect((yield* session.get(created.id)).location.directory).toBe(destination) - const messages = yield* session.messages({ sessionID: created.id, order: "asc" }) - expect(messages).toEqual([ - expect.objectContaining({ - type: "location-switched", - location: { directory: destination }, - projectID: Project.ID.global, - previous: { - location: { directory: path.join(tmp.path, "deleted") }, - projectID: Project.ID.global, - subpath: "", - }, - subpath: "", - }), + expect((yield* session.get(created.id)).location.directory).toBe( + AbsolutePath.make(path.join(tmp.path, "deleted")), + ) + expect(yield* session.pending(created.id)).toMatchObject([ + { + type: "move", + data: { location: { directory: destination }, projectID: Project.ID.global }, + }, ]) - yield* session.move({ sessionID: created.id, directory: destination }) - expect(yield* session.messages({ sessionID: created.id, order: "asc" })).toEqual(messages) + const replacement = AbsolutePath.make(path.join(tmp.path, "replacement")) + yield* Effect.promise(() => fs.mkdir(replacement)) + yield* session.move({ sessionID: created.id, directory: replacement }) + + expect(yield* session.pending(created.id)).toMatchObject([ + { + type: "move", + data: { location: { directory: replacement }, projectID: Project.ID.global }, + }, + ]) }), ), ), diff --git a/packages/core/test/session-run-coordinator.test.ts b/packages/core/test/session-run-coordinator.test.ts index c566c3f33c..9238899ac7 100644 --- a/packages/core/test/session-run-coordinator.test.ts +++ b/packages/core/test/session-run-coordinator.test.ts @@ -143,19 +143,26 @@ describe("SessionRunCoordinator", () => { it.effect("cleans active executions when its scope closes", () => Effect.gen(function* () { const started = yield* Deferred.make() + let runs = 0 const coordinator = yield* Effect.scoped( Effect.gen(function* () { const coordinator = yield* SessionRunCoordinator.make({ - drain: () => Deferred.succeed(started, undefined).pipe(Effect.andThen(Effect.never)), + drain: () => + Effect.sync(() => runs++).pipe( + Effect.andThen(Deferred.succeed(started, undefined)), + Effect.andThen(Effect.never), + ), }) yield* coordinator.wake("session") yield* Deferred.await(started) + yield* coordinator.wake("session") expect(Array.from(yield* coordinator.active)).toEqual(["session"]) return coordinator }), ) expect(Array.from(yield* coordinator.active)).toEqual([]) + expect(runs).toBe(1) }), ) @@ -517,6 +524,31 @@ describe("SessionRunCoordinator", () => { ), ) + it.effect("starts one successor when settlement requests it", () => + Effect.scoped( + Effect.gen(function* () { + const successor = yield* Deferred.make() + let drains = 0 + let settlements = 0 + const coordinator = yield* SessionRunCoordinator.make({ + drain: () => + Effect.sync(() => { + drains++ + if (drains === 2) Deferred.doneUnsafe(successor, Effect.void) + }), + settled: () => Effect.sync(() => ++settlements === 1), + }) + + yield* coordinator.wake("session") + yield* Deferred.await(successor) + yield* coordinator.awaitIdle("session") + + expect(drains).toBe(2) + expect(settlements).toBe(2) + }), + ), + ) + it.effect("trampolines synchronous self-waking execution", () => Effect.scoped( Effect.gen(function* () { diff --git a/packages/protocol/src/groups/session.ts b/packages/protocol/src/groups/session.ts index d6b8a328c2..4fde363253 100644 --- a/packages/protocol/src/groups/session.ts +++ b/packages/protocol/src/groups/session.ts @@ -308,7 +308,7 @@ export const makeSessionGroup = (sessionLo OpenApi.annotations({ identifier: "v2.session.move", summary: "Move session", - description: "Move a session to another project directory, optionally transferring local changes.", + description: "Move a session to another project directory after any active execution settles.", }), ), ) diff --git a/packages/schema/src/session-event.ts b/packages/schema/src/session-event.ts index 813f1cf749..fa51744e20 100644 --- a/packages/schema/src/session-event.ts +++ b/packages/schema/src/session-event.ts @@ -90,6 +90,7 @@ export const Moved = Event.durable({ ...options, schema: { ...Base, + moveID: Event.ID.pipe(optional), location: Location.Ref, projectID: Project.ID.pipe(optional), subpath: RelativePath.pipe(optional), @@ -97,6 +98,16 @@ export const Moved = Event.durable({ }) export type Moved = typeof Moved.Type +export const MoveAdmitted = Event.durable({ + type: "session.move.admitted", + ...options, + schema: { + ...Base, + move: SessionPending.MoveData, + }, +}) +export type MoveAdmitted = typeof MoveAdmitted.Type + export const Renamed = Event.durable({ type: "session.renamed", ...options, @@ -597,6 +608,7 @@ export const Definitions = Event.inventory( Created, AgentSelected, ModelSelected, + MoveAdmitted, Moved, Renamed, UsageUpdated, diff --git a/packages/schema/src/session-pending.ts b/packages/schema/src/session-pending.ts index f687c4ecf2..491ef942bc 100644 --- a/packages/schema/src/session-pending.ts +++ b/packages/schema/src/session-pending.ts @@ -7,6 +7,10 @@ import { DateTimeUtcFromMillis } from "./schema.js" import { SessionDelivery } from "./session-delivery.js" import { SessionID } from "./session-id.js" import { SessionMessage } from "./session-message.js" +import { Event } from "./event.js" +import { Location } from "./location.js" +import { Project } from "./project.js" +import { RelativePath } from "./schema.js" export const Delivery = SessionDelivery.Delivery export type Delivery = SessionDelivery.Delivery @@ -68,7 +72,23 @@ export const Compaction = Schema.Struct({ type: Schema.tag("compaction"), }).annotate({ identifier: "SessionPending.Compaction" }) -export const Info = Schema.Union([User, Synthetic, Compaction]).pipe( +export interface MoveData extends Schema.Schema.Type {} +export const MoveData = Schema.Struct({ + location: Location.Ref, + projectID: Project.ID, + subpath: RelativePath.pipe(optional), +}).annotate({ identifier: "SessionPending.MoveData" }) + +export interface Move extends Schema.Schema.Type {} +export const Move = Schema.Struct({ + id: Event.ID, + sessionID: SessionID, + timeCreated: DateTimeUtcFromMillis, + type: Schema.tag("move"), + data: MoveData, +}).annotate({ identifier: "SessionPending.Move" }) + +export const Info = Schema.Union([User, Synthetic, Compaction, Move]).pipe( Schema.toTaggedUnion("type"), Schema.annotate({ identifier: "SessionPending.Info" }), ) diff --git a/packages/schema/test/contract-hygiene.test.ts b/packages/schema/test/contract-hygiene.test.ts index de6993beba..3960cfb0d5 100644 --- a/packages/schema/test/contract-hygiene.test.ts +++ b/packages/schema/test/contract-hygiene.test.ts @@ -90,6 +90,16 @@ describe("contract hygiene", () => { }) }) + test("pending moves omit absent placement details", () => { + expect( + Schema.encodeSync(SessionPending.MoveData)({ + location: { directory: AbsolutePath.make("/project"), workspaceID: undefined }, + projectID: Project.ID.global, + subpath: undefined, + }), + ).toEqual({ location: { directory: "/project" }, projectID: "global" }) + }) + test("forms require at least one field", () => { expect(() => Schema.decodeUnknownSync(Form.Info)({ diff --git a/packages/schema/test/event-manifest.test.ts b/packages/schema/test/event-manifest.test.ts index 7896bdc0d6..e20e9afc22 100644 --- a/packages/schema/test/event-manifest.test.ts +++ b/packages/schema/test/event-manifest.test.ts @@ -78,6 +78,7 @@ describe("public event manifest", () => { "session.deleted.2", "session.agent.selected.1", "session.model.selected.1", + "session.move.admitted.1", "session.moved.1", "session.renamed.1", "session.usage.recorded.1",