diff --git a/packages/client/src/effect/api/api.ts b/packages/client/src/effect/api/api.ts index 230c8bc2054a..a3ca89c28a5c 100644 --- a/packages/client/src/effect/api/api.ts +++ b/packages/client/src/effect/api/api.ts @@ -605,7 +605,10 @@ export type SessionLogOutput = readonly type: "session.execution.interrupted" 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 reason: "user" | "shutdown" | "superseded" } + readonly data: { + readonly sessionID: Session.ID + readonly reason: "user" | "shutdown" | "superseded" | "inactivity" + } } | { readonly id: Event.ID diff --git a/packages/client/src/promise/generated/types.ts b/packages/client/src/promise/generated/types.ts index b60c805ddf94..f343fc706ab7 100644 --- a/packages/client/src/promise/generated/types.ts +++ b/packages/client/src/promise/generated/types.ts @@ -710,7 +710,7 @@ export type SessionExecutionInterrupted = { type: "session.execution.interrupted" durable: { aggregateID: string; seq: number; version: 1 } location?: LocationRef - data: { sessionID: string; reason: "user" | "shutdown" | "superseded" } + data: { sessionID: string; reason: "user" | "shutdown" | "superseded" | "inactivity" } } export type SessionInstructionsUpdated = { diff --git a/packages/core/src/location-activity.ts b/packages/core/src/location-activity.ts index 713b1a3392cd..81e9bddea9fa 100644 --- a/packages/core/src/location-activity.ts +++ b/packages/core/src/location-activity.ts @@ -5,6 +5,8 @@ import { Bus } from "./bus.js" import { Location } from "./location.js" import { LocationServiceMap } from "./location-service-map.js" import { SessionEvent } from "./session/event.js" +import { SessionExecution } from "./session/execution.js" +import { SessionStore } from "./session/store.js" import { makeGlobalNode } from "@opencode-ai/util/effect/app-node" const isSessionEvent = Schema.is(SessionEvent.Durable) @@ -18,9 +20,11 @@ export function layer(options: { readonly timeToLive?: Duration.Input; readonly const clock = yield* Clock.Clock const bus = yield* Bus.Service const locations = yield* LocationServiceMap.Service + const execution = yield* SessionExecution.Service + const sessions = yield* SessionStore.Service const timeToLive = Duration.toMillis(options.timeToLive ?? "60 minutes") const entries = new Map() - const key = (ref: Location.Ref) => `${ref.directory}\0${ref.workspaceID ?? ""}` + const key = (ref: Location.Ref) => `${LocationServiceMap.canonical(ref).directory}\0${ref.workspaceID ?? ""}` const touch = (ref: Location.Ref) => Effect.sync(() => { entries.set(key(ref), { ref, expiresAt: clock.currentTimeMillisUnsafe() + timeToLive }) @@ -39,26 +43,44 @@ export function layer(options: { readonly timeToLive?: Duration.Input; readonly yield* Effect.sleep(options.sweepInterval ?? "1 minute") const refs = Array.from(yield* RcMap.keys(locations.rcMap)) const cached = new Set(refs.map(key)) - yield* Effect.forEach( - refs, - (ref) => (entries.has(key(ref)) ? Effect.void : touch(ref)), - { discard: true }, - ) + yield* Effect.forEach(refs, (ref) => (entries.has(key(ref)) ? Effect.void : touch(ref)), { discard: true }) for (const id of entries.keys()) { if (!cached.has(id)) entries.delete(id) } const now = clock.currentTimeMillisUnsafe() const expired = Array.from(entries.values()).filter((entry) => entry.expiresAt <= now) + if (expired.length === 0) return + const active = yield* Effect.forEach(yield* execution.active, (sessionID) => sessions.get(sessionID)) yield* Effect.forEach( expired, - (entry) => { - entries.delete(key(entry.ref)) - return Effect.logInfo("location services evicted", { - directory: entry.ref.directory, - workspaceID: entry.ref.workspaceID, - }).pipe(Effect.andThen(locations.invalidate(entry.ref))) - }, - { discard: true }, + (entry) => + Effect.gen(function* () { + const owners = active.flatMap((session) => + session && key(session.location) === key(entry.ref) ? [session] : [], + ) + // Invalidation only detaches the cache entry; borrowers retain the old + // graph. Stop its executions and settle tool cleanup before detaching it. + yield* Effect.forEach( + owners, + (session) => execution.interrupt(session.id, { reason: "inactivity", awaitSettlement: true }), + { + discard: true, + concurrency: "unbounded", + }, + ) + const remaining = yield* Effect.forEach(yield* execution.active, (sessionID) => sessions.get(sessionID)) + // New work admitted during cleanup may now own the cached graph. + if (remaining.some((session) => session && key(session.location) === key(entry.ref))) { + yield* touch(entry.ref) + return + } + entries.delete(key(entry.ref)) + yield* Effect.logInfo("location services evicted", { + directory: entry.ref.directory, + workspaceID: entry.ref.workspaceID, + }).pipe(Effect.andThen(locations.invalidate(entry.ref))) + }), + { discard: true, concurrency: "unbounded" }, ) }).pipe(Effect.forever, Effect.forkScoped) @@ -70,5 +92,5 @@ export function layer(options: { readonly timeToLive?: Duration.Input; readonly export const node = makeGlobalNode({ service: Service, layer: layer(), - deps: [Bus.node, LocationServiceMap.node], + deps: [Bus.node, LocationServiceMap.node, SessionExecution.node, SessionStore.node], }) diff --git a/packages/core/src/session/execution.ts b/packages/core/src/session/execution.ts index 9ff89f5113f1..3e15f306ba48 100644 --- a/packages/core/src/session/execution.ts +++ b/packages/core/src/session/execution.ts @@ -28,9 +28,17 @@ export interface Interface { * Interrupt active work owned by this process. Idle interruption is a no-op. Resolves once * the interruption is accepted; cleanup settles asynchronously in the execution fiber. * Returns whether an active execution was interrupted. Compose with `awaitIdle` when - * settlement matters. + * settlement matters. `awaitSettlement` waits only for the interrupted execution, + * rather than fresh work admitted during its cleanup. */ - readonly interrupt: (sessionID: SessionSchema.ID, options?: { readonly continue?: boolean }) => Effect.Effect + readonly interrupt: ( + sessionID: SessionSchema.ID, + options?: { + readonly continue?: boolean + readonly reason?: "user" | "inactivity" + readonly awaitSettlement?: boolean + }, + ) => Effect.Effect /** Resolves once this process owns no active execution for the Session. Returns immediately when idle and never starts work. */ readonly awaitIdle: (sessionID: SessionSchema.ID) => Effect.Effect } @@ -38,7 +46,7 @@ export interface Interface { /** Routes execution from a Session ID to its selected instance's runner. */ export class Service extends Context.Service()("@opencode/SessionExecution") {} -type InterruptReason = "user" | "shutdown" +type InterruptReason = "user" | "shutdown" | "inactivity" export function terminal(exit: Exit.Exit, reason?: InterruptReason) { if (Exit.isSuccess(exit)) return { type: "succeeded" as const } @@ -120,9 +128,8 @@ export const layer = Layer.effect( return } if (outcome.type === "interrupted") { - // A user cancel releases the claim: the turn must not resurrect at the next - // boot. Shutdown interruption keeps it for restart continuity. - if (outcome.reason === "user") yield* jobs.cancel(sessionID) + // Deliberate stops release the claim; shutdown keeps it for restart continuity. + if (outcome.reason !== "shutdown") yield* jobs.cancel(sessionID) yield* bus.publish( SessionEvent.Execution.Interrupted, { sessionID, reason: outcome.reason }, @@ -147,7 +154,7 @@ export const layer = Layer.effect( isActive: coordinator.isActive, interrupt: (sessionID, options) => Effect.gen(function* () { - const interrupted = yield* coordinator.interrupt(sessionID, "user") + const interrupted = yield* coordinator.interrupt(sessionID, options?.reason ?? "user", options) if (!options?.continue) return interrupted // Resume steering input and between-turn control work from the interrupted // intent. Queued next-turn prompts stay parked: a steer-scoped drain never diff --git a/packages/core/src/session/run-coordinator.ts b/packages/core/src/session/run-coordinator.ts index 7130f5ed5e2c..9c4ba4b09a97 100644 --- a/packages/core/src/session/run-coordinator.ts +++ b/packages/core/src/session/run-coordinator.ts @@ -17,9 +17,14 @@ export interface Coordinator { * Stops the active execution and clears its doorbell. No-op when idle. Resolves once the * interruption is accepted, not when cleanup settles: the execution fiber finishes its * finalizers and settled hook on its own time. Returns whether an active execution was - * interrupted. Compose with `awaitIdle` for settlement. + * interrupted. `awaitSettlement` waits for this execution's cleanup and settled hook, + * without following fresh work admitted during cleanup. `awaitIdle` follows successors too. */ - readonly interrupt: (key: Key, reason?: Reason) => Effect.Effect + readonly interrupt: ( + key: Key, + reason?: Reason, + options?: { readonly awaitSettlement?: boolean }, + ) => 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 } @@ -170,5 +175,22 @@ export const make = (options: { return Deferred.await(execution.done).pipe(Effect.ignoreCause, Effect.andThen(awaitIdle(key))) }) - return { active: Effect.sync(() => new Set(executions.keys())), isActive, run, wake, interrupt, awaitIdle } + return { + active: Effect.sync(() => new Set(executions.keys())), + isActive, + run, + wake, + interrupt: (key, reason, options) => + Effect.suspend(() => { + const execution = executions.get(key) + return interrupt(key, reason).pipe( + Effect.tap(() => + options?.awaitSettlement && execution + ? Deferred.await(execution.done).pipe(Effect.ignoreCause) + : Effect.void, + ), + ) + }), + awaitIdle, + } }) diff --git a/packages/core/test/location-activity.test.ts b/packages/core/test/location-activity.test.ts new file mode 100644 index 000000000000..689d0392220f --- /dev/null +++ b/packages/core/test/location-activity.test.ts @@ -0,0 +1,221 @@ +import { describe, expect } from "bun:test" +import { Context, Deferred, Duration, Effect, Fiber, Layer, LayerMap, RcMap, Schema } from "effect" +import { TestClock } from "effect/testing" +import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" +import { LayerNode } from "@opencode-ai/util/effect/layer-node" +import { makeGlobalNode } from "@opencode-ai/util/effect/app-node" +import { Bus } from "@opencode-ai/core/bus" +import { Database } from "@opencode-ai/core/database/database" +import { Form } from "@opencode-ai/core/form" +import { Location } from "@opencode-ai/core/location" +import { LocationActivity } from "@opencode-ai/core/location-activity" +import { LocationServiceMap, 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 { Session } from "@opencode-ai/core/session" +import { SessionExecution } from "@opencode-ai/core/session/execution" +import { SessionEvent } from "@opencode-ai/core/session/event" +import { SessionRunner } from "@opencode-ai/core/session/runner/index" +import { SessionTable } from "@opencode-ai/core/session/sql" +import { SessionStore } from "@opencode-ai/core/session/store" +import { Workspace } from "@opencode-ai/core/workspace" +import { testEffect } from "./lib/effect" + +// Keep real execution ownership, location caching, forms, and eviction. The fixture +// runner waits on a form instead of making a model request before asking a question. +const locations = Layer.effect( + LocationServiceMap.Service, + Effect.gen(function* () { + const bus = yield* Bus.Service + const map = yield* LayerMap.make( + (ref: Location.Ref) => + // The fixture only exercises these three Location services. + // oxlint-disable-next-line typescript-eslint/no-unsafe-type-assertion + Layer.merge( + Layer.succeed( + Location.Service, + Location.Service.of({ + directory: ref.directory, + workspaceID: ref.workspaceID, + project: { id: Project.ID.global, directory: ref.directory, canonical: ref.directory }, + }), + ), + Layer.effect( + SessionRunner.Service, + Effect.gen(function* () { + const forms = yield* Form.Service + return SessionRunner.Service.of({ + drain: ({ sessionID }) => + forms + .ask({ + sessionID, + title: "Questions", + fields: [{ key: "runtime", type: "string" }], + }) + .pipe( + Effect.orDie, + Effect.as(SessionRunner.DrainResult.Complete()), + Effect.onInterrupt(() => Effect.sleep("5 minutes")), + ), + }) + }), + ), + ).pipe( + Layer.provideMerge(Form.layer), + Layer.provide(Layer.succeed(Bus.Service, bus)), + Layer.fresh, + ) as unknown as Layer.Layer, + { idleTimeToLive: Duration.infinity }, + ) + return { + ...map, + get: (ref: Location.Ref) => map.get(LocationServiceMap.canonical(ref)), + contextEffect: (ref: Location.Ref) => map.contextEffect(LocationServiceMap.canonical(ref)), + contextEffectOption: (ref: Location.Ref) => map.contextEffectOption(LocationServiceMap.canonical(ref)), + invalidate: (ref: Location.Ref) => map.invalidate(LocationServiceMap.canonical(ref)), + } + }), +) + +const it = testEffect( + AppNodeBuilder.build( + LayerNode.group([ + Database.node, + Bus.node, + SessionStore.node, + LocationServiceMap.node, + SessionExecution.node, + LocationActivity.node, + ]), + [ + LocationServiceMap.node.replace( + makeGlobalNode({ + service: LocationServiceMap.Service, + layer: locations, + deps: [Bus.node], + }), + ), + ], + ), +) + +describe("LocationActivity eviction", () => { + for (const [count, admission] of [ + [1, "none"], + [2, "none"], + [1, "other"], + [1, "same"], + ] as const) { + const newWork = admission !== "none" + it.effect( + `interrupts ${count} waiting executions before eviction (${admission} session admitted during cleanup)`, + () => + Effect.gen(function* () { + const db = (yield* Database.Service).db + const bus = yield* Bus.Service + const map = yield* LocationServiceMap.Service + const execution = yield* SessionExecution.Service + const store = yield* SessionStore.Service + const sessionIDs = Array.from({ length: count }, (_, index) => + Session.ID.make(`ses_waiting_question_${index}`), + ) + const newcomer = admission === "same" ? sessionIDs[0] : Session.ID.make("ses_new_question") + const ref = LocationServiceMap.canonical({ directory: AbsolutePath.make("/project") }) + const idle = Location.Ref.make({ directory: ref.directory, workspaceID: Workspace.ID.make("wrk_idle") }) + yield* db + .insert(ProjectTable) + .values({ id: Project.ID.global, worktree: ref.directory, sandboxes: [] }) + .run() + .pipe(Effect.orDie) + yield* db + .insert(SessionTable) + .values( + Array.from(new Set([...sessionIDs, newcomer]), (sessionID) => ({ + id: sessionID, + project_id: Project.ID.global, + slug: "question", + directory: ref.directory, + title: "Waiting question", + version: "test", + })), + ) + .run() + .pipe(Effect.orDie) + + const created = yield* Deferred.make() + const newCreated = yield* Deferred.make() + const pending: Form.Info[] = [] + const interrupted: SessionEvent.Execution.Interrupted["data"][] = [] + const unsubscribe = yield* bus.listen((event) => + Effect.gen(function* () { + if (event.type === SessionEvent.Execution.Interrupted.type) { + interrupted.push(Schema.decodeUnknownSync(SessionEvent.Execution.Interrupted.data)(event.data)) + } + if (event.type !== Form.Event.Created.type) return + pending.push(Schema.decodeUnknownSync(Form.Event.Created.data)(event.data).form) + if (pending.length === count) yield* Deferred.succeed(created, undefined) + if (pending.length > count) yield* Deferred.succeed(newCreated, undefined) + }), + ) + yield* Effect.addFinalizer(() => unsubscribe) + const running = yield* Effect.forEach(sessionIDs, (sessionID) => + execution.resume(sessionID).pipe(Effect.exit, Effect.forkScoped), + ) + yield* Effect.addFinalizer(() => + Effect.forEach([...sessionIDs, newcomer], (sessionID) => execution.interrupt(sessionID)).pipe( + Effect.andThen(TestClock.adjust("5 minutes")), + ), + ) + yield* Deferred.await(created) + const context = yield* map.contextEffect(ref).pipe(Effect.scoped) + const forms = Context.get(context, Form.Service) + expect((yield* store.listSuspended()).toSorted()).toEqual(sessionIDs.toSorted()) + yield* Location.Service.pipe(Effect.provide(map.get(idle)), Effect.scoped) + + // Human input produces no durable activity while the question is pending. + yield* TestClock.adjust("1 minute") + yield* TestClock.adjust("62 minutes") + // Interruption has cancelled each question, but slow cleanup still owns the graph. + expect(Array.from(yield* execution.active).toSorted()).toEqual(sessionIDs.toSorted()) + expect(Array.from(yield* RcMap.keys(map.rcMap))).toEqual([ref]) + expect(yield* forms.list()).toEqual([]) + for (const form of pending) expect(yield* forms.state(form.id)).toEqual({ status: "cancelled" }) + + if (newWork) { + yield* execution.wake(newcomer) + if (admission === "other") yield* Deferred.await(newCreated) + } + yield* TestClock.adjust("5 minutes") + if (newWork) yield* Deferred.await(newCreated) + const results = yield* Effect.forEach(running, Fiber.join) + expect(results.every((exit) => exit._tag === "Failure")).toBe(true) + expect(Array.from(yield* execution.active)).toEqual(newWork ? [newcomer] : []) + expect(yield* store.listSuspended()).toEqual(newWork ? [newcomer] : []) + expect(interrupted.toSorted((a, b) => a.sessionID.localeCompare(b.sessionID))).toEqual( + sessionIDs.map((sessionID) => ({ sessionID, reason: "inactivity" })), + ) + expect(Array.from(yield* RcMap.keys(map.rcMap))).toEqual(newWork ? [ref] : []) + if (newWork) { + expect(yield* forms.list({ sessionID: newcomer })).toEqual([pending[count]]) + if (admission === "same") { + const later = LocationServiceMap.canonical({ directory: AbsolutePath.make("/later") }) + yield* Location.Service.pipe(Effect.provide(map.get(later)), Effect.scoped) + yield* TestClock.adjust("30 minutes") + // Keep fresh work active while a different graph reaches its own deadline. + yield* bus.publish(SessionEvent.Execution.Started, { sessionID: newcomer }, { location: ref }) + yield* TestClock.adjust("32 minutes") + expect(Array.from(yield* execution.active)).toEqual([newcomer]) + expect(Array.from(yield* RcMap.keys(map.rcMap))).toEqual([ref]) + } + yield* execution.interrupt(newcomer) + yield* TestClock.adjust("5 minutes") + yield* execution.awaitIdle(newcomer) + yield* TestClock.adjust("62 minutes") + expect(yield* store.listSuspended()).toEqual([]) + expect(Array.from(yield* RcMap.keys(map.rcMap))).toEqual([]) + } + }), + ) + } +}) diff --git a/packages/schema/src/session-event.ts b/packages/schema/src/session-event.ts index 34ebe69b1067..3d597ddd8866 100644 --- a/packages/schema/src/session-event.ts +++ b/packages/schema/src/session-event.ts @@ -233,7 +233,7 @@ export namespace Execution { export const Interrupted = Event.durable({ type: "session.execution.interrupted", ...options, - schema: { ...Base, reason: Schema.Literals(["user", "shutdown", "superseded"]) }, + schema: { ...Base, reason: Schema.Literals(["user", "shutdown", "superseded", "inactivity"]) }, }) export type Interrupted = typeof Interrupted.Type }