Compare commits

..

1 Commits

Author SHA1 Message Date
Filip Hejmowski c71ebe46df feat(core): route subagent models by role 2026-08-15 17:03:57 +00:00
24 changed files with 315 additions and 449 deletions
-5
View File
@@ -1,5 +0,0 @@
---
"@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.
+90
View File
@@ -0,0 +1,90 @@
export * as ModelRouting from "./model-routing.js"
import { Model } from "./model.js"
import { Provider } from "./provider.js"
export const roles = ["fast", "smart", "vision", "long-context"] as const
export type Role = (typeof roles)[number]
export function resolve(selection: string, available: readonly Model.Info[]) {
if (!isRole(selection)) return exact(selection, available)
return select(selection, available)
}
export function select(role: Role, available: readonly Model.Info[]) {
const candidates = available.filter(
(model) =>
model.status === "active" &&
model.capabilities.tools &&
model.capabilities.input.includes("text") &&
model.capabilities.output.includes("text"),
)
const eligible =
role === "vision"
? candidates.filter((model) => model.capabilities.input.includes("image"))
: role === "fast"
? candidates.filter((model) => !SLOW_MODEL_RE.test(identity(model)))
: candidates
if (eligible.length === 0) return
const sorted = eligible.toSorted((a, b) => {
if (role === "fast") {
const tagged = Number(fast(b)) - Number(fast(a))
if (tagged !== 0) return tagged
const price = cost(a) - cost(b)
if (price !== 0) return price
}
if (role === "smart" || role === "vision") {
const tagged = Number(smart(b)) - Number(smart(a))
if (tagged !== 0) return tagged
}
if (role === "long-context") {
const context = b.limit.context - a.limit.context
if (context !== 0) return context
}
const released = b.time.released - a.time.released
if (released !== 0) return released
return `${a.providerID}/${a.id}`.localeCompare(`${b.providerID}/${b.id}`)
})
const selected = sorted[0]
return Model.Ref.make({ providerID: selected.providerID, id: selected.id })
}
function isRole(selection: string): selection is Role {
return roles.includes(selection as Role)
}
function exact(selection: string, available: readonly Model.Info[]) {
const providerEnd = selection.indexOf("/")
if (providerEnd <= 0) return
const variantStart = selection.indexOf("#", providerEnd + 1)
const providerID = Provider.ID.make(selection.slice(0, providerEnd))
const id = Model.ID.make(selection.slice(providerEnd + 1, variantStart === -1 ? undefined : variantStart))
const variant = variantStart === -1 ? undefined : Model.VariantID.make(selection.slice(variantStart + 1))
if (!id || !providerID || (variantStart !== -1 && !variant)) return
const model = available.find((item) => item.providerID === providerID && item.id === id)
if (!model) return
if (variant && !model.variants.some((item) => item.id === variant)) return
return Model.Ref.make({ providerID, id, variant })
}
function cost(model: Model.Info) {
const price = model.cost[0]
return price ? price.input + price.output : Number.MAX_SAFE_INTEGER
}
function fast(model: Model.Info) {
return FAST_MODEL_RE.test(identity(model))
}
function smart(model: Model.Info) {
return SMART_MODEL_RE.test(identity(model))
}
function identity(model: Model.Info) {
return `${model.id} ${model.family ?? ""} ${model.name}`.toLowerCase()
}
const FAST_MODEL_RE = /\b(nano|flash|lite|mini|small|fast)\b/
const SLOW_MODEL_RE = /\b(haiku)\b/
const SMART_MODEL_RE = /\b(opus|pro|max|ultra|reasoner|reasoning)\b|\b(gpt-5|grok-4|deepseek-v4|kimi-k2)\b/
+7 -5
View File
@@ -789,10 +789,7 @@ const layer = Layer.effect(
return false return false
}), }),
) )
if (recovered) { if (recovered) return
yield* execution.wakeActive(input.sessionID)
return
}
yield* execution.wake(input.sessionID) yield* execution.wake(input.sessionID)
}), }),
compact: Effect.fn("Session.compact")(function* (input) { compact: Effect.fn("Session.compact")(function* (input) {
@@ -876,7 +873,12 @@ const layer = Layer.effect(
), ),
), ),
interrupt: Effect.fn("Session.interrupt")((sessionID, options) => interrupt: Effect.fn("Session.interrupt")((sessionID, options) =>
Effect.uninterruptible(execution.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)
}),
),
), ),
revert: { revert: {
stage: Effect.fn("Session.revert.stage")(function* (input) { stage: Effect.fn("Session.revert.stage")(function* (input) {
+11 -22
View File
@@ -1,8 +1,7 @@
export * as SessionExecution from "./execution.js" export * as SessionExecution from "./execution.js"
import { Cause, Context, Effect, Exit, Layer } from "effect" import { Cause, Context, Effect, Exit, Layer, Stream } from "effect"
import { Bus } from "../bus.js" import { Bus } from "../bus.js"
import { Database } from "../database/database.js"
import { LocationServiceMap } from "../location-service-map.js" import { LocationServiceMap } from "../location-service-map.js"
import { makeGlobalNode } from "@opencode-ai/util/effect/app-node" import { makeGlobalNode } from "@opencode-ai/util/effect/app-node"
import { SessionEvent } from "./event.js" import { SessionEvent } from "./event.js"
@@ -12,7 +11,6 @@ import { SessionSchema } from "./schema.js"
import { SessionStore } from "./store.js" import { SessionStore } from "./store.js"
import { toSessionError } from "./to-session-error.js" import { toSessionError } from "./to-session-error.js"
import { UserInterruptedError } from "./error.js" import { UserInterruptedError } from "./error.js"
import { SessionInbox } from "./inbox.js"
export interface Interface { export interface Interface {
/** Snapshots active execution owned by this process. */ /** Snapshots active execution owned by this process. */
@@ -21,10 +19,8 @@ export interface Interface {
readonly resume: (sessionID: SessionSchema.ID) => Effect.Effect<void, SessionRunner.RunError> readonly resume: (sessionID: SessionSchema.ID) => Effect.Effect<void, SessionRunner.RunError>
/** Registers newly recorded work. Repeated wakeups may coalesce. */ /** Registers newly recorded work. Repeated wakeups may coalesce. */
readonly wake: (sessionID: SessionSchema.ID) => Effect.Effect<void> 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. */ /** Interrupt active work owned by this process. Idle interruption is a no-op. */
readonly interrupt: (sessionID: SessionSchema.ID, options?: { readonly continue?: boolean }) => Effect.Effect<void> readonly interrupt: (sessionID: SessionSchema.ID) => Effect.Effect<void>
/** Resolves once this process owns no active execution for the Session. Returns immediately when idle and never starts work. */ /** 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> readonly awaitIdle: (sessionID: SessionSchema.ID) => Effect.Effect<void>
} }
@@ -49,7 +45,6 @@ export const layer = Layer.effect(
const store = yield* SessionStore.Service const store = yield* SessionStore.Service
const locations = yield* LocationServiceMap.Service const locations = yield* LocationServiceMap.Service
const bus = yield* Bus.Service const bus = yield* Bus.Service
const db = (yield* Database.Service).db
const reportLifecycle = <A>(sessionID: SessionSchema.ID, effect: Effect.Effect<A>) => const reportLifecycle = <A>(sessionID: SessionSchema.ID, effect: Effect.Effect<A>) =>
effect.pipe( effect.pipe(
Effect.tapCause((cause) => Effect.tapCause((cause) =>
@@ -76,13 +71,12 @@ export const layer = Layer.effect(
sessionID: SessionSchema.ID, sessionID: SessionSchema.ID,
force: boolean, force: boolean,
continuation?: SessionRunner.Continuation, continuation?: SessionRunner.Continuation,
promotable: SessionInbox.Promotable = "input",
): Effect.Effect<void, SessionRunner.RunError> { ): Effect.Effect<void, SessionRunner.RunError> {
return Effect.gen(function* () { return Effect.gen(function* () {
const session = yield* store.get(sessionID) const session = yield* store.get(sessionID)
if (!session) return yield* Effect.die(new Error(`Session not found: ${sessionID}`)) if (!session) return yield* Effect.die(new Error(`Session not found: ${sessionID}`))
const result = yield* SessionRunner.Service.use((runner) => const result = yield* SessionRunner.Service.use((runner) =>
runner.drain({ sessionID, force, continuation, promotable }), runner.drain({ sessionID, force, continuation }),
).pipe( ).pipe(
Effect.provide(locations.get(session.location)), Effect.provide(locations.get(session.location)),
Effect.tapCause((cause) => Effect.tapCause((cause) =>
@@ -92,7 +86,7 @@ export const layer = Layer.effect(
), ),
) )
if (result.type === "complete") return if (result.type === "complete") return
return yield* drain(sessionID, false, result.continuation, promotable) return yield* drain(sessionID, false, result.continuation)
}) })
} }
const coordinator = yield* SessionRunCoordinator.make<SessionSchema.ID, SessionRunner.RunError, InterruptReason>({ const coordinator = yield* SessionRunCoordinator.make<SessionSchema.ID, SessionRunner.RunError, InterruptReason>({
@@ -101,7 +95,7 @@ export const layer = Layer.effect(
sessionID, sessionID,
bus.publish(SessionEvent.Execution.Started, { sessionID }, claimOnCommit(sessionID)), bus.publish(SessionEvent.Execution.Started, { sessionID }, claimOnCommit(sessionID)),
), ),
drain: (sessionID, force, promotable) => drain(sessionID, force, undefined, promotable), drain: (sessionID, force) => drain(sessionID, force),
// One terminal observation per busy period, covering every coalesced drain. // One terminal observation per busy period, covering every coalesced drain.
settled: (sessionID, exit, reason) => settled: (sessionID, exit, reason) =>
reportLifecycle( reportLifecycle(
@@ -133,20 +127,16 @@ export const layer = Layer.effect(
}), }),
), ),
}) })
yield* bus.subscribe(SessionEvent.Moved).pipe(
Stream.runForEach((event) => coordinator.wake(event.data.sessionID)),
Effect.forkScoped,
)
return Service.of({ return Service.of({
active: coordinator.active, active: coordinator.active,
interrupt: (sessionID, options) => interrupt: (sessionID) => coordinator.interrupt(sessionID, "user"),
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, resume: coordinator.run,
wake: coordinator.wake, wake: coordinator.wake,
wakeActive: coordinator.wakeActive,
awaitIdle: coordinator.awaitIdle, awaitIdle: coordinator.awaitIdle,
}) })
}), }),
@@ -155,7 +145,7 @@ export const layer = Layer.effect(
export const node = makeGlobalNode({ export const node = makeGlobalNode({
service: Service, service: Service,
layer, layer,
deps: [SessionStore.node, LocationServiceMap.node, Bus.node, Database.node], deps: [SessionStore.node, LocationServiceMap.node, Bus.node],
}) })
/** Low-level compatibility layer for callers that only need durable Session recording. */ /** Low-level compatibility layer for callers that only need durable Session recording. */
@@ -165,7 +155,6 @@ export const noopLayer = Layer.succeed(
active: Effect.succeed(new Set()), active: Effect.succeed(new Set()),
resume: () => Effect.void, resume: () => Effect.void,
wake: () => Effect.void, wake: () => Effect.void,
wakeActive: () => Effect.void,
interrupt: () => Effect.void, interrupt: () => Effect.void,
awaitIdle: () => Effect.void, awaitIdle: () => Effect.void,
}), }),
-8
View File
@@ -349,14 +349,6 @@ export const nextSteer = Effect.fn("SessionInbox.nextSteer")(function* (
return row ? fromRow(row) : undefined 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 * Which pending rows count: "any" counts every row, while "input" means any
* item in either delivery mode. * item in either delivery mode.
+18 -33
View File
@@ -1,7 +1,6 @@
export * as SessionRunCoordinator from "./run-coordinator.js" export * as SessionRunCoordinator from "./run-coordinator.js"
import { Deferred, Effect, Exit, Fiber, FiberSet, Scope } from "effect" 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. */ /** Serializes execution for each key while allowing different keys to run concurrently. */
export interface Coordinator<Key, E, Reason = never> { export interface Coordinator<Key, E, Reason = never> {
@@ -10,9 +9,7 @@ export interface Coordinator<Key, E, Reason = never> {
/** Starts an execution while idle, or joins the active execution and returns its exit. */ /** Starts an execution while idle, or joins the active execution and returns its exit. */
readonly run: (key: Key) => Effect.Effect<void, E> readonly run: (key: Key) => Effect.Effect<void, E>
/** Rings the doorbell: an idle key starts an execution; an active one drains again before settling. */ /** Rings the doorbell: an idle key starts an execution; an active one drains again before settling. */
readonly wake: (key: Key, scope?: Promotable) => Effect.Effect<void> readonly wake: (key: Key) => 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. */ /** Stops the active execution, clears its doorbell, and waits for cleanup. No-op when idle. */
readonly interrupt: (key: Key, reason?: Reason) => Effect.Effect<void> 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. */ /** Resolves once no execution is active for the key. Returns immediately when already idle and never starts work. */
@@ -22,16 +19,14 @@ export interface Coordinator<Key, E, Reason = never> {
/** /**
* One execution is a busy period for one key: one fiber that drains from the first wake * 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 * until the key would stay idle. `pendingWake` is the doorbell: work recorded during the
* execution rings it with the scope that work needs, and the execution loop drains again * execution rings it, and the execution loop drains again instead of ending. The doorbell
* instead of ending. The doorbell closes the gap between a drain's last eligibility check * closes the gap between a drain's last eligibility check and the idle transition, since
* and the idle transition, since those cannot be one atomic step. `done` resolves joiners * those cannot be one atomic step. `done` resolves joiners with this execution's exit.
* with this execution's exit.
*/ */
type Execution<E, Reason> = { type Execution<E, Reason> = {
readonly done: Deferred.Deferred<void, E> readonly done: Deferred.Deferred<void, E>
owner?: Fiber.Fiber<void> owner?: Fiber.Fiber<void>
scope: Promotable pendingWake: boolean
pendingWake?: Promotable
stopping: boolean stopping: boolean
interruptionReason?: Reason interruptionReason?: Reason
} }
@@ -48,7 +43,7 @@ type Execution<E, Reason> = {
* ``` * ```
*/ */
export const make = <Key, E, Reason = never>(options: { export const make = <Key, E, Reason = never>(options: {
readonly drain: (key: Key, force: boolean, scope: Promotable) => Effect.Effect<void, E> readonly drain: (key: Key, force: boolean) => Effect.Effect<void, E>
/** Runs once when a process-local busy period begins, before its first drain. */ /** Runs once when a process-local busy period begins, before its first drain. */
readonly started?: (key: Key) => Effect.Effect<void> readonly started?: (key: Key) => Effect.Effect<void>
/** /**
@@ -62,22 +57,21 @@ export const make = <Key, E, Reason = never>(options: {
const fork = yield* FiberSet.makeRuntime<never, void, never>() const fork = yield* FiberSet.makeRuntime<never, void, never>()
const loop = (key: Key, execution: Execution<E, Reason>, force: boolean): Effect.Effect<void, E> => const loop = (key: Key, execution: Execution<E, Reason>, force: boolean): Effect.Effect<void, E> =>
Effect.suspend(() => options.drain(key, force, execution.scope)).pipe( Effect.suspend(() => options.drain(key, force)).pipe(
Effect.flatMap(() => Effect.flatMap(() =>
Effect.suspend(() => { Effect.suspend(() => {
if (execution.stopping || execution.pendingWake === undefined) return Effect.void if (execution.stopping || !execution.pendingWake) return Effect.void
execution.scope = execution.pendingWake execution.pendingWake = false
execution.pendingWake = undefined
// Trampoline so drains that complete synchronously cannot grow the stack. // Trampoline so drains that complete synchronously cannot grow the stack.
return Effect.yieldNow.pipe(Effect.andThen(loop(key, execution, false))) return Effect.yieldNow.pipe(Effect.andThen(loop(key, execution, false)))
}), }),
), ),
) )
const start = (key: Key, force: boolean, scope: Promotable) => { const start = (key: Key, force: boolean) => {
const execution: Execution<E, Reason> = { const execution: Execution<E, Reason> = {
done: Deferred.makeUnsafe<void, E>(), done: Deferred.makeUnsafe<void, E>(),
scope, pendingWake: false,
stopping: false, stopping: false,
} }
executions.set(key, execution) executions.set(key, execution)
@@ -104,7 +98,7 @@ export const make = <Key, E, Reason = never>(options: {
// A doorbell that survives the execution loop (rung after the loop decided to end, or // 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. // 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>) => { const settle = (key: Key, execution: Execution<E, Reason>, exit: Exit.Exit<void, E>) => {
if (execution.pendingWake) start(key, false, execution.pendingWake) if (execution.pendingWake) start(key, false)
else executions.delete(key) else executions.delete(key)
Deferred.doneUnsafe(execution.done, exit) Deferred.doneUnsafe(execution.done, exit)
} }
@@ -117,24 +111,17 @@ export const make = <Key, E, Reason = never>(options: {
if (execution.stopping) return Deferred.await(execution.done).pipe(Effect.andThen(run(key))) if (execution.stopping) return Deferred.await(execution.done).pipe(Effect.andThen(run(key)))
return Deferred.await(execution.done) return Deferred.await(execution.done)
} }
return Deferred.await(start(key, true, "input").done) return Deferred.await(start(key, true).done)
}) })
const wake = (key: Key, scope: Promotable = "input") => const wake = (key: Key) =>
Effect.sync(() => { Effect.sync(() => {
const execution = executions.get(key) const execution = executions.get(key)
if (execution !== undefined) { if (execution !== undefined) {
// Coalesced wakes keep the widest scope: "input" subsumes "steer". execution.pendingWake = true
execution.pendingWake = execution.pendingWake === "input" ? "input" : scope
return return
} }
start(key, false, scope) start(key, false)
})
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> => const interrupt = (key: Key, reason?: Reason): Effect.Effect<void> =>
@@ -142,9 +129,7 @@ export const make = <Key, E, Reason = never>(options: {
const execution = executions.get(key) const execution = executions.get(key)
if (execution?.owner === undefined || execution.stopping) return Effect.void if (execution?.owner === undefined || execution.stopping) return Effect.void
execution.stopping = true execution.stopping = true
// Wakes recorded so far belong to the interrupted intent; the interrupt claims them. execution.pendingWake = false
// Wakes arriving during cleanup are new admissions and restart normally at settle.
execution.pendingWake = undefined
execution.interruptionReason = reason execution.interruptionReason = reason
return Fiber.interrupt(execution.owner) return Fiber.interrupt(execution.owner)
}) })
@@ -158,5 +143,5 @@ export const make = <Key, E, Reason = never>(options: {
return Deferred.await(execution.done).pipe(Effect.exit, Effect.andThen(awaitIdle(key))) return Deferred.await(execution.done).pipe(Effect.exit, Effect.andThen(awaitIdle(key)))
}) })
return { active: Effect.sync(() => new Set(executions.keys())), run, wake, wakeActive, interrupt, awaitIdle } return { active: Effect.sync(() => new Set(executions.keys())), run, wake, interrupt, awaitIdle }
}) })
@@ -3,7 +3,6 @@ export * as SessionRunner from "./index.js"
import type { AIError } from "@opencode-ai/ai" import type { AIError } from "@opencode-ai/ai"
import { Context, Effect } from "effect" import { Context, Effect } from "effect"
import { SessionSchema } from "../schema.js" import { SessionSchema } from "../schema.js"
import type { Promotable } from "../inbox.js"
import type { AgentNotFoundError, MessageDecodeError, StepFailedError, UserInterruptedError } from "../error.js" import type { AgentNotFoundError, MessageDecodeError, StepFailedError, UserInterruptedError } from "../error.js"
import { SessionRunnerModel } from "./model.js" import { SessionRunnerModel } from "./model.js"
import type { Instructions } from "../../instructions/index.js" import type { Instructions } from "../../instructions/index.js"
@@ -30,8 +29,6 @@ export interface Interface {
readonly sessionID: SessionSchema.ID readonly sessionID: SessionSchema.ID
readonly force: boolean readonly force: boolean
readonly continuation?: Continuation readonly continuation?: Continuation
/** "steer" settles the active intent without promoting queued next-turn work. */
readonly promotable?: Promotable
}) => Effect.Effect<DrainResult, RunError> }) => Effect.Effect<DrainResult, RunError>
} }
+14 -16
View File
@@ -128,25 +128,22 @@ const layer = Layer.effect(
readonly sessionID: SessionSchema.ID readonly sessionID: SessionSchema.ID
readonly force: boolean readonly force: boolean
readonly continuation?: Continuation readonly continuation?: Continuation
readonly promotable?: SessionInbox.Promotable
}) { }) {
let force = input.force let force = input.force
let continuation = input.continuation let continuation = input.continuation
const promotable = input.promotable ?? "input" if (!force && !continuation && !(yield* SessionInbox.has(db, input.sessionID, "any")))
if (!force && !continuation && !(yield* SessionInbox.has(db, input.sessionID, promotable)))
return { type: "complete" as const } return { type: "complete" as const }
yield* settleStaleToolCalls(input.sessionID) yield* settleStaleToolCalls(input.sessionID)
while (true) { while (true) {
if (yield* runPendingCompaction(input.sessionID, promotable)) { if (yield* runPendingCompaction(input.sessionID)) {
force = false force = false
continue continue
} }
if (yield* runPendingMove(input.sessionID, promotable)) return { type: "moved" as const } if (yield* runPendingMove(input.sessionID, "input")) return { type: "moved" as const }
if (!force && !continuation && !(yield* SessionInbox.has(db, input.sessionID, promotable))) if (!force && !continuation && !(yield* SessionInbox.has(db, input.sessionID, "input")))
return { type: "complete" as const } return { type: "complete" as const }
const result = yield* runSteps(input.sessionID, continuation, promotable) const result = yield* runSteps(input.sessionID, continuation)
if (result.type === "moved") return result if (result.type === "moved") return result
if (promotable === "steer") return { type: "complete" as const }
force = false force = false
continuation = undefined continuation = undefined
} }
@@ -158,15 +155,14 @@ const layer = Layer.effect(
*/ */
const runSteps = Effect.fn("SessionRunner.runSteps")(function* ( const runSteps = Effect.fn("SessionRunner.runSteps")(function* (
sessionID: SessionSchema.ID, sessionID: SessionSchema.ID,
continuation: Continuation | undefined, continuation?: Continuation,
drainPromotable: SessionInbox.Promotable,
) { ) {
// Fresh work may promote queued input; resumed turns and later steps absorb steers only. // Fresh work may promote queued input; later steps absorb steers only.
let promotable: SessionInbox.Promotable = continuation ? "steer" : drainPromotable let promotable: SessionInbox.Promotable = continuation ? "steer" : "input"
let step = continuation?.step ?? 1 let step = continuation?.step ?? 1
let next = continuation let next = continuation
while (true) { while (true) {
if (yield* runPendingCompaction(sessionID, "steer")) continue if (yield* runPendingCompaction(sessionID)) continue
if (yield* runPendingMove(sessionID, "steer")) return { type: "moved" as const, continuation: next } if (yield* runPendingMove(sessionID, "steer")) return { type: "moved" as const, continuation: next }
const result = yield* runStep(sessionID, promotable, step) const result = yield* runStep(sessionID, promotable, step)
next = result.needsContinuation ? { step: result.step + 1 } : undefined next = result.needsContinuation ? { step: result.step + 1 } : undefined
@@ -519,14 +515,14 @@ const layer = Layer.effect(
/** Executes a previously admitted manual compaction request, if one is pending. */ /** Executes a previously admitted manual compaction request, if one is pending. */
const runPendingCompaction = Effect.fn("SessionRunner.runPendingCompaction")(function* ( const runPendingCompaction = Effect.fn("SessionRunner.runPendingCompaction")(function* (
sessionID: SessionSchema.ID, sessionID: SessionSchema.ID,
promotable: SessionInbox.Promotable,
) { ) {
return yield* Effect.uninterruptibleMask((restore) => return yield* Effect.uninterruptibleMask((restore) =>
Effect.gen(function* () { Effect.gen(function* () {
const pending = yield* SessionInbox.serialized( const pending = yield* SessionInbox.serialized(
sessionID, sessionID,
Effect.gen(function* () { Effect.gen(function* () {
const selected = yield* SessionInbox.nextPromotable(db, sessionID, promotable) const selected =
(yield* SessionInbox.nextSteer(db, sessionID)) ?? (yield* SessionInbox.nextQueued(db, sessionID))
if (selected?.type !== "compaction") return if (selected?.type !== "compaction") return
yield* bus.publishAll([ yield* bus.publishAll([
[SessionEvent.InboxDelivered, { sessionID, inboxID: selected.id }], [SessionEvent.InboxDelivered, { sessionID, inboxID: selected.id }],
@@ -568,7 +564,9 @@ const layer = Layer.effect(
return yield* SessionInbox.serialized( return yield* SessionInbox.serialized(
sessionID, sessionID,
Effect.gen(function* () { Effect.gen(function* () {
const pending = yield* SessionInbox.nextPromotable(db, sessionID, promotable) const pending =
(yield* SessionInbox.nextSteer(db, sessionID)) ??
(promotable === "input" ? yield* SessionInbox.nextQueued(db, sessionID) : undefined)
if (pending?.type !== "move") return false if (pending?.type !== "move") return false
yield* modelTransport.close(sessionID) yield* modelTransport.close(sessionID)
yield* bus.publishAll([ yield* bus.publishAll([
+15 -2
View File
@@ -4,7 +4,9 @@ import { ToolFailure } from "@opencode-ai/ai"
import type { Context as PluginContext } from "@opencode-ai/plugin/effect/plugin" import type { Context as PluginContext } from "@opencode-ai/plugin/effect/plugin"
import { Effect, Schema, Scope } from "effect" import { Effect, Schema, Scope } from "effect"
import { Agent } from "../../agent.js" import { Agent } from "../../agent.js"
import { Catalog } from "../../catalog.js"
import { Config } from "../../config.js" import { Config } from "../../config.js"
import { ModelRouting } from "../../model-routing.js"
import { PluginRuntime } from "../../plugin/runtime.js" import { PluginRuntime } from "../../plugin/runtime.js"
import { Permission } from "../../permission.js" import { Permission } from "../../permission.js"
import { SessionSchema } from "../../session/schema.js" import { SessionSchema } from "../../session/schema.js"
@@ -23,6 +25,10 @@ export const Input = Schema.Struct({
agent: Schema.String.annotate({ description: "The type of specialized agent to use for this task" }), agent: Schema.String.annotate({ description: "The type of specialized agent to use for this task" }),
description: Schema.String.annotate({ description: "A short 3-5 word label for the task, displayed to the user" }), description: Schema.String.annotate({ description: "A short 3-5 word label for the task, displayed to the user" }),
prompt: Schema.String.annotate({ description: "The task for the subagent to perform" }), prompt: Schema.String.annotate({ description: "The task for the subagent to perform" }),
model: Schema.optionalKey(Schema.String).annotate({
description:
'Optional model route. Use "fast" for cheap bounded work, "smart" for difficult reasoning, "vision" for image input, or "long-context" for very large inputs. Pass an exact provider/model ID only when the user requests one. Omit this field to use the agent model or inherit the parent model.',
}),
background: Schema.optionalKey(Schema.Boolean).annotate({ background: Schema.optionalKey(Schema.Boolean).annotate({
description: description:
"Run the subagent in the background and return immediately. You will be notified when it completes. DO NOT sleep, poll, or proactively check on its progress.", "Run the subagent in the background and return immediately. You will be notified when it completes. DO NOT sleep, poll, or proactively check on its progress.",
@@ -47,6 +53,7 @@ export const Plugin = {
effect: Effect.fn("SubagentTool.Plugin")(function* (ctx: PluginContext) { effect: Effect.fn("SubagentTool.Plugin")(function* (ctx: PluginContext) {
const runtime = yield* PluginRuntime.Service const runtime = yield* PluginRuntime.Service
const agents = yield* Agent.Service const agents = yield* Agent.Service
const catalog = yield* Catalog.Service
const config = yield* Config.Service const config = yield* Config.Service
const permission = yield* Permission.Service const permission = yield* Permission.Service
const scope = yield* Scope.Scope const scope = yield* Scope.Scope
@@ -163,8 +170,14 @@ export const Plugin = {
}) })
.pipe(Effect.mapError((error) => new ToolFailure({ message: `Subagent denied: ${agent.id}`, error }))) .pipe(Effect.mapError((error) => new ToolFailure({ message: `Subagent denied: ${agent.id}`, error })))
// Model selection is policy/config/session state, not an LLM-facing tool argument. const routed = input.model
const model = agent.model ?? parent.model ? ModelRouting.resolve(input.model, yield* catalog.model.available())
: undefined
if (input.model && !routed)
return yield* new ToolFailure({
message: `No available model matches route: ${input.model}`,
})
const model = routed ?? agent.model ?? parent.model
const child = yield* runtime.session const child = yield* runtime.session
.create({ .create({
parentID: context.sessionID, parentID: context.sessionID,
+75
View File
@@ -0,0 +1,75 @@
import { describe, expect, test } from "bun:test"
import { Money } from "@opencode-ai/schema/money"
import { ModelRouting } from "@opencode-ai/core/model-routing"
import { Model } from "@opencode-ai/core/model"
import { Provider } from "@opencode-ai/core/provider"
const model = (
providerID: string,
id: string,
input: readonly string[],
options: { cost?: number; context?: number; released?: number } = {},
) =>
Model.Info.make({
...Model.Info.default(Provider.ID.make(providerID), Model.ID.make(id)),
name: id,
capabilities: { tools: true, input: [...input], output: ["text"] },
time: { released: options.released ?? 1 },
cost: [
{
input: Money.USDPerMillionTokens.make(options.cost ?? 1),
output: Money.USDPerMillionTokens.make(options.cost ?? 1),
cache: { read: Money.USDPerMillionTokens.zero, write: Money.USDPerMillionTokens.zero },
},
],
limit: { context: options.context ?? 100_000, output: 10_000 },
})
describe("ModelRouting.select", () => {
test("routes fast work to an inexpensive fast family without selecting haiku", () => {
const selected = ModelRouting.select("fast", [
model("anthropic", "claude-haiku-4", ["text"], { cost: 0.1, released: 4 }),
model("google", "gemini-flash", ["text"], { cost: 0.2, released: 3 }),
model("openai", "gpt-5", ["text"], { cost: 2, released: 5 }),
])
expect(selected).toEqual(
Model.Ref.make({ providerID: Provider.ID.make("google"), id: Model.ID.make("gemini-flash") }),
)
})
test("routes smart work to a high-capability family", () => {
const selected = ModelRouting.select("smart", [
model("google", "gemini-flash", ["text"], { released: 5 }),
model("anthropic", "claude-opus-4", ["text"], { released: 3 }),
])
expect(selected).toEqual(
Model.Ref.make({ providerID: Provider.ID.make("anthropic"), id: Model.ID.make("claude-opus-4") }),
)
})
test("enforces role capabilities and provider availability through the candidate set", () => {
expect(
ModelRouting.select("vision", [
model("openai", "gpt-5", ["text"], { released: 5 }),
model("google", "gemini-pro-vision", ["text", "image"], { released: 3 }),
]),
).toEqual(Model.Ref.make({ providerID: Provider.ID.make("google"), id: Model.ID.make("gemini-pro-vision") }))
expect(
ModelRouting.select("long-context", [
model("openai", "gpt-5", ["text"], { context: 200_000 }),
model("google", "gemini-pro", ["text"], { context: 1_000_000 }),
]),
).toEqual(Model.Ref.make({ providerID: Provider.ID.make("google"), id: Model.ID.make("gemini-pro") }))
})
test("resolves only exact models present in the available catalog", () => {
const available = [model("xai", "grok-4", ["text"])]
expect(ModelRouting.resolve("xai/grok-4", available)).toEqual(
Model.Ref.make({ providerID: Provider.ID.make("xai"), id: Model.ID.make("grok-4") }),
)
expect(ModelRouting.resolve("xai/grok-5", available)).toBeUndefined()
})
})
+1 -110
View File
@@ -14,10 +14,8 @@ import { SessionExecution } from "@opencode-ai/core/session/execution"
import { SessionRestart } from "@opencode-ai/core/session/execution/restart" import { SessionRestart } from "@opencode-ai/core/session/execution/restart"
import { UserInterruptedError } from "@opencode-ai/core/session/error" import { UserInterruptedError } from "@opencode-ai/core/session/error"
import { SessionEvent } from "@opencode-ai/core/session/event" 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 { SessionRunner } from "@opencode-ai/core/session/runner/index"
import { SessionInboxTable, SessionTable } from "@opencode-ai/core/session/sql" import { SessionTable } from "@opencode-ai/core/session/sql"
import { SessionStore } from "@opencode-ai/core/session/store" import { SessionStore } from "@opencode-ai/core/session/store"
import { Context, Deferred, Effect, Exit, Fiber, Layer, LayerMap, Scope } from "effect" import { Context, Deferred, Effect, Exit, Fiber, Layer, LayerMap, Scope } from "effect"
import { eq } from "drizzle-orm" import { eq } from "drizzle-orm"
@@ -292,113 +290,6 @@ 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( function seedSessions(
database: Database.Service["Service"], database: Database.Service["Service"],
sessionIDs: ReadonlyArray<Session.ID>, sessionIDs: ReadonlyArray<Session.ID>,
+17 -7
View File
@@ -31,7 +31,6 @@ import { testEffect } from "./lib/effect"
const executionCalls: Session.ID[] = [] const executionCalls: Session.ID[] = []
const interruptCalls: Session.ID[] = [] const interruptCalls: Session.ID[] = []
const interruptContinuations: Array<boolean | undefined> = []
const wakeCalls: Session.ID[] = [] const wakeCalls: Session.ID[] = []
const activeSessions = new Set<Session.ID>() const activeSessions = new Set<Session.ID>()
const execution = Layer.succeed( const execution = Layer.succeed(
@@ -42,16 +41,14 @@ const execution = Layer.succeed(
Effect.sync(() => { Effect.sync(() => {
executionCalls.push(sessionID) executionCalls.push(sessionID)
}), }),
interrupt: (sessionID, options) => interrupt: (sessionID) =>
Effect.sync(() => { Effect.sync(() => {
interruptCalls.push(sessionID) interruptCalls.push(sessionID)
interruptContinuations.push(options?.continue)
}), }),
wake: (sessionID) => wake: (sessionID) =>
Effect.sync(() => { Effect.sync(() => {
wakeCalls.push(sessionID) wakeCalls.push(sessionID)
}), }),
wakeActive: () => Effect.void,
awaitIdle: () => Effect.void, awaitIdle: () => Effect.void,
}), }),
) )
@@ -180,18 +177,31 @@ describe("Session.prompt", () => {
}), }),
) )
it.effect("forwards interrupt continuation policy", () => 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", () =>
Effect.gen(function* () { Effect.gen(function* () {
yield* setup yield* setup
const session = yield* Session.Service const session = yield* Session.Service
interruptCalls.length = 0 interruptCalls.length = 0
interruptContinuations.length = 0
wakeCalls.length = 0 wakeCalls.length = 0
yield* session.interrupt(sessionID, { continue: true }) yield* session.interrupt(sessionID, { continue: true })
expect(interruptCalls).toEqual([sessionID]) expect(interruptCalls).toEqual([sessionID])
expect(interruptContinuations).toEqual([true])
expect(wakeCalls).toEqual([]) expect(wakeCalls).toEqual([])
}), }),
) )
@@ -1,6 +1,5 @@
import { describe, expect } from "bun:test" import { describe, expect } from "bun:test"
import { Cause, Deferred, Effect, Exit, Fiber, Layer } from "effect" 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 { SessionRunCoordinator } from "@opencode-ai/core/session/run-coordinator"
import { testEffect } from "./lib/effect" import { testEffect } from "./lib/effect"
@@ -270,28 +269,6 @@ 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", () => it.effect("interrupts active execution and clears its pending wake", () =>
Effect.scoped( Effect.scoped(
Effect.gen(function* () { Effect.gen(function* () {
@@ -365,126 +342,6 @@ 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", () => it.effect("starts a resume registered during interruption cleanup", () =>
Effect.scoped( Effect.scoped(
Effect.gen(function* () { Effect.gen(function* () {
@@ -126,8 +126,7 @@ const execution = (llmClient: Layer.Layer<typeof LLMClient.Service>) =>
active: coordinator.active, active: coordinator.active,
resume: coordinator.run, resume: coordinator.run,
wake: coordinator.wake, wake: coordinator.wake,
wakeActive: coordinator.wakeActive, interrupt: coordinator.interrupt,
interrupt: (sessionID) => coordinator.interrupt(sessionID),
awaitIdle: coordinator.awaitIdle, awaitIdle: coordinator.awaitIdle,
}) })
}), }),
+1 -58
View File
@@ -413,8 +413,7 @@ const execution = Layer.effect(
active: coordinator.active, active: coordinator.active,
resume: coordinator.run, resume: coordinator.run,
wake: coordinator.wake, wake: coordinator.wake,
wakeActive: coordinator.wakeActive, interrupt: coordinator.interrupt,
interrupt: (sessionID) => coordinator.interrupt(sessionID),
awaitIdle: coordinator.awaitIdle, awaitIdle: coordinator.awaitIdle,
}) })
}), }),
@@ -1384,44 +1383,6 @@ 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", () => it.effect("seeds a fork with the parent's newest instruction values", () =>
Effect.gen(function* () { Effect.gen(function* () {
const session = yield* setup const session = yield* setup
@@ -3127,24 +3088,6 @@ 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", () => it.effect("promotes queued input after steering continuation ends", () =>
Effect.gen(function* () { Effect.gen(function* () {
const session = yield* setup const session = yield* setup
-1
View File
@@ -115,7 +115,6 @@ const executionNode = makeGlobalNode({
active: Effect.succeed(new Set()), active: Effect.succeed(new Set()),
resume: complete, resume: complete,
wake: () => Effect.void, wake: () => Effect.void,
wakeActive: () => Effect.void,
interrupt: () => Effect.void, interrupt: () => Effect.void,
awaitIdle: (id) => complete(id).pipe(Effect.exit, Effect.asVoid), awaitIdle: (id) => complete(id).pipe(Effect.exit, Effect.asVoid),
}) })
+55 -2
View File
@@ -8,6 +8,7 @@ import { Global } from "@opencode-ai/util/global"
import { makeGlobalNode, makeLocationNode } from "@opencode-ai/util/effect/app-node" import { makeGlobalNode, makeLocationNode } from "@opencode-ai/util/effect/app-node"
import { Database } from "@opencode-ai/core/database/database" import { Database } from "@opencode-ai/core/database/database"
import { Bus } from "@opencode-ai/core/bus" import { Bus } from "@opencode-ai/core/bus"
import { Catalog } from "@opencode-ai/core/catalog"
import { Config } from "@opencode-ai/core/config" import { Config } from "@opencode-ai/core/config"
import { Location } from "@opencode-ai/core/location" import { Location } from "@opencode-ai/core/location"
import { Model } from "@opencode-ai/core/model" import { Model } from "@opencode-ai/core/model"
@@ -36,6 +37,7 @@ import { executeTool, registerToolPlugin, toolIdentity } from "./lib/tool"
const childText = "child final response" const childText = "child final response"
const childModel = Model.Ref.make({ id: Model.ID.make("child"), providerID: Provider.ID.make("test") }) const childModel = Model.Ref.make({ id: Model.ID.make("child"), providerID: Provider.ID.make("test") })
const parentModel = Model.Ref.make({ id: Model.ID.make("parent"), providerID: Provider.ID.make("test") }) const parentModel = Model.Ref.make({ id: Model.ID.make("parent"), providerID: Provider.ID.make("test") })
const fastModel = Model.Ref.make({ id: Model.ID.make("gemini-flash"), providerID: Provider.ID.make("route") })
const tokens = { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } } const tokens = { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } }
const outputSessionID = (value: unknown) => const outputSessionID = (value: unknown) =>
@@ -86,7 +88,6 @@ const executionNode = makeGlobalNode({
active: Effect.succeed(new Set()), active: Effect.succeed(new Set()),
resume: complete, resume: complete,
wake: () => Effect.void, wake: () => Effect.void,
wakeActive: () => Effect.void,
interrupt: () => Effect.void, interrupt: () => Effect.void,
awaitIdle: (sessionID) => complete(sessionID).pipe(Effect.exit, Effect.asVoid), awaitIdle: (sessionID) => complete(sessionID).pipe(Effect.exit, Effect.asVoid),
}) })
@@ -101,7 +102,7 @@ const subagentPluginSupervisor = makeLocationNode({
PluginSupervisor.Service, PluginSupervisor.Service,
registerToolPlugin(SubagentTool.Plugin).pipe(Effect.as(PluginSupervisor.Service.of({ flush: Effect.void }))), registerToolPlugin(SubagentTool.Plugin).pipe(Effect.as(PluginSupervisor.Service.of({ flush: Effect.void }))),
), ),
deps: [Agent.node, Config.node, Permission.node, PluginRuntime.node, Tool.node], deps: [Agent.node, Catalog.node, Config.node, Permission.node, PluginRuntime.node, Tool.node],
}) })
const nodes = LayerNode.group([ const nodes = LayerNode.group([
@@ -143,6 +144,27 @@ const withSubagent = (location: Location.Ref) =>
}) })
}), }),
).pipe(Effect.provide(locations.get(location))) ).pipe(Effect.provide(locations.get(location)))
yield* Catalog.Service.use((catalog) =>
catalog.transform((draft) => {
draft.provider.update(fastModel.providerID, (provider) => {
provider.activation = "enabled"
})
draft.model.update(fastModel.providerID, fastModel.id, (model) => {
Object.assign(model, Model.Info.default(fastModel.providerID, fastModel.id), {
name: "Gemini Flash",
family: Model.Family.make("gemini-flash"),
time: { released: Date.now() },
cost: [
{
input: Money.USDPerMillionTokens.make(0.1),
output: Money.USDPerMillionTokens.make(0.2),
cache: { read: Money.USDPerMillionTokens.zero, write: Money.USDPerMillionTokens.zero },
},
],
})
})
}),
).pipe(Effect.provide(locations.get(location)))
}) })
describe("SubagentTool", () => { describe("SubagentTool", () => {
@@ -321,6 +343,37 @@ describe("SubagentTool", () => {
}) })
const fallbackChild = yield* sessions.get(outputSessionID(fallback.metadata)) const fallbackChild = yield* sessions.get(outputSessionID(fallback.metadata))
expect(fallbackChild).toMatchObject({ parentID: parent.id, model: parentModel }) expect(fallbackChild).toMatchObject({ parentID: parent.id, model: parentModel })
const routed = yield* executeTool(registry, {
sessionID: parent.id,
...toolIdentity,
call: {
type: "tool-call",
id: "call-subagent-routed",
name: SubagentTool.name,
input: { agent: "reviewer", description: "fast", prompt: "quick check", model: "fast" },
},
})
expect(yield* sessions.get(outputSessionID(routed.metadata))).toMatchObject({
parentID: parent.id,
model: fastModel,
})
expect(
yield* executeTool(registry, {
sessionID: parent.id,
...toolIdentity,
call: {
type: "tool-call",
id: "call-subagent-missing-model",
name: SubagentTool.name,
input: { agent: "reviewer", description: "missing", prompt: "check", model: "missing/model" },
},
}),
).toEqual({
status: "error",
error: { type: "tool.execution", message: "No available model matches route: missing/model" },
})
}), }),
), ),
), ),
+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 pending steering input while queued work remains parked.", "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.",
"summary": "Interrupt session execution" "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", identifier: "v2.session.interrupt",
summary: "Interrupt session execution", summary: "Interrupt session execution",
description: 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.", "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.",
}), }),
), ),
) )
@@ -1,7 +1,6 @@
import { createMemo, createResource, createSignal, onMount, Show } from "solid-js" import { createMemo, createResource, createSignal, onMount, Show } from "solid-js"
import path from "path" import path from "path"
import type { SessionInfo } from "@opencode-ai/client" import type { SessionInfo } from "@opencode-ai/client"
import { Project } from "@opencode-ai/schema/project"
import { TextAttributes } from "@opentui/core" import { TextAttributes } from "@opentui/core"
import type { RGBA } from "@opentui/core" import type { RGBA } from "@opentui/core"
import { useDialog } from "../ui/dialog" import { useDialog } from "../ui/dialog"
@@ -55,12 +54,10 @@ export function DialogSessionList() {
const response = await client.api.session.list({ const response = await client.api.session.list({
...(allProjects ...(allProjects
? {} ? {}
: current.project.id === Project.ID.global : {
? { directory: current.directory } project: current.project.id,
: { subpath: path.relative(current.project.directory, current.directory).replaceAll("\\", "/"),
project: current.project.id, }),
subpath: path.relative(current.project.directory, current.directory).replaceAll("\\", "/"),
}),
...(query ? { search: query } : {}), ...(query ? { search: query } : {}),
limit: 50, limit: 50,
order: "desc", order: "desc",
+2 -17
View File
@@ -71,7 +71,6 @@ import { DialogImagePreview } from "../dialog-image-preview"
import { useDirectoryRecents } from "../../prompt/directory-recents" import { useDirectoryRecents } from "../../prompt/directory-recents"
import { directoryRecentValue } from "../../prompt/directory-completion" import { directoryRecentValue } from "../../prompt/directory-completion"
import { useWorkingDirectoryActions } from "../../ui/working-directory-actions" import { useWorkingDirectoryActions } from "../../ui/working-directory-actions"
import { truncateFilePath } from "../../ui/file-path"
export type PromptProps = { export type PromptProps = {
sessionID?: string sessionID?: string
@@ -1556,12 +1555,6 @@ export function Prompt(props: PromptProps) {
const branch = data.location.vcs.info(location)?.branch.current const branch = data.location.vcs.info(location)?.branch.current
return branch ? `${directory}:${branch}` : directory 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({ const locationActions = useWorkingDirectoryActions({
directory: () => footerLocation()?.directory, directory: () => footerLocation()?.directory,
onMove: () => void move.open(), onMove: () => void move.open(),
@@ -1847,15 +1840,7 @@ export function Prompt(props: PromptProps) {
<box width="100%" flexDirection="row" justifyContent="space-between" gap={2}> <box width="100%" flexDirection="row" justifyContent="space-between" gap={2}>
<Slot path="prompt.footer" input={footerInput()}> <Slot path="prompt.footer" input={footerInput()}>
<Slot path="prompt.footer.status" input={footerInput()}> <Slot path="prompt.footer.status" input={footerInput()}>
<box <box flexGrow={1} flexShrink={1} minWidth={0}>
flexGrow={1}
flexShrink={1}
minWidth={0}
onSizeChange={function (this: BoxRenderable) {
const width = this.width
queueMicrotask(() => setLocationWidth(width))
}}
>
<Switch> <Switch>
<Match when={status() === "running"}> <Match when={status() === "running"}>
<box flexDirection="row" gap={1} flexGrow={1} justifyContent="flex-start"> <box flexDirection="row" gap={1} flexGrow={1} justifyContent="flex-start">
@@ -1892,7 +1877,7 @@ export function Prompt(props: PromptProps) {
</box> </box>
</Match> </Match>
<Match when={true}> <Match when={true}>
<Show when={!props.hint && locationLabelDisplay()} fallback={props.hint ?? <text />}> <Show when={!props.hint && locationLabel()} fallback={props.hint ?? <text />}>
{(location) => ( {(location) => (
<text <text
id="prompt.footer.location" id="prompt.footer.location"
-4
View File
@@ -14,10 +14,6 @@ describe("truncateFilePath", () => {
expect(truncateFilePath(path, 19)).toBe("…/dialog-select.tsx") 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", () => { test("uses remaining width for part of a long parent segment", () => {
const path = "/private/var/folders/run-17f048ec-dbb2-4b36-860c-98637bb51a8d/files" const path = "/private/var/folders/run-17f048ec-dbb2-4b36-860c-98637bb51a8d/files"
expect(truncateFilePath(path, 40)).toBe("/…/run-17f048ec-dbb2-4b36-860c-98…/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 pending steering input while queued work remains parked.", "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.",
"summary": "Interrupt session execution" "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 pending steering input while queued work remains parked.", "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.",
"summary": "Interrupt session execution" "summary": "Interrupt session execution"
} }
}, },