Compare commits

...

5 Commits

Author SHA1 Message Date
Kit Langton 6cb2f00d15 refactor(core): simplify interrupt continuation 2026-08-15 14:57:49 -04:00
opencode-agent[bot] 73700065bf chore: generate 2026-08-15 18:35:53 +00:00
Kit Langton 07a82a442e fix(core): keep queued work parked after interrupt (#42682) 2026-08-15 18:34:43 +00:00
Kit Langton 55be895e14 fix(tui): scope session picker by directory (#42804) 2026-08-15 18:30:35 +00:00
Kit Langton 58a36a1560 fix(tui): preserve path suffix in prompt footer (#42800) 2026-08-15 18:21:07 +00:00
21 changed files with 446 additions and 80 deletions
+5
View File
@@ -0,0 +1,5 @@
---
"@opencode-ai/core": patch
---
Simplify interrupt continuation: the steer-scoped resume decision now lives in SessionExecution as a post-cleanup inbox check, and the run coordinator drops its continuation state machine. Wakes arriving during cancellation cleanup now restart a normal full drain, and interrupting an idle session with continue now resumes pending steering input.
+5 -7
View File
@@ -789,7 +789,10 @@ const layer = Layer.effect(
return false
}),
)
if (recovered) return
if (recovered) {
yield* execution.wakeActive(input.sessionID)
return
}
yield* execution.wake(input.sessionID)
}),
compact: Effect.fn("Session.compact")(function* (input) {
@@ -873,12 +876,7 @@ const layer = Layer.effect(
),
),
interrupt: Effect.fn("Session.interrupt")((sessionID, options) =>
Effect.uninterruptible(
Effect.gen(function* () {
yield* execution.interrupt(sessionID)
if (options?.continue && (yield* SessionInbox.has(db, sessionID, "any"))) yield* execution.wake(sessionID)
}),
),
Effect.uninterruptible(execution.interrupt(sessionID, options)),
),
revert: {
stage: Effect.fn("Session.revert.stage")(function* (input) {
+22 -11
View File
@@ -1,7 +1,8 @@
export * as SessionExecution from "./execution.js"
import { Cause, Context, Effect, Exit, Layer, Stream } from "effect"
import { Cause, Context, Effect, Exit, Layer } from "effect"
import { Bus } from "../bus.js"
import { Database } from "../database/database.js"
import { LocationServiceMap } from "../location-service-map.js"
import { makeGlobalNode } from "@opencode-ai/util/effect/app-node"
import { SessionEvent } from "./event.js"
@@ -11,6 +12,7 @@ import { SessionSchema } from "./schema.js"
import { SessionStore } from "./store.js"
import { toSessionError } from "./to-session-error.js"
import { UserInterruptedError } from "./error.js"
import { SessionInbox } from "./inbox.js"
export interface Interface {
/** Snapshots active execution owned by this process. */
@@ -19,8 +21,10 @@ export interface Interface {
readonly resume: (sessionID: SessionSchema.ID) => Effect.Effect<void, SessionRunner.RunError>
/** Registers newly recorded work. Repeated wakeups may coalesce. */
readonly wake: (sessionID: SessionSchema.ID) => Effect.Effect<void>
/** Wakes only an active execution, preserving its current input eligibility. */
readonly wakeActive: (sessionID: SessionSchema.ID) => Effect.Effect<void>
/** Interrupt active work owned by this process. Idle interruption is a no-op. */
readonly interrupt: (sessionID: SessionSchema.ID) => Effect.Effect<void>
readonly interrupt: (sessionID: SessionSchema.ID, options?: { readonly continue?: boolean }) => Effect.Effect<void>
/** 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<void>
}
@@ -45,6 +49,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 = <A>(sessionID: SessionSchema.ID, effect: Effect.Effect<A>) =>
effect.pipe(
Effect.tapCause((cause) =>
@@ -71,12 +76,13 @@ export const layer = Layer.effect(
sessionID: SessionSchema.ID,
force: boolean,
continuation?: SessionRunner.Continuation,
promotable: SessionInbox.Promotable = "input",
): Effect.Effect<void, SessionRunner.RunError> {
return Effect.gen(function* () {
const session = yield* store.get(sessionID)
if (!session) return yield* Effect.die(new Error(`Session not found: ${sessionID}`))
const result = yield* SessionRunner.Service.use((runner) =>
runner.drain({ sessionID, force, continuation }),
runner.drain({ sessionID, force, continuation, promotable }),
).pipe(
Effect.provide(locations.get(session.location)),
Effect.tapCause((cause) =>
@@ -86,7 +92,7 @@ export const layer = Layer.effect(
),
)
if (result.type === "complete") return
return yield* drain(sessionID, false, result.continuation)
return yield* drain(sessionID, false, result.continuation, promotable)
})
}
const coordinator = yield* SessionRunCoordinator.make<SessionSchema.ID, SessionRunner.RunError, InterruptReason>({
@@ -95,7 +101,7 @@ export const layer = Layer.effect(
sessionID,
bus.publish(SessionEvent.Execution.Started, { sessionID }, claimOnCommit(sessionID)),
),
drain: (sessionID, force) => drain(sessionID, force),
drain: (sessionID, force, promotable) => drain(sessionID, force, undefined, promotable),
// One terminal observation per busy period, covering every coalesced drain.
settled: (sessionID, exit, reason) =>
reportLifecycle(
@@ -127,16 +133,20 @@ export const layer = Layer.effect(
}),
),
})
yield* bus.subscribe(SessionEvent.Moved).pipe(
Stream.runForEach((event) => coordinator.wake(event.data.sessionID)),
Effect.forkScoped,
)
return Service.of({
active: coordinator.active,
interrupt: (sessionID) => coordinator.interrupt(sessionID, "user"),
interrupt: (sessionID, options) =>
Effect.gen(function* () {
yield* coordinator.interrupt(sessionID, "user")
if (!options?.continue) return
// Resume only steering input from the interrupted intent. Queued next-turn work
// stays parked: a steer-scoped drain never promotes queue-delivery rows.
if (yield* SessionInbox.has(db, sessionID, "steer")) yield* coordinator.wake(sessionID, "steer")
}),
resume: coordinator.run,
wake: coordinator.wake,
wakeActive: coordinator.wakeActive,
awaitIdle: coordinator.awaitIdle,
})
}),
@@ -145,7 +155,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. */
@@ -155,6 +165,7 @@ export const noopLayer = Layer.succeed(
active: Effect.succeed(new Set()),
resume: () => Effect.void,
wake: () => Effect.void,
wakeActive: () => Effect.void,
interrupt: () => Effect.void,
awaitIdle: () => Effect.void,
}),
+8
View File
@@ -349,6 +349,14 @@ export const nextSteer = Effect.fn("SessionInbox.nextSteer")(function* (
return row ? fromRow(row) : undefined
})
export const nextPromotable = Effect.fn("SessionInbox.nextPromotable")(function* (
db: DatabaseService,
sessionID: SessionSchema.ID,
promotable: Promotable,
) {
return (yield* nextSteer(db, sessionID)) ?? (promotable === "input" ? yield* nextQueued(db, sessionID) : undefined)
})
/**
* Which pending rows count: "any" counts every row, while "input" means any
* item in either delivery mode.
+33 -18
View File
@@ -1,6 +1,7 @@
export * as SessionRunCoordinator from "./run-coordinator.js"
import { Deferred, Effect, Exit, Fiber, FiberSet, Scope } from "effect"
import type { Promotable } from "./inbox.js"
/** Serializes execution for each key while allowing different keys to run concurrently. */
export interface Coordinator<Key, E, Reason = never> {
@@ -9,7 +10,9 @@ export interface Coordinator<Key, E, Reason = never> {
/** Starts an execution while idle, or joins the active execution and returns its exit. */
readonly run: (key: Key) => Effect.Effect<void, E>
/** Rings the doorbell: an idle key starts an execution; an active one drains again before settling. */
readonly wake: (key: Key) => Effect.Effect<void>
readonly wake: (key: Key, scope?: Promotable) => Effect.Effect<void>
/** Rings the current execution's doorbell with its existing scope. Idle keys remain idle. */
readonly wakeActive: (key: Key) => Effect.Effect<void>
/** Stops the active execution, clears its doorbell, and waits for cleanup. No-op when idle. */
readonly interrupt: (key: Key, reason?: Reason) => Effect.Effect<void>
/** Resolves once no execution is active for the key. Returns immediately when already idle and never starts work. */
@@ -19,14 +22,16 @@ export interface Coordinator<Key, E, Reason = never> {
/**
* One execution is a busy period for one key: one fiber that drains from the first wake
* until the key would stay idle. `pendingWake` is the doorbell: work recorded during the
* execution rings it, and the execution loop drains again instead of ending. The doorbell
* closes the gap between a drain's last eligibility check and the idle transition, since
* those cannot be one atomic step. `done` resolves joiners with this execution's exit.
* execution rings it with the scope that work needs, and the execution loop drains again
* instead of ending. The doorbell closes the gap between a drain's last eligibility check
* and the idle transition, since those cannot be one atomic step. `done` resolves joiners
* with this execution's exit.
*/
type Execution<E, Reason> = {
readonly done: Deferred.Deferred<void, E>
owner?: Fiber.Fiber<void>
pendingWake: boolean
scope: Promotable
pendingWake?: Promotable
stopping: boolean
interruptionReason?: Reason
}
@@ -43,7 +48,7 @@ type Execution<E, Reason> = {
* ```
*/
export const make = <Key, E, Reason = never>(options: {
readonly drain: (key: Key, force: boolean) => Effect.Effect<void, E>
readonly drain: (key: Key, force: boolean, scope: Promotable) => Effect.Effect<void, E>
/** Runs once when a process-local busy period begins, before its first drain. */
readonly started?: (key: Key) => Effect.Effect<void>
/**
@@ -57,21 +62,22 @@ export const make = <Key, E, Reason = never>(options: {
const fork = yield* FiberSet.makeRuntime<never, void, never>()
const loop = (key: Key, execution: Execution<E, Reason>, force: boolean): Effect.Effect<void, E> =>
Effect.suspend(() => options.drain(key, force)).pipe(
Effect.suspend(() => options.drain(key, force, execution.scope)).pipe(
Effect.flatMap(() =>
Effect.suspend(() => {
if (execution.stopping || !execution.pendingWake) return Effect.void
execution.pendingWake = false
if (execution.stopping || execution.pendingWake === undefined) return Effect.void
execution.scope = execution.pendingWake
execution.pendingWake = undefined
// Trampoline so drains that complete synchronously cannot grow the stack.
return Effect.yieldNow.pipe(Effect.andThen(loop(key, execution, false)))
}),
),
)
const start = (key: Key, force: boolean) => {
const start = (key: Key, force: boolean, scope: Promotable) => {
const execution: Execution<E, Reason> = {
done: Deferred.makeUnsafe<void, E>(),
pendingWake: false,
scope,
stopping: false,
}
executions.set(key, execution)
@@ -98,7 +104,7 @@ export const make = <Key, E, Reason = never>(options: {
// A doorbell that survives the execution loop (rung after the loop decided to end, or
// during failure or interruption cleanup) starts a fresh execution for the remaining work.
const settle = (key: Key, execution: Execution<E, Reason>, exit: Exit.Exit<void, E>) => {
if (execution.pendingWake) start(key, false)
if (execution.pendingWake) start(key, false, execution.pendingWake)
else executions.delete(key)
Deferred.doneUnsafe(execution.done, exit)
}
@@ -111,17 +117,24 @@ export const make = <Key, E, Reason = never>(options: {
if (execution.stopping) return Deferred.await(execution.done).pipe(Effect.andThen(run(key)))
return Deferred.await(execution.done)
}
return Deferred.await(start(key, true).done)
return Deferred.await(start(key, true, "input").done)
})
const wake = (key: Key) =>
const wake = (key: Key, scope: Promotable = "input") =>
Effect.sync(() => {
const execution = executions.get(key)
if (execution !== undefined) {
execution.pendingWake = true
// Coalesced wakes keep the widest scope: "input" subsumes "steer".
execution.pendingWake = execution.pendingWake === "input" ? "input" : scope
return
}
start(key, false)
start(key, false, scope)
})
const wakeActive = (key: Key) =>
Effect.suspend(() => {
const execution = executions.get(key)
return execution ? wake(key, execution.scope) : Effect.void
})
const interrupt = (key: Key, reason?: Reason): Effect.Effect<void> =>
@@ -129,7 +142,9 @@ export const make = <Key, E, Reason = never>(options: {
const execution = executions.get(key)
if (execution?.owner === undefined || execution.stopping) return Effect.void
execution.stopping = true
execution.pendingWake = false
// Wakes recorded so far belong to the interrupted intent; the interrupt claims them.
// Wakes arriving during cleanup are new admissions and restart normally at settle.
execution.pendingWake = undefined
execution.interruptionReason = reason
return Fiber.interrupt(execution.owner)
})
@@ -143,5 +158,5 @@ export const make = <Key, E, Reason = never>(options: {
return Deferred.await(execution.done).pipe(Effect.exit, Effect.andThen(awaitIdle(key)))
})
return { active: Effect.sync(() => new Set(executions.keys())), run, wake, interrupt, awaitIdle }
return { active: Effect.sync(() => new Set(executions.keys())), run, wake, wakeActive, interrupt, awaitIdle }
})
@@ -3,6 +3,7 @@ export * as SessionRunner from "./index.js"
import type { AIError } from "@opencode-ai/ai"
import { Context, Effect } from "effect"
import { SessionSchema } from "../schema.js"
import type { Promotable } from "../inbox.js"
import type { AgentNotFoundError, MessageDecodeError, StepFailedError, UserInterruptedError } from "../error.js"
import { SessionRunnerModel } from "./model.js"
import type { Instructions } from "../../instructions/index.js"
@@ -29,6 +30,8 @@ export interface Interface {
readonly sessionID: SessionSchema.ID
readonly force: boolean
readonly continuation?: Continuation
/** "steer" settles the active intent without promoting queued next-turn work. */
readonly promotable?: Promotable
}) => Effect.Effect<DrainResult, RunError>
}
+16 -14
View File
@@ -128,22 +128,25 @@ const layer = Layer.effect(
readonly sessionID: SessionSchema.ID
readonly force: boolean
readonly continuation?: Continuation
readonly promotable?: SessionInbox.Promotable
}) {
let force = input.force
let continuation = input.continuation
if (!force && !continuation && !(yield* SessionInbox.has(db, input.sessionID, "any")))
const promotable = input.promotable ?? "input"
if (!force && !continuation && !(yield* SessionInbox.has(db, input.sessionID, promotable)))
return { type: "complete" as const }
yield* settleStaleToolCalls(input.sessionID)
while (true) {
if (yield* runPendingCompaction(input.sessionID)) {
if (yield* runPendingCompaction(input.sessionID, promotable)) {
force = false
continue
}
if (yield* runPendingMove(input.sessionID, "input")) return { type: "moved" as const }
if (!force && !continuation && !(yield* SessionInbox.has(db, input.sessionID, "input")))
if (yield* runPendingMove(input.sessionID, promotable)) return { type: "moved" as const }
if (!force && !continuation && !(yield* SessionInbox.has(db, input.sessionID, promotable)))
return { type: "complete" as const }
const result = yield* runSteps(input.sessionID, continuation)
const result = yield* runSteps(input.sessionID, continuation, promotable)
if (result.type === "moved") return result
if (promotable === "steer") return { type: "complete" as const }
force = false
continuation = undefined
}
@@ -155,14 +158,15 @@ const layer = Layer.effect(
*/
const runSteps = Effect.fn("SessionRunner.runSteps")(function* (
sessionID: SessionSchema.ID,
continuation?: Continuation,
continuation: Continuation | undefined,
drainPromotable: SessionInbox.Promotable,
) {
// Fresh work may promote queued input; later steps absorb steers only.
let promotable: SessionInbox.Promotable = continuation ? "steer" : "input"
// Fresh work may promote queued input; resumed turns and later steps absorb steers only.
let promotable: SessionInbox.Promotable = continuation ? "steer" : drainPromotable
let step = continuation?.step ?? 1
let next = continuation
while (true) {
if (yield* runPendingCompaction(sessionID)) continue
if (yield* runPendingCompaction(sessionID, "steer")) continue
if (yield* runPendingMove(sessionID, "steer")) return { type: "moved" as const, continuation: next }
const result = yield* runStep(sessionID, promotable, step)
next = result.needsContinuation ? { step: result.step + 1 } : undefined
@@ -515,14 +519,14 @@ const layer = Layer.effect(
/** Executes a previously admitted manual compaction request, if one is pending. */
const runPendingCompaction = Effect.fn("SessionRunner.runPendingCompaction")(function* (
sessionID: SessionSchema.ID,
promotable: SessionInbox.Promotable,
) {
return yield* Effect.uninterruptibleMask((restore) =>
Effect.gen(function* () {
const pending = yield* SessionInbox.serialized(
sessionID,
Effect.gen(function* () {
const selected =
(yield* SessionInbox.nextSteer(db, sessionID)) ?? (yield* SessionInbox.nextQueued(db, sessionID))
const selected = yield* SessionInbox.nextPromotable(db, sessionID, promotable)
if (selected?.type !== "compaction") return
yield* bus.publishAll([
[SessionEvent.InboxDelivered, { sessionID, inboxID: selected.id }],
@@ -564,9 +568,7 @@ const layer = Layer.effect(
return yield* SessionInbox.serialized(
sessionID,
Effect.gen(function* () {
const pending =
(yield* SessionInbox.nextSteer(db, sessionID)) ??
(promotable === "input" ? yield* SessionInbox.nextQueued(db, sessionID) : undefined)
const pending = yield* SessionInbox.nextPromotable(db, sessionID, promotable)
if (pending?.type !== "move") return false
yield* modelTransport.close(sessionID)
yield* bus.publishAll([
+110 -1
View File
@@ -14,8 +14,10 @@ import { SessionExecution } from "@opencode-ai/core/session/execution"
import { SessionRestart } from "@opencode-ai/core/session/execution/restart"
import { UserInterruptedError } from "@opencode-ai/core/session/error"
import { SessionEvent } from "@opencode-ai/core/session/event"
import { SessionInbox } from "@opencode-ai/core/session/inbox"
import { SessionMessage } from "@opencode-ai/core/session/message"
import { SessionRunner } from "@opencode-ai/core/session/runner/index"
import { SessionTable } from "@opencode-ai/core/session/sql"
import { SessionInboxTable, SessionTable } from "@opencode-ai/core/session/sql"
import { SessionStore } from "@opencode-ai/core/session/store"
import { Context, Deferred, Effect, Exit, Fiber, Layer, LayerMap, Scope } from "effect"
import { eq } from "drizzle-orm"
@@ -290,6 +292,113 @@ describe("SessionExecution lifecycle", () => {
)
})
describe("SessionExecution interrupt continuation", () => {
it.effect("resumes only steering input after an interrupt with continue", () =>
Effect.gen(function* () {
const database = yield* Database.Service
const sessionID = Session.ID.make("ses_continue_steer")
yield* seedSessions(database, [sessionID])
yield* seedInbox(database, sessionID, ["steer", "queue"])
const draining = yield* Deferred.make<void>()
const drains: Array<{ force: boolean; promotable?: SessionInbox.Promotable }> = []
const scope = yield* Scope.make()
yield* Effect.addFinalizer(() => Scope.close(scope, Exit.void))
const context = yield* buildExecution(scope, (input) =>
Effect.suspend(() => {
drains.push({ force: input.force, promotable: input.promotable })
if (drains.length > 1) return Effect.void
return Deferred.succeed(draining, undefined).pipe(Effect.andThen(Effect.never))
}),
)
const execution = Context.get(context, SessionExecution.Service)
yield* execution.resume(sessionID).pipe(Effect.forkScoped)
yield* Deferred.await(draining)
yield* execution.interrupt(sessionID, { continue: true })
yield* execution.awaitIdle(sessionID)
// The successor drain is steer-scoped: queued next-turn work stays parked.
expect(drains).toEqual([
{ force: true, promotable: "input" },
{ force: false, promotable: "steer" },
])
}),
)
it.effect("stays parked after an interrupt with continue when only queued work remains", () =>
Effect.gen(function* () {
const database = yield* Database.Service
const sessionID = Session.ID.make("ses_continue_parked")
yield* seedSessions(database, [sessionID])
yield* seedInbox(database, sessionID, ["queue"])
const draining = yield* Deferred.make<void>()
const drains: Array<SessionInbox.Promotable | undefined> = []
const scope = yield* Scope.make()
yield* Effect.addFinalizer(() => Scope.close(scope, Exit.void))
const context = yield* buildExecution(scope, (input) =>
Effect.suspend(() => {
drains.push(input.promotable)
return Deferred.succeed(draining, undefined).pipe(Effect.andThen(Effect.never))
}),
)
const execution = Context.get(context, SessionExecution.Service)
yield* execution.resume(sessionID).pipe(Effect.forkScoped)
yield* Deferred.await(draining)
yield* execution.interrupt(sessionID, { continue: true })
yield* execution.awaitIdle(sessionID)
expect(drains).toEqual(["input"])
expect(yield* execution.active).toEqual(new Set())
}),
)
it.effect("an idle interrupt with continue resumes pending steers", () =>
Effect.gen(function* () {
const database = yield* Database.Service
const sessionID = Session.ID.make("ses_continue_idle")
yield* seedSessions(database, [sessionID])
yield* seedInbox(database, sessionID, ["steer"])
const drains: Array<{ force: boolean; promotable?: SessionInbox.Promotable }> = []
const scope = yield* Scope.make()
yield* Effect.addFinalizer(() => Scope.close(scope, Exit.void))
const context = yield* buildExecution(scope, (input) =>
Effect.sync(() => void drains.push({ force: input.force, promotable: input.promotable })),
)
const execution = Context.get(context, SessionExecution.Service)
yield* execution.interrupt(sessionID, { continue: true })
yield* execution.awaitIdle(sessionID)
expect(drains).toEqual([{ force: false, promotable: "steer" }])
}),
)
})
function seedInbox(
database: Database.Service["Service"],
sessionID: Session.ID,
deliveries: ReadonlyArray<SessionInbox.Delivery>,
) {
return database.db
.insert(SessionInboxTable)
.values(
deliveries.map((delivery, index) => ({
id: SessionMessage.ID.create(),
session_id: sessionID,
type: "compaction" as const,
payload: {},
delivery,
enqueued_seq: index + 1,
})),
)
.run()
.pipe(Effect.orDie)
}
function seedSessions(
database: Database.Service["Service"],
sessionIDs: ReadonlyArray<Session.ID>,
+7 -17
View File
@@ -31,6 +31,7 @@ import { testEffect } from "./lib/effect"
const executionCalls: Session.ID[] = []
const interruptCalls: Session.ID[] = []
const interruptContinuations: Array<boolean | undefined> = []
const wakeCalls: Session.ID[] = []
const activeSessions = new Set<Session.ID>()
const execution = Layer.succeed(
@@ -41,14 +42,16 @@ const execution = Layer.succeed(
Effect.sync(() => {
executionCalls.push(sessionID)
}),
interrupt: (sessionID) =>
interrupt: (sessionID, options) =>
Effect.sync(() => {
interruptCalls.push(sessionID)
interruptContinuations.push(options?.continue)
}),
wake: (sessionID) =>
Effect.sync(() => {
wakeCalls.push(sessionID)
}),
wakeActive: () => Effect.void,
awaitIdle: () => Effect.void,
}),
)
@@ -177,31 +180,18 @@ describe("Session.prompt", () => {
}),
)
it.effect("continues after interruption when pending work remains", () =>
Effect.gen(function* () {
yield* setup
const session = yield* Session.Service
yield* session.synthetic({ sessionID, text: "Continue after interrupt", resume: false })
interruptCalls.length = 0
wakeCalls.length = 0
yield* session.interrupt(sessionID, { continue: true })
expect(interruptCalls).toEqual([sessionID])
expect(wakeCalls).toEqual([sessionID])
}),
)
it.effect("does not continue after interruption without pending work", () =>
it.effect("forwards interrupt continuation policy", () =>
Effect.gen(function* () {
yield* setup
const session = yield* Session.Service
interruptCalls.length = 0
interruptContinuations.length = 0
wakeCalls.length = 0
yield* session.interrupt(sessionID, { continue: true })
expect(interruptCalls).toEqual([sessionID])
expect(interruptContinuations).toEqual([true])
expect(wakeCalls).toEqual([])
}),
)
@@ -1,5 +1,6 @@
import { describe, expect } from "bun:test"
import { Cause, Deferred, Effect, Exit, Fiber, Layer } from "effect"
import { SessionInbox } from "@opencode-ai/core/session/inbox"
import { SessionRunCoordinator } from "@opencode-ai/core/session/run-coordinator"
import { testEffect } from "./lib/effect"
@@ -269,6 +270,28 @@ describe("SessionRunCoordinator", () => {
),
)
it.effect("a settlement-window wake starts a fresh execution with its own scope", () =>
Effect.scoped(
Effect.gen(function* () {
const settling = yield* Deferred.make<void>()
const release = yield* Deferred.make<void>()
const scopes: SessionInbox.Promotable[] = []
const coordinator = yield* SessionRunCoordinator.make({
drain: (_key, _force, scope) => Effect.sync(() => scopes.push(scope)),
settled: () => Deferred.succeed(settling, undefined).pipe(Effect.andThen(Deferred.await(release))),
})
yield* coordinator.wake("session", "steer")
yield* Deferred.await(settling)
yield* coordinator.wake("session", "input")
yield* Deferred.succeed(release, undefined)
yield* coordinator.awaitIdle("session")
expect(scopes).toEqual(["steer", "input"])
}),
),
)
it.effect("interrupts active execution and clears its pending wake", () =>
Effect.scoped(
Effect.gen(function* () {
@@ -342,6 +365,126 @@ describe("SessionRunCoordinator", () => {
),
)
it.effect("coalesces drain scopes with input taking precedence", () =>
Effect.scoped(
Effect.gen(function* () {
const firstStarted = yield* Deferred.make<void>()
const release = yield* Deferred.make<void>()
const scopes: SessionInbox.Promotable[] = []
const coordinator = yield* SessionRunCoordinator.make({
drain: (_key, _force, scope) =>
Effect.gen(function* () {
scopes.push(scope)
if (scopes.length !== 1) return
yield* Deferred.succeed(firstStarted, undefined)
yield* Deferred.await(release)
}),
})
yield* coordinator.wake("session", "steer")
yield* Deferred.await(firstStarted)
yield* coordinator.wake("session", "steer")
yield* coordinator.wake("session", "input")
yield* Deferred.succeed(release, undefined)
yield* coordinator.awaitIdle("session")
expect(scopes).toEqual(["steer", "input"])
}),
),
)
it.effect("does not carry a completed input scope into a steer drain", () =>
Effect.scoped(
Effect.gen(function* () {
const firstStarted = yield* Deferred.make<void>()
const release = yield* Deferred.make<void>()
const scopes: SessionInbox.Promotable[] = []
const coordinator = yield* SessionRunCoordinator.make({
drain: (_key, _force, scope) =>
Effect.gen(function* () {
scopes.push(scope)
if (scopes.length !== 1) return
yield* Deferred.succeed(firstStarted, undefined)
yield* Deferred.await(release)
}),
})
yield* coordinator.wake("session", "input")
yield* Deferred.await(firstStarted)
yield* coordinator.wake("session", "steer")
yield* Deferred.succeed(release, undefined)
yield* coordinator.awaitIdle("session")
expect(scopes).toEqual(["input", "steer"])
}),
),
)
it.effect("an active wake inherits scope without starting idle work", () =>
Effect.scoped(
Effect.gen(function* () {
const firstStarted = yield* Deferred.make<void>()
const release = yield* Deferred.make<void>()
const scopes: SessionInbox.Promotable[] = []
const coordinator = yield* SessionRunCoordinator.make({
drain: (_key, _force, scope) =>
Effect.gen(function* () {
scopes.push(scope)
if (scopes.length !== 1) return
yield* Deferred.succeed(firstStarted, undefined)
yield* Deferred.await(release)
}),
})
yield* coordinator.wakeActive("session")
yield* coordinator.wake("session", "steer")
yield* Deferred.await(firstStarted)
yield* coordinator.wakeActive("session")
yield* Deferred.succeed(release, undefined)
yield* coordinator.awaitIdle("session")
expect(scopes).toEqual(["steer", "steer"])
}),
),
)
it.effect("a cleanup-era wake starts a successor with its own scope", () =>
Effect.scoped(
Effect.gen(function* () {
const firstStarted = yield* Deferred.make<void>()
const cleanupStarted = yield* Deferred.make<void>()
const cleanupGate = yield* Deferred.make<void>()
const scopes: SessionInbox.Promotable[] = []
const coordinator = yield* SessionRunCoordinator.make({
drain: (_key, _force, scope) =>
Effect.gen(function* () {
scopes.push(scope)
if (scopes.length !== 1) return
yield* Deferred.succeed(firstStarted, undefined)
yield* Effect.never.pipe(
Effect.onInterrupt(() =>
Deferred.succeed(cleanupStarted, undefined).pipe(Effect.andThen(Deferred.await(cleanupGate))),
),
)
}),
})
yield* coordinator.wake("session", "input")
yield* Deferred.await(firstStarted)
const interrupt = yield* coordinator.interrupt("session").pipe(Effect.forkChild)
yield* Deferred.await(cleanupStarted)
// A new admission during cancellation restarts normally: interruption only
// claims the wakes recorded before it.
yield* coordinator.wake("session", "input")
yield* Deferred.succeed(cleanupGate, undefined)
yield* Fiber.join(interrupt)
yield* coordinator.awaitIdle("session")
expect(scopes).toEqual(["input", "input"])
}),
),
)
it.effect("starts a resume registered during interruption cleanup", () =>
Effect.scoped(
Effect.gen(function* () {
@@ -126,7 +126,8 @@ const execution = (llmClient: Layer.Layer<typeof LLMClient.Service>) =>
active: coordinator.active,
resume: coordinator.run,
wake: coordinator.wake,
interrupt: coordinator.interrupt,
wakeActive: coordinator.wakeActive,
interrupt: (sessionID) => coordinator.interrupt(sessionID),
awaitIdle: coordinator.awaitIdle,
})
}),
+58 -1
View File
@@ -413,7 +413,8 @@ const execution = Layer.effect(
active: coordinator.active,
resume: coordinator.run,
wake: coordinator.wake,
interrupt: coordinator.interrupt,
wakeActive: coordinator.wakeActive,
interrupt: (sessionID) => coordinator.interrupt(sessionID),
awaitIdle: coordinator.awaitIdle,
})
}),
@@ -1383,6 +1384,44 @@ describe("SessionRunnerLLM", () => {
}),
)
it.effect("keeps queued input parked across a mid-turn move", () =>
Effect.gen(function* () {
const session = yield* setup
const bus = yield* Bus.Service
const { db } = yield* Database.Service
yield* admit(session, "Echo before moving")
yield* TestLLM.push(
TestLLM.tool("call-move", "echo", { text: "moving" }),
TestLLM.text("Done", "text-after-move"),
TestLLM.text("Handled queue", "text-after-queue"),
)
const tools = yield* blockTools()
const run = yield* session.resume(sessionID).pipe(Effect.forkChild)
yield* tools.started
yield* session.prompt({ sessionID, text: "Queued for later", delivery: "queue", resume: false })
yield* SessionInbox.admit(db, bus, {
id: SessionMessage.ID.create(),
sessionID,
item: {
type: "move",
payload: {
location: Location.Ref.make({ directory: AbsolutePath.make("/project") }),
projectID: Project.ID.global,
},
delivery: "steer",
},
})
yield* tools.release
yield* Fiber.join(run)
// The resumed turn absorbs steers only; queued input waits for the turn to end.
expect(requests).toHaveLength(3)
expect(userTexts(requests[1])).not.toContain("Queued for later")
expect(userTexts(requests[2])).toContain("Queued for later")
}),
)
it.effect("seeds a fork with the parent's newest instruction values", () =>
Effect.gen(function* () {
const session = yield* setup
@@ -3088,6 +3127,24 @@ describe("SessionRunnerLLM", () => {
}),
)
it.effect("stops a steer-scoped drain before queued input", () =>
Effect.gen(function* () {
const session = yield* setup
const { db } = yield* Database.Service
yield* session.prompt({ sessionID, text: "Queue for later", delivery: "queue", resume: false })
yield* session.prompt({ sessionID, text: "Steer now", resume: false })
yield* TestLLM.push(TestLLM.stop())
const runner = yield* SessionRunner.Service
yield* runner.drain({ sessionID, force: false, promotable: "steer" })
expect(requests).toHaveLength(1)
expect(userTexts(requests[0])).toEqual(["Steer now"])
expect(yield* SessionInbox.has(db, sessionID, "steer")).toBe(false)
expect(yield* SessionInbox.has(db, sessionID, "queue")).toBe(true)
}),
)
it.effect("promotes queued input after steering continuation ends", () =>
Effect.gen(function* () {
const session = yield* setup
+1
View File
@@ -115,6 +115,7 @@ const executionNode = makeGlobalNode({
active: Effect.succeed(new Set()),
resume: complete,
wake: () => Effect.void,
wakeActive: () => Effect.void,
interrupt: () => Effect.void,
awaitIdle: (id) => complete(id).pipe(Effect.exit, Effect.asVoid),
})
+1
View File
@@ -86,6 +86,7 @@ const executionNode = makeGlobalNode({
active: Effect.succeed(new Set()),
resume: complete,
wake: () => Effect.void,
wakeActive: () => Effect.void,
interrupt: () => Effect.void,
awaitIdle: (sessionID) => complete(sessionID).pipe(Effect.exit, Effect.asVoid),
})
+1 -1
View File
@@ -3964,7 +3964,7 @@
}
}
},
"description": "Interrupt active execution owned by this OpenCode process. Idle interruption is a no-op. When continue=true, execution resumes if durable inbox work remains after interruption.",
"description": "Interrupt active execution owned by this OpenCode process. Idle interruption is a no-op. When continue=true, execution resumes pending steering input while queued work remains parked.",
"summary": "Interrupt session execution"
}
},
+1 -1
View File
@@ -660,7 +660,7 @@ export const makeSessionGroup = <I extends HttpApiMiddleware.AnyId, S>(sessionLo
identifier: "v2.session.interrupt",
summary: "Interrupt session execution",
description:
"Interrupt active execution owned by this OpenCode process. Idle interruption is a no-op. When continue=true, execution resumes if durable inbox work remains after interruption.",
"Interrupt active execution owned by this OpenCode process. Idle interruption is a no-op. When continue=true, execution resumes pending steering input while queued work remains parked.",
}),
),
)
@@ -1,6 +1,7 @@
import { createMemo, createResource, createSignal, onMount, Show } from "solid-js"
import path from "path"
import type { SessionInfo } from "@opencode-ai/client"
import { Project } from "@opencode-ai/schema/project"
import { TextAttributes } from "@opentui/core"
import type { RGBA } from "@opentui/core"
import { useDialog } from "../ui/dialog"
@@ -54,10 +55,12 @@ export function DialogSessionList() {
const response = await client.api.session.list({
...(allProjects
? {}
: {
project: current.project.id,
subpath: path.relative(current.project.directory, current.directory).replaceAll("\\", "/"),
}),
: current.project.id === Project.ID.global
? { directory: current.directory }
: {
project: current.project.id,
subpath: path.relative(current.project.directory, current.directory).replaceAll("\\", "/"),
}),
...(query ? { search: query } : {}),
limit: 50,
order: "desc",
+17 -2
View File
@@ -71,6 +71,7 @@ import { DialogImagePreview } from "../dialog-image-preview"
import { useDirectoryRecents } from "../../prompt/directory-recents"
import { directoryRecentValue } from "../../prompt/directory-completion"
import { useWorkingDirectoryActions } from "../../ui/working-directory-actions"
import { truncateFilePath } from "../../ui/file-path"
export type PromptProps = {
sessionID?: string
@@ -1555,6 +1556,12 @@ export function Prompt(props: PromptProps) {
const branch = data.location.vcs.info(location)?.branch.current
return branch ? `${directory}:${branch}` : directory
})
const [locationWidth, setLocationWidth] = createSignal(dimensions().width)
const locationLabelDisplay = createMemo(() => {
const label = locationLabel()
if (!label) return
return truncateFilePath(label, locationWidth())
})
const locationActions = useWorkingDirectoryActions({
directory: () => footerLocation()?.directory,
onMove: () => void move.open(),
@@ -1840,7 +1847,15 @@ export function Prompt(props: PromptProps) {
<box width="100%" flexDirection="row" justifyContent="space-between" gap={2}>
<Slot path="prompt.footer" input={footerInput()}>
<Slot path="prompt.footer.status" input={footerInput()}>
<box flexGrow={1} flexShrink={1} minWidth={0}>
<box
flexGrow={1}
flexShrink={1}
minWidth={0}
onSizeChange={function (this: BoxRenderable) {
const width = this.width
queueMicrotask(() => setLocationWidth(width))
}}
>
<Switch>
<Match when={status() === "running"}>
<box flexDirection="row" gap={1} flexGrow={1} justifyContent="flex-start">
@@ -1877,7 +1892,7 @@ export function Prompt(props: PromptProps) {
</box>
</Match>
<Match when={true}>
<Show when={!props.hint && locationLabel()} fallback={props.hint ?? <text />}>
<Show when={!props.hint && locationLabelDisplay()} fallback={props.hint ?? <text />}>
{(location) => (
<text
id="prompt.footer.location"
+4
View File
@@ -14,6 +14,10 @@ describe("truncateFilePath", () => {
expect(truncateFilePath(path, 19)).toBe("…/dialog-select.tsx")
})
test("preserves the working directory and branch suffix", () => {
expect(truncateFilePath("~/code/experiments/category-theory:main", 30)).toBe("…/experi…/category-theory:main")
})
test("uses remaining width for part of a long parent segment", () => {
const path = "/private/var/folders/run-17f048ec-dbb2-4b36-860c-98637bb51a8d/files"
expect(truncateFilePath(path, 40)).toBe("/…/run-17f048ec-dbb2-4b36-860c-98…/files")
+1 -1
View File
@@ -3964,7 +3964,7 @@
}
}
},
"description": "Interrupt active execution owned by this OpenCode process. Idle interruption is a no-op. When continue=true, execution resumes if durable inbox work remains after interruption.",
"description": "Interrupt active execution owned by this OpenCode process. Idle interruption is a no-op. When continue=true, execution resumes pending steering input while queued work remains parked.",
"summary": "Interrupt session execution"
}
},
+1 -1
View File
@@ -3964,7 +3964,7 @@
}
}
},
"description": "Interrupt active execution owned by this OpenCode process. Idle interruption is a no-op. When continue=true, execution resumes if durable inbox work remains after interruption.",
"description": "Interrupt active execution owned by this OpenCode process. Idle interruption is a no-op. When continue=true, execution resumes pending steering input while queued work remains parked.",
"summary": "Interrupt session execution"
}
},