diff --git a/packages/client/src/promise/generated/types.ts b/packages/client/src/promise/generated/types.ts index 1a450e13981c..dc2bfb514cf0 100644 --- a/packages/client/src/promise/generated/types.ts +++ b/packages/client/src/promise/generated/types.ts @@ -1900,7 +1900,7 @@ export type FormField = export type FormState = | { status: "pending" } | { status: "answered"; answer: FormAnswer } - | { status: "cancelled"; message?: string } + | { status: "cancelled"; message?: string; cause?: "evicted" } export type CredentialKey = { type: "key" diff --git a/packages/core/src/form.ts b/packages/core/src/form.ts index fc8e2edb9186..886fe00eab6c 100644 --- a/packages/core/src/form.ts +++ b/packages/core/src/form.ts @@ -7,6 +7,19 @@ import { Bus } from "./bus.js" const RETENTION = Duration.minutes(10) +// Client-facing close messages must stay bounded for UI surfaces. +const CLOSE_MESSAGE_CAP = 160 + +function truncate(text: string, cap: number): string { + return text.length > cap ? `${text.slice(0, cap)}…` : text +} + +// Identifies the pending content the way toField filled it: the field's question +// text when present, else its title, else its key. +function describePendingContent(form: Info): string { + return form.fields.map((field) => `"${field.description ?? field.title ?? field.key}"`).join("; ") +} + export const ID = Form.ID export type ID = typeof ID.Type @@ -76,6 +89,13 @@ export interface ReplyInput { export interface CancelOptions { readonly message?: string + /** Why the form is being cancelled by the close path; absent for caller-initiated cancels. */ + readonly cause?: "evicted" +} + +export interface CloseOptions { + /** Why the close path is cancelling pending forms; absent for teardown without a known cause. */ + readonly cause?: "evicted" } export interface ListInput { @@ -83,7 +103,7 @@ export interface ListInput { } export interface Interface { - readonly close: Effect.Effect + readonly close: (options?: CloseOptions) => Effect.Effect readonly create: (input: CreateInput) => Effect.Effect readonly ask: (input: CreateInput) => Effect.Effect readonly get: (id: ID) => Effect.Effect @@ -204,6 +224,7 @@ export const layer = Layer.effect( const next: TerminalState = { status: "cancelled", ...(options?.message === undefined ? {} : { message: options.message }), + ...(options?.cause === undefined ? {} : { cause: options.cause }), } yield* bus.publish(Form.Event.Cancelled, { id, sessionID: entry.form.sessionID }) yield* Cache.set(forms, id, { ...entry, state: next }) @@ -212,19 +233,31 @@ export const layer = Layer.effect( ), ) - const close = Effect.sync(() => { - closed = true - }).pipe( - Effect.andThen(Cache.values(forms)), - Effect.flatMap((entries) => - Effect.forEach( - Array.from(entries).filter((entry) => entry.state.status === "pending"), - (entry) => cancel(entry.form.id).pipe(Effect.ignore), - { discard: true }, + const close = (options?: CloseOptions) => + Effect.sync(() => { + closed = true + }).pipe( + Effect.andThen(Cache.values(forms)), + Effect.flatMap((entries) => + Effect.forEach( + Array.from(entries).filter((entry) => entry.state.status === "pending"), + (entry) => + cancel(entry.form.id, { + ...(options?.cause === undefined + ? {} + : { + cause: options.cause, + message: truncate( + `The location was evicted for inactivity before this form received a reply ("${entry.form.title}": ${describePendingContent(entry.form)})`, + CLOSE_MESSAGE_CAP, + ), + }), + }).pipe(Effect.ignore), + { discard: true }, + ), ), - ), - ) - yield* Effect.addFinalizer(() => close) + ) + yield* Effect.addFinalizer(() => close()) return Service.of({ create, ask, get, list, state, reply, cancel, close }) }), diff --git a/packages/core/src/instance.ts b/packages/core/src/instance.ts index 0699bc2dfb27..f00a134b4e47 100644 --- a/packages/core/src/instance.ts +++ b/packages/core/src/instance.ts @@ -159,7 +159,7 @@ export function layer(ref: Location.Ref, options: Options = {}): Layer.Layer Effect.addFinalizer(() => Context.get(context, LocationLifecycle.Service).shutdown)), + Layer.tap((context) => Effect.addFinalizer(() => Context.get(context, LocationLifecycle.Service).shutdown())), Layer.tap(() => Effect.logInfo("location services booted", { directory: ref.directory, diff --git a/packages/core/src/location-activity.ts b/packages/core/src/location-activity.ts index 0425e9acab04..18c8a435a49c 100644 --- a/packages/core/src/location-activity.ts +++ b/packages/core/src/location-activity.ts @@ -78,7 +78,7 @@ export function layer(options: { readonly timeToLive?: Duration.Input; readonly yield* Effect.logInfo("location services evicted", { directory: entry.ref.directory, workspaceID: entry.ref.workspaceID, - }).pipe(Effect.andThen(locations.invalidate(entry.ref))) + }).pipe(Effect.andThen(locations.invalidate(entry.ref, { cause: "evicted" }))) }), { discard: true, concurrency: "unbounded" }, ) diff --git a/packages/core/src/location-lifecycle.ts b/packages/core/src/location-lifecycle.ts index ea99ab959841..1d5df7af0351 100644 --- a/packages/core/src/location-lifecycle.ts +++ b/packages/core/src/location-lifecycle.ts @@ -16,7 +16,10 @@ const isSessionEvent = Schema.is(SessionEvent.Durable) export class Service extends Context.Service< Service, - { readonly isClosed: () => boolean; readonly shutdown: Effect.Effect } + { + readonly isClosed: () => boolean + readonly shutdown: (options?: Form.CloseOptions) => Effect.Effect + } >()("@opencode/LocationLifecycle") {} const layer = Layer.effect( @@ -39,21 +42,25 @@ const layer = Layer.effect( }) yield* Effect.addFinalizer(() => unsubscribe) let closed = false - const shutdown = yield* Effect.cached( - Effect.gen(function* () { + // The close happens once: the first invocation fixes the cause. The instance + // finalizer and the invalidate close hook share this guard. + const shutdown = (options?: Form.CloseOptions) => + Effect.suspend(() => { + if (closed) return Effect.void closed = true - yield* permission.close - yield* forms.close - yield* rpc.close - yield* bus.publish( - LocationEvent.Shutdown, - {}, - { - location: Location.Ref.make({ directory: location.directory, workspaceID: location.workspaceID }), - }, - ) - }).pipe(Effect.uninterruptible), - ) + return Effect.gen(function* () { + yield* permission.close + yield* forms.close(options) + yield* rpc.close + yield* bus.publish( + LocationEvent.Shutdown, + {}, + { + location: Location.Ref.make({ directory: location.directory, workspaceID: location.workspaceID }), + }, + ) + }).pipe(Effect.uninterruptible) + }) return Service.of({ isClosed: () => closed, shutdown, diff --git a/packages/core/src/location-service-map.ts b/packages/core/src/location-service-map.ts index c24adc3b6d95..906b1e3bf195 100644 --- a/packages/core/src/location-service-map.ts +++ b/packages/core/src/location-service-map.ts @@ -3,12 +3,16 @@ import { LayerNode } from "@opencode/util/effect/layer-node" import { Node } from "@opencode/util/effect/app-node" import { AbsolutePath } from "@opencode/schema/schema" import path from "path" +import type { Form } from "./form.js" import { Location } from "./location.js" import type { Instance } from "./instance.js" export class Service extends Context.Service< Service, - LayerMap.LayerMap + LayerMap.LayerMap & { + /** Invalidate with an optional close cause, forwarded to the entry's close hook. */ + readonly invalidate: (ref: Location.Ref, options?: Form.CloseOptions) => Effect.Effect + } >()("@opencode/example/LocationServiceMap") { static get(ref: Location.Ref) { return Layer.unwrap(Effect.map(Service, (locations) => locations.get(ref))) diff --git a/packages/core/src/location-services.ts b/packages/core/src/location-services.ts index 893c4d215b73..8afb3de1dcdb 100644 --- a/packages/core/src/location-services.ts +++ b/packages/core/src/location-services.ts @@ -1,5 +1,6 @@ import { Context, Duration, Effect, Exit, Layer, LayerMap, MutableHashMap, Option } from "effect" import { LayerNode } from "@opencode/util/effect/layer-node" +import { Form } from "./form.js" import { Instance } from "./instance.js" import { Location } from "./location.js" import { LocationLifecycle } from "./location-lifecycle.js" @@ -9,6 +10,13 @@ export { LocationServiceMap } from "./location-service-map.js" export type LocationServices = Instance.Services +type BuildRecord = { + /** Cancels the entry's pending forms, carrying the invalidate cause (the scope finalizer cannot). */ + closeForms?: (options?: Form.CloseOptions) => Effect.Effect + /** The entry's lifecycle shutdown, invoked after routing detach as today. */ + close?: (options?: Form.CloseOptions) => Effect.Effect +} + export function buildLocationServiceMap( replacements: LayerNode.Replacements = [], ): Layer.Layer { @@ -16,10 +24,10 @@ export function buildLocationServiceMap( LocationServiceMap.Service, Effect.gen(function* () { const owner = yield* Effect.scope - const builds = MutableHashMap.empty }>() + const builds = MutableHashMap.empty() const inner: LayerMap.LayerMap = yield* LayerMap.make( (ref: Location.Ref) => { - const build: { close?: Effect.Effect } = {} + const build: BuildRecord = {} MutableHashMap.set(builds, ref, build) return Layer.fromBuild((memoMap, scope) => Effect.suspend(() => @@ -32,8 +40,9 @@ export function buildLocationServiceMap( const lifecycle = Context.get(exit.value, LocationLifecycle.Service) // A boot detached while in flight still needs shutdown and cancellation. if (Option.getOrUndefined(MutableHashMap.get(builds, ref)) !== build) - return yield* lifecycle.shutdown + return yield* lifecycle.shutdown() build.close = lifecycle.shutdown + build.closeForms = (options) => Context.get(exit.value, Form.Service).close(options) }) } // An explicitly invalidated build must not evict its replacement. @@ -60,15 +69,24 @@ export function buildLocationServiceMap( get: (ref: Location.Ref) => inner.get(LocationServiceMap.canonical(ref)), contextEffect: (ref: Location.Ref) => inner.contextEffect(LocationServiceMap.canonical(ref)), contextEffectOption: (ref: Location.Ref) => inner.contextEffectOption(LocationServiceMap.canonical(ref)), - invalidate: (ref: Location.Ref) => + invalidate: (ref: Location.Ref, options?: Form.CloseOptions) => Effect.suspend(() => { const key = LocationServiceMap.canonical(ref) const build = Option.getOrUndefined(MutableHashMap.get(builds, key)) MutableHashMap.remove(builds, key) + if (options?.cause !== undefined) + // A cause-carrying invalidate must deliver the cause to the pending forms' + // cancellation: with no borrowers, inner.invalidate closes the entry scope + // synchronously, and the scope finalizer's close cannot carry a cause. Cancel + // the forms with the cause before detaching; the shutdown that follows (its + // own forms close included) finds them settled. + return (build?.closeForms?.(options) ?? Effect.void).pipe( + Effect.andThen(() => inner.invalidate(key).pipe(Effect.andThen(build?.close?.(options) ?? Effect.void))), + ) // Detach routing first, then cancel interactions and notify clients. Running // steps retain their borrowed graph until they can hand off at a boundary. // Do not await a boot here: failed/in-flight builds have their own cleanup path. - return inner.invalidate(key).pipe(Effect.andThen(build?.close ?? Effect.void)) + return inner.invalidate(key).pipe(Effect.andThen(build?.close?.(options) ?? Effect.void)) }).pipe(Effect.uninterruptible), } // Cached instances borrow their owner instead of retaining its Layer scope. diff --git a/packages/core/src/session/execution.ts b/packages/core/src/session/execution.ts index eeed241ae14e..24b884fa46d5 100644 --- a/packages/core/src/session/execution.ts +++ b/packages/core/src/session/execution.ts @@ -45,7 +45,8 @@ 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" | "inactivity" +/** Causes the coordinator records for `interrupt`; published verbatim on the interrupted event. */ +export type InterruptReason = "user" | "shutdown" | "inactivity" export function terminal(exit: Exit.Exit, reason?: InterruptReason) { if (Exit.isSuccess(exit)) return { type: "succeeded" as const } @@ -93,7 +94,13 @@ export const layer = Layer.effect( const session = yield* store.get(sessionID) if (!session) return yield* Effect.die(new Error(`Session not found: ${sessionID}`)) const result = yield* SessionRunner.Service.use((runner) => - runner.drain({ sessionID, force, continuation, promotable }), + runner.drain({ + sessionID, + force, + continuation, + promotable, + interruptionReason: coordinator.interruptionReason(sessionID), + }), ).pipe( instances.provide(session), Effect.tapCause((cause) => diff --git a/packages/core/src/session/run-coordinator.ts b/packages/core/src/session/run-coordinator.ts index 9c4ba4b09a97..c7dfe9491018 100644 --- a/packages/core/src/session/run-coordinator.ts +++ b/packages/core/src/session/run-coordinator.ts @@ -25,6 +25,8 @@ export interface Coordinator { reason?: Reason, options?: { readonly awaitSettlement?: boolean }, ) => Effect.Effect + /** Reads the recorded interruption cause for the current execution. Undefined when idle or never interrupted. */ + readonly interruptionReason: (key: Key) => 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 } @@ -166,6 +168,10 @@ export const make = (options: { return true }) + // The record stays in the map until settle, so drains can read the cause of their own + // interruption while they settle (the settled hook reads the field directly instead). + const interruptionReason = (key: Key) => Effect.sync(() => executions.get(key)?.interruptionReason) + // One execution's `done` already spans coalesced continuations; re-check after it // settles to cover a successor execution started by a late doorbell. const awaitIdle = (key: Key): Effect.Effect => @@ -180,6 +186,7 @@ export const make = (options: { isActive, run, wake, + interruptionReason, interrupt: (key, reason, options) => Effect.suspend(() => { const execution = executions.get(key) diff --git a/packages/core/src/session/runner/index.ts b/packages/core/src/session/runner/index.ts index 644983587d0b..4fe4927a2bd5 100644 --- a/packages/core/src/session/runner/index.ts +++ b/packages/core/src/session/runner/index.ts @@ -5,6 +5,7 @@ import { Context, Data, Effect } from "effect" import { SessionSchema } from "../schema.js" import type { Promotable } from "../inbox.js" import type { AgentNotFoundError, MessageDecodeError, StepFailedError } from "../error.js" +import type { InterruptReason } from "../execution.js" import { SessionRunnerModel } from "./model.js" import type { Instructions } from "../../instructions/index.js" @@ -34,6 +35,8 @@ export interface Interface { readonly continuation?: Continuation /** "steer" settles the active intent without promoting queued next-turn work. */ readonly promotable?: Promotable + /** Reads the caller's recorded interruption cause; absent for direct drains with no coordinator. */ + readonly interruptionReason?: Effect.Effect }) => Effect.Effect } diff --git a/packages/core/src/session/runner/llm.ts b/packages/core/src/session/runner/llm.ts index 3ffe4ed6e900..d0b29866cc53 100644 --- a/packages/core/src/session/runner/llm.ts +++ b/packages/core/src/session/runner/llm.ts @@ -6,6 +6,7 @@ import { Cause, Effect, Exit, FiberMap, Layer } from "effect" import { Database } from "../../database/database.js" import { Bus } from "../../bus.js" import { LocationLifecycle } from "../../location-lifecycle.js" +import type { InterruptReason } from "../execution.js" import { InstructionState } from "../instruction-state.js" import { SessionCompaction } from "../compaction.js" import { SessionContext } from "../context.js" @@ -58,6 +59,8 @@ const layer = Layer.effect( let step = input.continuation?.step ?? 1 let entering = true const promotable = input.promotable ?? "input" + // Cause reads happen at settlement time; the accessor is bound per drain call. + const interruptionReason = input.interruptionReason ?? Effect.succeed(undefined) if (!force && !continuing) { const pending = yield* SessionInbox.nextPromotable(db, sessionID, "input") if (!pending) return DrainResult.Complete() @@ -188,7 +191,7 @@ const layer = Layer.effect( while (true) { const next = yield* advanceToStep() if (next._tag !== "Ready") return next - continuing = yield* runStep(next.context, step) + continuing = yield* runStep(next.context, step, interruptionReason) step++ force = false entering = false @@ -203,7 +206,11 @@ const layer = Layer.effect( }) /** Owns logical Step policy; each attempt owns its streaming, tools, and durable settlement. */ - const runStep = Effect.fn("SessionRunner.runStep")(function* (first: SessionContext.Loaded, step: number) { + const runStep = Effect.fn("SessionRunner.runStep")(function* ( + first: SessionContext.Loaded, + step: number, + interruptionReason: Effect.Effect, + ) { const sessionID = first.session.id let assistantMessageID = SessionMessage.ID.create() const retry = yield* SessionRunnerRetry.make(bus, sessionID) @@ -244,6 +251,8 @@ const layer = Layer.effect( }) const outcome = yield* steps.attempt({ isLocationClosed: lifecycle.isClosed, + // Read at settlement time inside the attempt: the interrupt lands mid-step. + interruptionReason, sessionID, assistantMessageID, agent: loaded.agent.id, diff --git a/packages/core/src/session/runner/step.ts b/packages/core/src/session/runner/step.ts index f42f334d5b32..2ef5ea3a400e 100644 --- a/packages/core/src/session/runner/step.ts +++ b/packages/core/src/session/runner/step.ts @@ -25,6 +25,7 @@ import { SessionModelRequest } from "../model-request.js" import { SessionSchema } from "../schema.js" import { toSessionError } from "../to-session-error.js" import { SessionUsage } from "../usage.js" +import type { InterruptReason } from "../execution.js" import { SessionRunnerModel } from "./model.js" import { createLLMEventPublisher } from "./publish-llm-event.js" import { SessionRunnerRetry } from "./retry.js" @@ -43,6 +44,8 @@ export const Outcome = Data.taggedEnum() interface Input { readonly isLocationClosed: () => boolean + /** Reads the coordinator's recorded interruption cause; evaluated here at settlement, after the interrupt lands. */ + readonly interruptionReason: Effect.Effect readonly sessionID: SessionSchema.ID readonly assistantMessageID: SessionMessage.ID readonly agent: Agent.ID @@ -59,6 +62,10 @@ interface Input { } const TOOLS_INTERRUPTED = { type: "aborted", message: "Tool execution interrupted" } as const +const TOOLS_INTERRUPTED_EVICTION = { + type: "aborted", + message: "Tool execution interrupted: location idle eviction", +} as const const STEP_INTERRUPTED = { type: "aborted", message: "Step interrupted" } as const const RESULT_MISSING = { type: "tool.result-missing", message: "Provider did not return a tool result" } as const @@ -199,15 +206,21 @@ export const make = Effect.gen(function* () { for (const decline of tools.declines) yield* publisher.failTool(decline.call.id, { type: "aborted", - message: input.isLocationClosed() - ? "Interaction cancelled because the location shut down" - : decline.reason._tag === "QuestionTool.CancelledError" + // A question cancellation carries its own truthful message (with the pending + // content); it outranks the closed-location blanket. Other declines keep it. + message: + decline.reason._tag === "QuestionTool.CancelledError" ? decline.reason.message - : "The user declined this tool call", + : input.isLocationClosed() + ? "Interaction cancelled because the location shut down" + : "The user declined this tool call", }) const interrupted = tools.declines.length > 0 || streamInterrupted || tools.interrupted const toolFailure = interrupted - ? TOOLS_INTERRUPTED + ? // Other causes keep the blanket wording; only eviction is distinguishable today. + (yield* input.interruptionReason) === "inactivity" + ? TOOLS_INTERRUPTED_EVICTION + : TOOLS_INTERRUPTED : tools.failure !== undefined ? toSessionError(Cause.squash(tools.failure)) : recorded.providerFailed diff --git a/packages/core/src/tool/plugin/question.ts b/packages/core/src/tool/plugin/question.ts index 60d115eae35f..64df4faa14e0 100644 --- a/packages/core/src/tool/plugin/question.ts +++ b/packages/core/src/tool/plugin/question.ts @@ -29,12 +29,27 @@ export const Output = Schema.Struct({ }) export type Output = typeof Output.Type -export class CancelledError extends Schema.TaggedError()("QuestionTool.CancelledError", {}) { +export class CancelledError extends Schema.TaggedError()( + "QuestionTool.CancelledError", + // Pre-composed truthful sentence for the declines message; absent = plain dismissal. + { detail: Schema.optional(Schema.String) }, +) { override get message() { - return "The user dismissed this question" + return this.detail ?? "The user dismissed this question" } } +// Model-facing decline details must stay bounded. +const DETAIL_CONTENT_CAP = 200 + +function truncate(text: string, cap: number): string { + return text.length > cap ? `${text.slice(0, cap)}…` : text +} + +// The pending content, quoted the way the answered path reports it (toModelContent). +const renderPendingQuestions = (questions: ReadonlyArray) => + truncate(questions.map((question) => `"${question.header}"="${question.question}"`).join("; "), DETAIL_CONTENT_CAP) + export const toModelContent = (questions: ReadonlyArray, answers: ReadonlyArray) => { const formatted = questions .map( @@ -92,9 +107,23 @@ export const Plugin = { // resurfaces as a typed failure at SessionModelRequest.executeTool. // A dismissal with a message (e.g. from a non-interactive client) is ordinary // model-facing feedback: the model continues instead of the step ending. + // A cause-carrying cancellation is the close path ending the step; its detail + // names the eviction and the pending content. The bare dismissal also carries + // the pending content so the declines message is never content-free. + if (state.status === "cancelled" && state.cause === "evicted") + return Effect.die( + new CancelledError({ + detail: `This question was not answered: the location was evicted for inactivity while it awaited a reply — ${renderPendingQuestions(input.questions)}`, + }), + ) if (state.status === "cancelled" && state.message !== undefined) return Effect.fail(new ToolFailure({ message: state.message })) - if (state.status === "cancelled") return Effect.die(new CancelledError()) + if (state.status === "cancelled") + return Effect.die( + new CancelledError({ + detail: `The user dismissed this question: ${renderPendingQuestions(input.questions)}`, + }), + ) const output = { answers: input.questions.map((_, index): Question.Answer => { const value = state.answer[`q${index}`] diff --git a/packages/core/test/form.test.ts b/packages/core/test/form.test.ts index fe929fd97b4d..f6202b8117eb 100644 --- a/packages/core/test/form.test.ts +++ b/packages/core/test/form.test.ts @@ -354,4 +354,36 @@ describe("Form", () => { expect(yield* service.state(formID)).toEqual({ status: "answered", answer: { name: "Ava" } }) }), ) + + it.effect("marks pending forms with the eviction cause and content when closed with a cause", () => + Effect.gen(function* () { + const service = yield* Form.Service + const pending = yield* service.create({ + sessionID: SessionSchema.ID.make("ses_evicted"), + title: "Questions", + fields: [{ key: "runtime", title: "Runtime", description: "Which runtime?", type: "string" }], + }) + + yield* service.close({ cause: "evicted" }) + + const state = yield* service.state(pending.id) + expect(state).toMatchObject({ status: "cancelled", cause: "evicted" }) + expect(state).toHaveProperty( + "message", + expect.stringContaining("The location was evicted for inactivity before this form received a reply"), + ) + expect(state).toHaveProperty("message", expect.stringContaining('"Which runtime?"')) + }), + ) + + it.effect("cancels pending forms without a cause on a plain close", () => + Effect.gen(function* () { + const service = yield* Form.Service + const pending = yield* service.create(input) + + yield* service.close() + + expect(yield* service.state(pending.id)).toEqual({ status: "cancelled" }) + }), + ) }) diff --git a/packages/core/test/location-activity.test.ts b/packages/core/test/location-activity.test.ts index 4ca873afbc01..377f2c1c7ebb 100644 --- a/packages/core/test/location-activity.test.ts +++ b/packages/core/test/location-activity.test.ts @@ -1,5 +1,5 @@ import { describe, expect } from "bun:test" -import { Context, Deferred, Duration, Effect, Fiber, Layer, LayerMap, RcMap, Schema } from "effect" +import { Context, Deferred, Duration, Effect, Exit, Fiber, Layer, LayerMap, MutableHashMap, Option, RcMap, Schema } from "effect" import { TestClock } from "effect/testing" import { AppNodeBuilder } from "@opencode/core/effect/app-node-builder" import { LayerNode } from "@opencode/util/effect/layer-node" @@ -24,48 +24,76 @@ 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. +// The build record + closeForms hook mirrors production's buildLocationServiceMap: the +// entry's forms close receives the invalidate cause (the scope finalizer cannot). const locations = Layer.effect( LocationServiceMap.Service, Effect.gen(function* () { const bus = yield* Bus.Service + const builds = MutableHashMap.empty< + Location.Ref, + { closeForms?: (options?: Form.CloseOptions) => Effect.Effect } + >() 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" }], + (ref: Location.Ref) => { + const build: { closeForms?: (options?: Form.CloseOptions) => Effect.Effect } = {} + MutableHashMap.set(builds, ref, build) + return Layer.fromBuild((memoMap, scope) => + Effect.suspend(() => + Layer.buildWithMemoMap( + // 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( - 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, + memoMap, + scope, + ), + ).pipe( + Effect.onExit((exit) => { + if (Exit.isSuccess(exit)) { + const forms = Context.get(exit.value, Form.Service) + build.closeForms = (options) => forms.close(options) + return Effect.void + } + if (Option.getOrUndefined(MutableHashMap.get(builds, ref)) !== build) return Effect.void + MutableHashMap.remove(builds, ref) + return Effect.void }), ), - ).pipe( - Layer.provideMerge(Form.layer), - Layer.provide(Layer.succeed(Bus.Service, bus)), - Layer.fresh, - ) as unknown as Layer.Layer, + ) + }, { idleTimeToLive: Duration.infinity }, ) return { @@ -73,7 +101,15 @@ const locations = Layer.effect( 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)), + invalidate: (ref: Location.Ref, options?: Form.CloseOptions) => { + const key = LocationServiceMap.canonical(ref) + const build = Option.getOrUndefined(MutableHashMap.get(builds, key)) + // Mirror production: a cause-carrying invalidate cancels pending forms with the + // cause before detaching the entry, so the scope finalizer never bare-cancels. + if (options?.cause !== undefined) + return (build?.closeForms?.(options) ?? Effect.void).pipe(Effect.andThen(() => map.invalidate(key))) + return map.invalidate(key).pipe(Effect.andThen(build?.closeForms?.(options) ?? Effect.void)) + }, } }), ) @@ -218,4 +254,31 @@ describe("LocationActivity eviction", () => { }), ) } + + it.effect("cancels an idle location's pending form with the eviction cause on the close path", () => + Effect.gen(function* () { + const map = yield* LocationServiceMap.Service + const ref = LocationServiceMap.canonical({ directory: AbsolutePath.make("/project") }) + const context = yield* map.contextEffect(ref).pipe(Effect.scoped) + const forms = Context.get(context, Form.Service) + // No execution owns this form (the elicitation shape): the sweep finds no + // owner, so the close path — not an interrupt — cancels it. + const created = yield* forms.create({ + sessionID: Session.ID.make("ses_idle_question"), + title: "Questions", + fields: [{ key: "runtime", title: "Runtime", description: "Which runtime?", type: "string" }], + }) + + yield* TestClock.adjust("62 minutes") + + const state = yield* forms.state(created.id) + expect(state).toMatchObject({ status: "cancelled", cause: "evicted" }) + expect(state).toHaveProperty( + "message", + expect.stringContaining("The location was evicted for inactivity before this form received a reply"), + ) + expect(state).toHaveProperty("message", expect.stringContaining('"Which runtime?"')) + expect(Array.from(yield* RcMap.keys(map.rcMap))).toEqual([]) + }), + ) }) diff --git a/packages/core/test/session-runner.test.ts b/packages/core/test/session-runner.test.ts index ce6058c37204..bebba8eb81bf 100644 --- a/packages/core/test/session-runner.test.ts +++ b/packages/core/test/session-runner.test.ts @@ -34,6 +34,7 @@ import { EventTable } from "@opencode/core/event/sql" import { Project } from "@opencode/core/project" import { ProjectTable } from "@opencode/core/project/sql" import { Form } from "@opencode/core/form" +import { LocationLifecycle } from "@opencode/core/location-lifecycle" import { AbsolutePath } from "@opencode/core/schema" import { Session } from "@opencode/core/session" import { Snapshot } from "@opencode/core/snapshot" @@ -439,14 +440,23 @@ const layer = Layer.unwrap( continuation?: SessionRunner.Continuation, ): Effect.Effect { return sessionRunner - .drain({ sessionID, force, continuation }) + .drain({ + sessionID, + force, + continuation, + interruptionReason: coordinator.interruptionReason(sessionID), + }) .pipe( Effect.flatMap((result) => result._tag === "Complete" ? Effect.void : drain(sessionID, false, result.continuation), ), ) } - const coordinator = yield* SessionRunCoordinator.make({ + const coordinator = yield* SessionRunCoordinator.make< + Session.ID, + SessionRunner.RunError, + SessionExecution.InterruptReason + >({ drain: (sessionID, force) => drain(sessionID, force), }) return SessionExecution.Service.of({ @@ -454,7 +464,7 @@ const layer = Layer.unwrap( isActive: coordinator.isActive, resume: coordinator.run, wake: coordinator.wake, - interrupt: (sessionID) => coordinator.interrupt(sessionID), + interrupt: (sessionID, options) => coordinator.interrupt(sessionID, options?.reason ?? "user", options), awaitIdle: coordinator.awaitIdle, }) }), @@ -492,6 +502,16 @@ const layer = Layer.unwrap( LocationServiceMap.node.replace(promptLocationNode), Model.node.replace(promptModels), SessionExecution.node.replace(execution), + // Controllable closed-state: the runner reads isClosed at declines/settlement. + LocationLifecycle.node.replace( + Layer.succeed( + LocationLifecycle.Service, + LocationLifecycle.Service.of({ + isClosed: () => harnessLocationClosed, + shutdown: () => Effect.void, + }), + ), + ), ], ) }), @@ -499,6 +519,9 @@ const layer = Layer.unwrap( const it = testEffect(layer) const sessionID = Session.ID.make("ses_runner_test") const otherSessionID = Session.ID.make("ses_runner_other") +// Flipped by scenarios that need the runner to observe a closed location mid-step +// (the real lifecycle never shuts down inside this harness); reset by setup. +let harnessLocationClosed = false const insertSession = (id: Session.ID) => Effect.gen(function* () { @@ -519,6 +542,7 @@ const insertSession = (id: Session.ID) => }) const setup = Effect.gen(function* () { + harnessLocationClosed = false const { db } = yield* Database.Service const bus = yield* Bus.Service const sessionInbox = yield* SessionInbox.Service @@ -5016,6 +5040,84 @@ describe("SessionRunnerLLM", () => { ]) }) + scenario("reports the question cancellation detail over the closed-location blanket", function* (s) { + const registry = yield* Tool.Service + yield* transformTools( + registry, + { + question: { + name: "question", + description: "Ask the user", + input: Schema.Struct({}), + output: Schema.Struct({}), + execute: () => + Effect.sync(() => { + // The location closes while the question is parked; the decline settles closed. + harnessLocationClosed = true + }).pipe(Effect.andThen(Effect.die(new QuestionTool.CancelledError({ detail: "The question carried its content" })))), + }, + }, + { codemode: false }, + ) + yield* s.admit("Ask after close") + const failed = yield* Deferred.make() + const unsubscribe = yield* s.bus.listen((event) => + event.type === SessionEvent.Tool.Failed.type + ? Deferred.succeed(failed, Schema.decodeUnknownSync(SessionEvent.Tool.Failed.data)(event.data)) + : Effect.void, + ) + yield* Effect.addFinalizer(() => unsubscribe) + yield* s.llm.push(TestLLM.tool("call-question", "question", {})) + + // A decline on a closed location completes with needsContinuation; this harness's + // drain loop has no reload handoff, so the run is interrupted after the event lands. + const run = yield* s.resume.pipe(Effect.forkChild) + const failure = yield* Deferred.await(failed) + yield* Fiber.interrupt(run) + + expect(failure.id).toBe("call-question") + expect(failure.error).toEqual({ type: "aborted", message: "The question carried its content" }) + }) + + scenario("keeps the closed-location blanket for permission declines", function* (s) { + const registry = yield* Tool.Service + yield* transformTools( + registry, + { + declined: { + name: "declined", + description: "Fail because the user declined approval", + input: Schema.Struct({}), + output: Schema.Struct({}), + execute: () => + Effect.sync(() => { + harnessLocationClosed = true + }).pipe(Effect.andThen(Effect.die(new Permission.DeclinedError()))), + }, + }, + { codemode: false }, + ) + yield* s.admit("Reject permission after close") + const failed = yield* Deferred.make() + const unsubscribe = yield* s.bus.listen((event) => + event.type === SessionEvent.Tool.Failed.type + ? Deferred.succeed(failed, Schema.decodeUnknownSync(SessionEvent.Tool.Failed.data)(event.data)) + : Effect.void, + ) + yield* Effect.addFinalizer(() => unsubscribe) + yield* s.llm.push(TestLLM.tool("call-declined", "declined", {})) + + const run = yield* s.resume.pipe(Effect.forkChild) + const failure = yield* Deferred.await(failed) + yield* Fiber.interrupt(run) + + expect(failure.id).toBe("call-declined") + expect(failure.error).toEqual({ + type: "aborted", + message: "Interaction cancelled because the location shut down", + }) + }) + scenario("awaits started local tools before surfacing provider stream failure", function* (s) { yield* s.admit("Settle before failing") // Non-retryable so the step settles terminally instead of continuing after tool output. @@ -5096,6 +5198,62 @@ describe("SessionRunnerLLM", () => { expect(messageRoles(s.requests[0])).toEqual(["user", "assistant", "tool"]) }) + scenario("reports the eviction cause on interrupted tools", function* (s) { + yield* s.admit("Evict blocked tool") + const tools = yield* s.blockTools() + yield* s.llm.push( + TestLLM.hangAfter( + LLMEvent.stepStart({ index: 0 }), + LLMEvent.toolCall({ id: "call-evicted", name: "echo", input: { text: "blocked" } }), + ), + ) + + const run = yield* s.resume.pipe(Effect.forkChild) + yield* tools.started + const execution = yield* SessionExecution.Service + expect(yield* execution.interrupt(sessionID, { reason: "inactivity" })).toBe(true) + + expect(yield* Fiber.await(run)).toMatchObject({ _tag: "Failure" }) + const context = yield* s.context + expect(context).toMatchObject([ + Expected.user("Evict blocked tool"), + Expected.assistant({}, [ + Expected.failedTool( + { id: "call-evicted" }, + { error: { type: "aborted", message: "Tool execution interrupted: location idle eviction" } }, + ), + ]), + ]) + }) + + scenario("keeps the blanket wording for user-caused interrupts", function* (s) { + yield* s.admit("Stop blocked tool") + const tools = yield* s.blockTools() + yield* s.llm.push( + TestLLM.hangAfter( + LLMEvent.stepStart({ index: 0 }), + LLMEvent.toolCall({ id: "call-stopped", name: "echo", input: { text: "blocked" } }), + ), + ) + + const run = yield* s.resume.pipe(Effect.forkChild) + yield* tools.started + const execution = yield* SessionExecution.Service + expect(yield* execution.interrupt(sessionID, { reason: "user" })).toBe(true) + + expect(yield* Fiber.await(run)).toMatchObject({ _tag: "Failure" }) + const context = yield* s.context + expect(context).toMatchObject([ + Expected.user("Stop blocked tool"), + Expected.assistant({}, [ + Expected.failedTool( + { id: "call-stopped" }, + { error: { type: "aborted", message: "Tool execution interrupted" } }, + ), + ]), + ]) + }) + scenario("interrupts a blocked step without local tool execution", function* (s) { yield* s.admit("Interrupt provider") const stream = yield* s.llm.gate diff --git a/packages/core/test/session-step.test.ts b/packages/core/test/session-step.test.ts index 7cbe0a991783..00c64112773b 100644 --- a/packages/core/test/session-step.test.ts +++ b/packages/core/test/session-step.test.ts @@ -101,6 +101,7 @@ for (const fixture of [ const result = yield* steps .attempt({ isLocationClosed: () => false, + interruptionReason: Effect.succeed(undefined), sessionID, assistantMessageID, agent: Agent.defaultID, diff --git a/packages/core/test/tool-question.test.ts b/packages/core/test/tool-question.test.ts index bd397f8ea459..131b069116a9 100644 --- a/packages/core/test/tool-question.test.ts +++ b/packages/core/test/tool-question.test.ts @@ -19,6 +19,8 @@ const assertions: Permission.AssertInput[] = [] let captured: Form.CreateInput | undefined let reject = false let deny = false +let cancelCause: "evicted" | undefined +let cancelMessage: string | undefined const capturedInput = () => captured const questionInput = { questions: [ @@ -53,7 +55,13 @@ const form = Layer.mock(Form.Service, { Effect.andThen( Effect.sync( (): Form.TerminalState => - reject ? { status: "cancelled" } : { status: "answered", answer: { q0: "Build", q1: ["Dev"] } }, + reject + ? { + status: "cancelled", + ...(cancelCause === undefined ? {} : { cause: cancelCause }), + ...(cancelMessage === undefined ? {} : { message: cancelMessage }), + } + : { status: "answered", answer: { q0: "Build", q1: ["Dev"] } }, ), ), ), @@ -256,8 +264,69 @@ describe("QuestionTool", () => { if (Exit.isFailure(exit)) { const error = Cause.squash(exit.cause) expect(error).toBeInstanceOf(QuestionTool.CancelledError) - expect(error).toHaveProperty("message", "The user dismissed this question") + // The dismissal detail names the pending content, so the declines message is never + // content-free; the fixed class message stays the default when no detail is set. + expect(error).toHaveProperty("message", expect.stringContaining("The user dismissed this question")) + expect(error).toHaveProperty("message", expect.stringContaining('"Continue"="Continue?"')) } }), ) + + it.effect("reports the eviction cause with the pending question content", () => + Effect.gen(function* () { + captured = undefined + reject = true + deny = false + cancelCause = "evicted" + yield* Effect.addFinalizer(() => + Effect.sync(() => { + cancelCause = undefined + }), + ) + const registryService = yield* Tool.Service + const fiber = yield* executeTool(registryService, { + sessionID, + ...toolIdentity, + call: { type: "tool-call", id: "call-question", name: "question", input: questionInput }, + }).pipe(Effect.forkScoped) + + const exit = yield* Fiber.await(fiber) + expect(Exit.isFailure(exit)).toBe(true) + if (Exit.isFailure(exit)) { + const error = Cause.squash(exit.cause) + expect(error).toBeInstanceOf(QuestionTool.CancelledError) + expect(error).toHaveProperty( + "message", + expect.stringContaining("the location was evicted for inactivity"), + ) + expect(error).toHaveProperty("message", expect.stringContaining('"Continue"="Continue?"')) + } + }), + ) + + it.effect("surfaces a client-provided cancellation message as model-facing feedback", () => + Effect.gen(function* () { + captured = undefined + reject = true + deny = false + cancelMessage = "Nobody is watching the session" + yield* Effect.addFinalizer(() => + Effect.sync(() => { + cancelMessage = undefined + }), + ) + const registryService = yield* Tool.Service + + expect( + yield* executeTool(registryService, { + sessionID, + ...toolIdentity, + call: { type: "tool-call", id: "call-question", name: "question", input: questionInput }, + }), + ).toEqual({ + status: "error", + error: { type: "tool.execution", message: "Nobody is watching the session" }, + }) + }), + ) }) diff --git a/packages/schema/src/form.ts b/packages/schema/src/form.ts index adbf14a9b65e..66aef91bbdd9 100644 --- a/packages/schema/src/form.ts +++ b/packages/schema/src/form.ts @@ -150,7 +150,12 @@ export const State = Schema.Union([ Schema.Struct({ status: Schema.Literal("pending") }), Schema.Struct({ status: Schema.Literal("answered"), answer: Answer }), // A message tells the asker why nobody answered, e.g. a non-interactive client. - Schema.Struct({ status: Schema.Literal("cancelled"), message: Schema.String.pipe(optional) }), + // A cause records the close-path event that cancelled the form, e.g. idle eviction. + Schema.Struct({ + status: Schema.Literal("cancelled"), + message: Schema.String.pipe(optional), + cause: Schema.Literal("evicted").pipe(optional), + }), ]) .pipe(Schema.toTaggedUnion("status")) .annotate({ identifier: "Form.State" })