mirror of
https://github.com/anomalyco/opencode.git
synced 2026-08-15 17:08:21 -04:00
Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| c71ebe46df |
@@ -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.
|
|
||||||
@@ -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/
|
||||||
@@ -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) {
|
||||||
|
|||||||
@@ -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,
|
||||||
}),
|
}),
|
||||||
|
|||||||
@@ -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.
|
||||||
|
|||||||
@@ -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>
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -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([
|
||||||
|
|||||||
@@ -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,
|
||||||
|
|||||||
@@ -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()
|
||||||
|
})
|
||||||
|
})
|
||||||
@@ -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>,
|
||||||
|
|||||||
@@ -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,
|
||||||
})
|
})
|
||||||
}),
|
}),
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
@@ -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),
|
||||||
})
|
})
|
||||||
|
|||||||
@@ -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" },
|
||||||
|
})
|
||||||
}),
|
}),
|
||||||
),
|
),
|
||||||
),
|
),
|
||||||
|
|||||||
@@ -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"
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
|
|||||||
@@ -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",
|
||||||
|
|||||||
@@ -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"
|
||||||
|
|||||||
@@ -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")
|
||||||
|
|||||||
@@ -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"
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
|
|||||||
@@ -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"
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
|
|||||||
Reference in New Issue
Block a user