vrr/opencode
2
0
Fork 0
mirror of https://github.com/anomalyco/opencode.git synced 2026-08-04 13:09:54 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 6

refactor(core): specialize session run coordination

Browse source
This commit is contained in:
Kit Langton 2026-06-06 23:13:52 -04:00
parent fc0cf2a710
commit e6bb88bc1d
4 changed files with 390 additions and 747 deletions
Download patch file Download diff file Expand all files Collapse all files

358
packages/core/src/session/run-coordinator-machine.ts
View file

@ -1,358 +0,0 @@
/** @internal Pure state machine for the process-local Session run coordinator. */
export * as SessionRunCoordinatorMachine from "./run-coordinator-machine"
/** @internal */
export type Mode = "run" | "wake"
/** @internal */
export type Demand = {
readonly explicit: boolean
readonly wakeSeq?: number
readonly unsequencedWake: boolean
}
type NonEmptyDemand = Demand &
({ readonly explicit: true } | { readonly wakeSeq: number } | { readonly unsequencedWake: true })
/** @internal */
export const Demand = {
empty: { explicit: false, unsequencedWake: false } satisfies Demand,
explicit: { explicit: true, unsequencedWake: false } satisfies NonEmptyDemand,
wake: (seq?: number): NonEmptyDemand =>
seq === undefined
? { explicit: false, unsequencedWake: true }
: { explicit: false, wakeSeq: seq, unsequencedWake: false },
combine: (left: Demand, right: Demand): Demand => ({
explicit: left.explicit || right.explicit,
wakeSeq:
left.wakeSeq === undefined
? right.wakeSeq
: right.wakeSeq === undefined
? left.wakeSeq
: Math.max(left.wakeSeq, right.wakeSeq),
unsequencedWake: left.unsequencedWake || right.unsequencedWake,
}),
afterBoundary: (demand: Demand, boundary?: number): Demand => ({
explicit: false,
wakeSeq:
boundary !== undefined && demand.wakeSeq !== undefined && demand.wakeSeq > boundary ? demand.wakeSeq : undefined,
unsequencedWake: false,
}),
nonEmpty: (demand: Demand): demand is NonEmptyDemand =>
demand.explicit || demand.wakeSeq !== undefined || demand.unsequencedWake,
mode: (demand: Demand): Mode => (demand.explicit ? "run" : "wake"),
}
type Running = {
readonly _tag: "Running"
readonly chain: number
readonly attempt: number
readonly current: NonEmptyDemand
readonly pending: Demand
readonly waiter?: number
}
type Stopping = {
readonly _tag: "Stopping"
readonly chain: number
readonly attempt: number
readonly current: NonEmptyDemand
readonly pending: Demand
readonly waiter?: number
readonly stopBoundary?: number
}
/** @internal */
export type Lane = Running | Stopping
/** @internal */
export type State<Key> = {
readonly closed: boolean
readonly nextID: number
readonly lanes: ReadonlyMap<Key, Lane>
readonly interruptSeq: ReadonlyMap<Key, number>
}
/** @internal */
export const initial = <Key>(): State<Key> => ({ closed: false, nextID: 1, lanes: new Map(), interruptSeq: new Map() })
/** @internal */
export type Outcome = "Success" | "Failure" | "Interrupted"
/** @internal */
export type Event<Key> =
| { readonly _tag: "Close" }
| { readonly _tag: "Run"; readonly key: Key }
| { readonly _tag: "Wake"; readonly key: Key; readonly seq?: number }
| { readonly _tag: "Interrupt"; readonly key: Key; readonly seq?: number }
| { readonly _tag: "Observe"; readonly key: Key }
| {
readonly _tag: "Settled"
readonly key: Key
readonly chain: number
readonly attempt: number
readonly outcome: Outcome
}
/** @internal */
export type Action<Key> =
| {
readonly _tag: "Start"
readonly key: Key
readonly chain: number
readonly attempt: number
readonly demand: NonEmptyDemand
readonly successor: boolean
}
| { readonly _tag: "Interrupt"; readonly attempt: number }
| { readonly _tag: "CompleteChain"; readonly chain: number }
| { readonly _tag: "CompleteWaiter"; readonly waiter: number }
| { readonly _tag: "Report"; readonly key: Key }
/** @internal */
export type Response =
| { readonly _tag: "None" }
| { readonly _tag: "AwaitChain"; readonly chain: number }
| { readonly _tag: "AwaitWaiter"; readonly waiter: number }
| { readonly _tag: "RetryAfter"; readonly chain: number }
| { readonly _tag: "ObserveChain"; readonly chain: number }
| { readonly _tag: "Idle" }
| { readonly _tag: "Closed" }
/** @internal */
export type Transition<Key> = {
readonly state: State<Key>
readonly actions: ReadonlyArray<Action<Key>>
readonly response: Response
}
const none: Response = { _tag: "None" }
/** @internal */
export const reduce = <Key>(state: State<Key>, event: Event<Key>): Transition<Key> => {
if (event._tag === "Close")
return { state: { ...state, closed: true, lanes: new Map(), interruptSeq: new Map() }, actions: [], response: none }
if (state.closed) return { state, actions: [], response: event._tag === "Run" ? { _tag: "Closed" } : none }
if (event._tag === "Run") return run(state, event)
if (event._tag === "Wake") return wake(state, event)
if (event._tag === "Interrupt") return interrupt(state, event)
if (event._tag === "Observe") {
const lane = state.lanes.get(event.key)
return {
state,
actions: [],
response: lane === undefined ? { _tag: "Idle" } : { _tag: "ObserveChain", chain: lane.chain },
}
}
return settled(state, event)
}
const run = <Key>(state: State<Key>, event: Extract<Event<Key>, { _tag: "Run" }>): Transition<Key> => {
const lane = state.lanes.get(event.key)
if (lane?._tag === "Stopping") return { state, actions: [], response: { _tag: "RetryAfter", chain: lane.chain } }
if (lane !== undefined && lane.current.explicit)
return { state, actions: [], response: { _tag: "AwaitChain", chain: lane.chain } }
if (lane !== undefined) {
const [allocated, waiter] = lane.waiter === undefined ? allocate(state) : [state, lane.waiter]
return {
state: setLane(allocated, event.key, { ...lane, pending: Demand.combine(lane.pending, Demand.explicit), waiter }),
actions: [],
response: { _tag: "AwaitWaiter", waiter },
}
}
const [withChain, chain] = allocate(state)
const [allocated, attempt] = allocate(withChain)
const next: Lane = {
_tag: "Running",
chain,
attempt,
current: Demand.explicit,
pending: Demand.empty,
}
return {
state: setLane(allocated, event.key, next),
actions: [
{
_tag: "Start",
key: event.key,
chain: next.chain,
attempt: next.attempt,
demand: next.current,
successor: false,
},
],
response: { _tag: "AwaitChain", chain: next.chain },
}
}
const wake = <Key>(state: State<Key>, event: Extract<Event<Key>, { _tag: "Wake" }>): Transition<Key> => {
const boundary = state.interruptSeq.get(event.key)
if (boundary !== undefined && (event.seq === undefined || event.seq <= boundary))
return { state, actions: [], response: none }
const lane = state.lanes.get(event.key)
if (lane !== undefined) {
if (
lane._tag === "Stopping" &&
(lane.stopBoundary === undefined || event.seq === undefined || event.seq <= lane.stopBoundary)
)
return { state, actions: [], response: none }
return {
state: setLane(state, event.key, { ...lane, pending: Demand.combine(lane.pending, Demand.wake(event.seq)) }),
actions: [],
response: none,
}
}
const [withChain, chain] = allocate(state)
const [allocated, attempt] = allocate(withChain)
const next: Lane = {
_tag: "Running",
chain,
attempt,
current: Demand.wake(event.seq),
pending: Demand.empty,
}
return {
state: setLane(allocated, event.key, next),
actions: [
{
_tag: "Start",
key: event.key,
chain: next.chain,
attempt: next.attempt,
demand: next.current,
successor: false,
},
],
response: none,
}
}
const interrupt = <Key>(state: State<Key>, event: Extract<Event<Key>, { _tag: "Interrupt" }>): Transition<Key> => {
const latest = state.interruptSeq.get(event.key)
const lane = state.lanes.get(event.key)
if (event.seq !== undefined && latest !== undefined && event.seq <= latest)
return {
state,
actions: lane?._tag === "Stopping" ? [{ _tag: "Interrupt", attempt: lane.attempt }] : [],
response: none,
}
const bounded = event.seq === undefined ? state : setInterruptSeq(state, event.key, event.seq)
if (lane === undefined) return { state: bounded, actions: [], response: none }
if (
!lane.current.explicit &&
event.seq !== undefined &&
lane.current.wakeSeq !== undefined &&
lane.current.wakeSeq > event.seq
)
return { state: bounded, actions: [], response: none }
const pending = Demand.combine(
Demand.afterBoundary(lane.current, event.seq),
Demand.afterBoundary(lane.pending, event.seq),
)
return {
state: setLane(bounded, event.key, {
_tag: "Stopping",
chain: lane.chain,
attempt: lane.attempt,
current: lane.current,
pending,
waiter: lane.waiter,
stopBoundary: lane._tag === "Stopping" ? maxSeq(lane.stopBoundary, event.seq) : event.seq,
}),
actions: [{ _tag: "Interrupt", attempt: lane.attempt }],
response: none,
}
}
const settled = <Key>(state: State<Key>, event: Extract<Event<Key>, { _tag: "Settled" }>): Transition<Key> => {
const lane = state.lanes.get(event.key)
if (lane?.chain !== event.chain || lane.attempt !== event.attempt) return { state, actions: [], response: none }
const completesWaiter = lane.current.explicit || (lane._tag === "Stopping" && !lane.current.explicit)
const waiterActions: ReadonlyArray<Action<Key>> =
completesWaiter && lane.waiter !== undefined ? [{ _tag: "CompleteWaiter", waiter: lane.waiter }] : []
const waiter = completesWaiter ? undefined : lane.waiter
if (event.outcome === "Success" && lane._tag === "Running" && Demand.nonEmpty(lane.pending)) {
const [allocated, attempt] = allocate(state)
const next = { ...lane, attempt, current: lane.pending, pending: Demand.empty, waiter }
return {
state: setLane(allocated, event.key, next),
actions: [
...waiterActions,
{
_tag: "Start",
key: event.key,
chain: next.chain,
attempt: next.attempt,
demand: next.current,
successor: true,
},
],
response: none,
}
}
const report: ReadonlyArray<Action<Key>> =
event.outcome !== "Success" &&
!(lane._tag === "Stopping" && event.outcome === "Interrupted") &&
!lane.current.explicit
? [{ _tag: "Report", key: event.key }]
: []
if (!Demand.nonEmpty(lane.pending))
return {
state: deleteLane(state, event.key),
actions: [...waiterActions, { _tag: "CompleteChain", chain: lane.chain }, ...report],
response: none,
}
const [withChain, chain] = allocate(state)
const [allocated, attempt] = allocate(withChain)
const next: Lane = {
_tag: "Running",
chain,
attempt,
current: lane.pending,
pending: Demand.empty,
waiter,
}
return {
state: setLane(allocated, event.key, next),
actions: [
...waiterActions,
{
_tag: "Start",
key: event.key,
chain: next.chain,
attempt: next.attempt,
demand: next.current,
successor: true,
},
{ _tag: "CompleteChain", chain: lane.chain },
...report,
],
response: none,
}
}
const maxSeq = (left?: number, right?: number) =>
left === undefined ? right : right === undefined ? left : Math.max(left, right)
const allocate = <Key>(state: State<Key>): readonly [State<Key>, number] => [
{ ...state, nextID: state.nextID + 1 },
state.nextID,
]
const setLane = <Key>(state: State<Key>, key: Key, lane: Lane): State<Key> => {
const lanes = new Map(state.lanes)
lanes.set(key, lane)
return { ...state, lanes }
}
const deleteLane = <Key>(state: State<Key>, key: Key): State<Key> => {
const lanes = new Map(state.lanes)
lanes.delete(key)
return { ...state, lanes }
}
const setInterruptSeq = <Key>(state: State<Key>, key: Key, seq: number): State<Key> => {
const interruptSeq = new Map(state.interruptSeq)
interruptSeq.set(key, seq)
return { ...state, interruptSeq }
}

491
packages/core/src/session/run-coordinator.ts
View file

@ -1,11 +1,23 @@
export * as SessionRunCoordinator from "./run-coordinator"
import { Cause, Context, Deferred, Effect, Exit, Fiber, FiberSet, Layer, Scope, SynchronizedRef } from "effect"
import {
Cause,
Context,
Data,
Deferred,
Effect,
Equal,
Exit,
Fiber,
FiberSet,
Layer,
Scope,
SynchronizedRef,
} from "effect"
import { SessionRunner } from "./runner"
import { SessionSchema } from "./schema"
import { SessionRunCoordinatorMachine } from "./run-coordinator-machine"
export type Mode = SessionRunCoordinatorMachine.Mode
export type Mode = "run" | "wake"
export interface Coordinator<Key, A, E> {
readonly run: (key: Key) => Effect.Effect<A, E>
@ -14,9 +26,102 @@ export interface Coordinator<Key, A, E> {
readonly interrupt: (key: Key, seq?: number) => Effect.Effect<void>
}
type Chain<A, E> = {
readonly done: Deferred.Deferred<A, E>
readonly settled: Deferred.Deferred<Exit.Exit<A, E>>
/** @internal */
export class Demand extends Data.Class<{
readonly explicit: boolean
readonly wakeSeq?: number
readonly unsequencedWake: boolean
}> {
static readonly empty = new Demand({ explicit: false, wakeSeq: undefined, unsequencedWake: false })
static readonly run = nonEmpty(new Demand({ explicit: true, wakeSeq: undefined, unsequencedWake: false }))
static wake(seq?: number) {
return nonEmpty(new Demand({ explicit: false, wakeSeq: seq, unsequencedWake: seq === undefined }))
}
combine(other: Demand) {
return new Demand({
explicit: this.explicit || other.explicit,
wakeSeq:
this.wakeSeq === undefined
? other.wakeSeq
: other.wakeSeq === undefined
? this.wakeSeq
: Math.max(this.wakeSeq, other.wakeSeq),
unsequencedWake: this.unsequencedWake || other.unsequencedWake,
})
}
afterBoundary(boundary?: number) {
return new Demand({
explicit: false,
wakeSeq:
boundary !== undefined && this.wakeSeq !== undefined && this.wakeSeq > boundary ? this.wakeSeq : undefined,
unsequencedWake: false,
})
}
isNonEmpty(): this is NonEmptyDemand {
return this.explicit || this.wakeSeq !== undefined || this.unsequencedWake
}
get mode(): Mode {
return this.explicit ? "run" : "wake"
}
}
type NonEmptyDemand = Demand &
({ readonly explicit: true } | { readonly wakeSeq: number } | { readonly unsequencedWake: true })
function nonEmpty(demand: Demand): NonEmptyDemand {
if (!demand.isNonEmpty()) throw new Error("Session run demand must not be empty")
return demand
}
type Lifecycle =
| { readonly _tag: "Running"; readonly token: object; readonly owner: Deferred.Deferred<Fiber.Fiber<void>> }
| {
readonly _tag: "Stopping"
readonly token: object
readonly owner: Deferred.Deferred<Fiber.Fiber<void>>
readonly boundary?: number
}
type Lane<A, E> = {
readonly current: NonEmptyDemand
readonly pending: Demand
readonly lifecycle: Lifecycle
readonly terminal: Deferred.Deferred<Exit.Exit<A, E>>
readonly waiter?: Deferred.Deferred<Exit.Exit<A, E>>
}
type State<Key, A, E> = {
readonly closed: boolean
readonly lanes: ReadonlyMap<Key, Lane<A, E>>
readonly interruptSeq: ReadonlyMap<Key, number>
}
type Start<Key, A, E> = {
readonly key: Key
readonly demand: NonEmptyDemand
readonly successor: boolean
readonly token: object
readonly owner: Deferred.Deferred<Fiber.Fiber<void>>
readonly ready: Deferred.Deferred<void>
readonly terminal: Deferred.Deferred<Exit.Exit<A, E>>
}
type RunRequest<Key, A, E> =
| { readonly _tag: "Closed" }
| { readonly _tag: "Await"; readonly terminal: Deferred.Deferred<Exit.Exit<A, E>> }
| { readonly _tag: "Retry"; readonly terminal: Deferred.Deferred<Exit.Exit<A, E>> }
| { readonly _tag: "Start"; readonly start: Start<Key, A, E>; readonly terminal: Deferred.Deferred<Exit.Exit<A, E>> }
type Completion<Key, A, E> = {
readonly start?: Start<Key, A, E>
readonly terminal?: Deferred.Deferred<Exit.Exit<A, E>>
readonly waiter?: Deferred.Deferred<Exit.Exit<A, E>>
readonly report?: Cause.Cause<E>
}
export const make = <Key, A, E>(options: {
@ -24,218 +129,256 @@ export const make = <Key, A, E>(options: {
readonly onFailure?: (key: Key, cause: Cause.Cause<E>) => Effect.Effect<void>
}): Effect.Effect<Coordinator<Key, A, E>, never, Scope.Scope> =>
Effect.gen(function* () {
const state = yield* SynchronizedRef.make(SessionRunCoordinatorMachine.initial<Key>())
const report = yield* FiberSet.makeRuntime<never, void, never>()
const state = yield* SynchronizedRef.make<State<Key, A, E>>({
closed: false,
lanes: new Map(),
interruptSeq: new Map(),
})
const fork = yield* FiberSet.makeRuntime<never, void, never>()
const chains = new Map<number, Chain<A, E>>()
const waiters = new Map<number, Deferred.Deferred<A, E>>()
const owners = new Map<number, Deferred.Deferred<Fiber.Fiber<void>>>()
const shutdown = Deferred.makeUnsafe<void>()
const chain = (chainID: number) => {
const existing = chains.get(chainID)
if (existing !== undefined) return existing
const created = { done: Deferred.makeUnsafe<A, E>(), settled: Deferred.makeUnsafe<Exit.Exit<A, E>>() }
chains.set(chainID, created)
return created
}
const waiter = (waiterID: number) => {
const existing = waiters.get(waiterID)
if (existing !== undefined) return existing
const created = Deferred.makeUnsafe<A, E>()
waiters.set(waiterID, created)
return created
}
const owner = (attemptID: number) => {
const existing = owners.get(attemptID)
if (existing !== undefined) return existing
const created = Deferred.makeUnsafe<Fiber.Fiber<void>>()
owners.set(attemptID, created)
return created
const updateLane = (current: State<Key, A, E>, key: Key, lane?: Lane<A, E>): State<Key, A, E> => {
const lanes = new Map(current.lanes)
if (lane === undefined) lanes.delete(key)
else lanes.set(key, lane)
return { ...current, lanes }
}
const requireChain = (chainID: number) => {
const existing = chains.get(chainID)
if (existing !== undefined) return existing
throw new Error(`Missing Session run chain ${chainID}`)
}
const requireWaiter = (waiterID: number) => {
const existing = waiters.get(waiterID)
if (existing !== undefined) return existing
throw new Error(`Missing Session run waiter ${waiterID}`)
const start = (input: {
readonly state: State<Key, A, E>
readonly key: Key
readonly demand: NonEmptyDemand
readonly terminal?: Deferred.Deferred<Exit.Exit<A, E>>
readonly waiter?: Deferred.Deferred<Exit.Exit<A, E>>
readonly successor?: boolean
}) => {
const instruction: Start<Key, A, E> = {
key: input.key,
demand: input.demand,
successor: input.successor ?? false,
token: {},
owner: Deferred.makeUnsafe<Fiber.Fiber<void>>(),
ready: Deferred.makeUnsafe<void>(),
terminal: input.terminal ?? Deferred.makeUnsafe<Exit.Exit<A, E>>(),
}
return {
state: updateLane(input.state, input.key, {
current: input.demand,
pending: Demand.empty,
lifecycle: { _tag: "Running", token: instruction.token, owner: instruction.owner },
terminal: instruction.terminal,
waiter: input.waiter,
}),
start: instruction,
result: instruction.terminal,
}
}
type RuntimeResponse =
| { readonly _tag: "None" | "Idle" | "Closed" }
| { readonly _tag: "Await"; readonly deferred: Deferred.Deferred<A, E> }
| { readonly _tag: "Retry" | "Observe"; readonly deferred: Deferred.Deferred<Exit.Exit<A, E>> }
const transition = (event: SessionRunCoordinatorMachine.Event<Key>) =>
SynchronizedRef.modifyEffect(state, (current) => {
const result = SessionRunCoordinatorMachine.reduce(current, event)
return Effect.sync(() => {
result.actions.forEach((action) => {
if (action._tag !== "Start") return
chain(action.chain)
owner(action.attempt)
})
if (result.response._tag === "AwaitChain") chain(result.response.chain)
if (result.response._tag === "AwaitWaiter") waiter(result.response.waiter)
const response: RuntimeResponse =
result.response._tag === "AwaitChain"
? { _tag: "Await", deferred: requireChain(result.response.chain).done }
: result.response._tag === "AwaitWaiter"
? { _tag: "Await", deferred: requireWaiter(result.response.waiter) }
: result.response._tag === "RetryAfter"
? { _tag: "Retry", deferred: requireChain(result.response.chain).settled }
: result.response._tag === "ObserveChain"
? { _tag: "Observe", deferred: requireChain(result.response.chain).settled }
: { _tag: result.response._tag }
return [{ actions: result.actions, response }, result.state] as const
})
const launch = (instruction: Start<Key, A, E>) =>
Effect.gen(function* () {
const fiber = fork(
Deferred.await(instruction.ready).pipe(
Effect.andThen(instruction.successor ? Effect.yieldNow : Effect.void),
Effect.andThen(Effect.suspend(() => options.drain(instruction.key, instruction.demand.mode))),
Effect.onExit((exit) => complete(instruction.key, instruction.token, exit)),
Effect.exit,
Effect.asVoid,
),
)
yield* Deferred.succeed(instruction.owner, fiber)
yield* Deferred.succeed(instruction.ready, undefined)
})
type Execution = { readonly _tag: "General" } | { readonly _tag: "Settlement"; readonly exit: Exit.Exit<A, E> }
const complete = (key: Key, token: object, exit: Exit.Exit<A, E>): Effect.Effect<void> => {
return SynchronizedRef.modify(state, (current): readonly [Completion<Key, A, E>, State<Key, A, E>] => {
const lane = current.lanes.get(key)
if (lane === undefined || lane.lifecycle.token !== token) return [{}, current]
const execute = (
actions: ReadonlyArray<SessionRunCoordinatorMachine.Action<Key>>,
execution: Execution,
): Effect.Effect<void> =>
Effect.forEach(
actions,
(action): Effect.Effect<void> => {
if (action._tag === "Start") {
requireChain(action.chain)
const ownerDeferred = owners.get(action.attempt)
if (ownerDeferred === undefined) return Effect.die(`Missing Session run attempt ${action.attempt}`)
const ready = Deferred.makeUnsafe<void>()
const drain = Effect.suspend(() =>
options.drain(action.key, SessionRunCoordinatorMachine.Demand.mode(action.demand)),
)
const fiber = fork(
(action.successor
? Effect.yieldNow.pipe(Effect.andThen(drain))
: Deferred.await(ready).pipe(Effect.andThen(drain))
).pipe(
Effect.onExit((result) => settle(action.key, action.chain, action.attempt, result)),
Effect.exit,
Effect.asVoid,
),
)
Deferred.doneUnsafe(ownerDeferred, Effect.succeed(fiber))
if (!action.successor) Deferred.doneUnsafe(ready, Effect.void)
return Effect.void
}
if (action._tag === "Interrupt") {
const ownerDeferred = owners.get(action.attempt)
return ownerDeferred === undefined
? Effect.void
: Deferred.await(ownerDeferred).pipe(Effect.flatMap(Fiber.interrupt))
}
if (execution._tag !== "Settlement") return Effect.die("Settlement action requires a settlement context")
if (action._tag === "CompleteWaiter") {
const deferred = requireWaiter(action.waiter)
waiters.delete(action.waiter)
Deferred.doneUnsafe(deferred, execution.exit)
return Effect.void
}
if (action._tag === "CompleteChain") {
const deferreds = requireChain(action.chain)
chains.delete(action.chain)
Deferred.doneUnsafe(deferreds.done, execution.exit)
Deferred.doneUnsafe(deferreds.settled, Effect.succeed(execution.exit))
return Effect.void
}
if (action._tag === "Report") {
const onFailure = options.onFailure
if (execution.exit._tag === "Success") return Effect.die("Failure report requires a failed settlement")
if (onFailure === undefined) return Effect.void
const cause = execution.exit.cause
report(Effect.suspend(() => onFailure(action.key, cause)))
}
return Effect.void
},
{ discard: true },
).pipe(Effect.asVoid)
const deliberateInterrupt =
lane.lifecycle._tag === "Stopping" && exit._tag === "Failure" && Cause.hasInterruptsOnly(exit.cause)
const report =
exit._tag === "Failure" && !deliberateInterrupt && !lane.current.explicit ? exit.cause : undefined
const completesWaiter = lane.current.explicit || (lane.lifecycle._tag === "Stopping" && !lane.current.explicit)
const waiter = completesWaiter ? undefined : lane.waiter
const settle = (key: Key, chainID: number, attemptID: number, exit: Exit.Exit<A, E>) => {
return transition({
_tag: "Settled",
key,
chain: chainID,
attempt: attemptID,
outcome: exit._tag === "Success" ? "Success" : Cause.hasInterruptsOnly(exit.cause) ? "Interrupted" : "Failure",
}).pipe(
Effect.flatMap((result) => execute(result.actions, { _tag: "Settlement", exit })),
Effect.ensuring(Effect.sync(() => owners.delete(attemptID))),
)
if (exit._tag === "Success" && lane.lifecycle._tag === "Running" && lane.pending.isNonEmpty()) {
const next = start({
state: current,
key,
demand: lane.pending,
terminal: lane.terminal,
waiter,
successor: true,
})
return [{ start: next.start, waiter: completesWaiter ? lane.waiter : undefined, report }, next.state]
}
const next = lane.pending.isNonEmpty()
? start({ state: current, key, demand: lane.pending, waiter, successor: true })
: { state: updateLane(current, key) }
return [
{
start: "start" in next ? next.start : undefined,
terminal: lane.terminal,
waiter: completesWaiter ? lane.waiter : undefined,
report,
},
next.state,
]
}).pipe(Effect.flatMap((instruction) => executeCompletion(key, exit, instruction)))
}
const dispatch = (event: SessionRunCoordinatorMachine.Event<Key>) =>
Effect.uninterruptible(
transition(event).pipe(Effect.flatMap((result) => execute(result.actions, { _tag: "General" }))),
const executeCompletion = (key: Key, exit: Exit.Exit<A, E>, instruction: Completion<Key, A, E>) =>
Effect.gen(function* () {
if (instruction.start !== undefined) yield* launch(instruction.start)
if (instruction.waiter !== undefined) yield* Deferred.succeed(instruction.waiter, exit)
if (instruction.terminal !== undefined) yield* Deferred.succeed(instruction.terminal, exit)
if (instruction.report !== undefined && options.onFailure !== undefined) {
const onFailure = options.onFailure
const cause = instruction.report
fork(Effect.suspend(() => onFailure(key, cause)).pipe(Effect.exit, Effect.asVoid))
}
})
const awaitTerminal = (terminal: Deferred.Deferred<Exit.Exit<A, E>>) =>
Effect.raceFirst(
Deferred.await(terminal).pipe(
Effect.flatMap(
Exit.match({
onSuccess: Effect.succeed,
onFailure: Effect.failCause,
}),
),
),
Deferred.await(shutdown).pipe(Effect.andThen(Effect.interrupt)),
)
const run = (key: Key): Effect.Effect<A, E> =>
Effect.suspend(() =>
Effect.uninterruptibleMask((restore) => {
return transition({ _tag: "Run", key }).pipe(
Effect.flatMap((result) => {
return execute(result.actions, { _tag: "General" }).pipe(
Effect.andThen(
result.response._tag === "Await"
? awaitResult(result.response.deferred)
: result.response._tag === "Retry"
? Effect.raceFirst(
Deferred.await(result.response.deferred).pipe(Effect.as(true)),
Deferred.await(shutdown).pipe(Effect.as(false)),
).pipe(Effect.flatMap((settled) => (settled ? run(key) : Effect.interrupt)))
: Effect.interrupt,
),
restore,
)
return SynchronizedRef.modify(state, (current): readonly [RunRequest<Key, A, E>, State<Key, A, E>] => {
if (current.closed) return [{ _tag: "Closed" }, current]
const lane = current.lanes.get(key)
if (lane?.lifecycle._tag === "Stopping") return [{ _tag: "Retry", terminal: lane.terminal }, current]
if (lane?.current.explicit) return [{ _tag: "Await", terminal: lane.terminal }, current]
if (lane !== undefined) {
const terminal = lane.waiter ?? Deferred.makeUnsafe<Exit.Exit<A, E>>()
const pending = lane.pending.combine(Demand.run)
if (Equal.equals(pending, lane.pending) && lane.waiter !== undefined)
return [{ _tag: "Await", terminal }, current]
return [{ _tag: "Await", terminal }, updateLane(current, key, { ...lane, pending, waiter: terminal })]
}
const next = start({ state: current, key, demand: Demand.run })
return [{ _tag: "Start", start: next.start, terminal: next.result }, next.state]
}).pipe(
Effect.flatMap((request) => {
if (request._tag === "Closed") return Effect.interrupt
if (request._tag === "Start")
return launch(request.start).pipe(Effect.andThen(awaitTerminal(request.terminal)))
if (request._tag === "Await") return awaitTerminal(request.terminal)
return Effect.raceFirst(
Deferred.await(request.terminal).pipe(Effect.as(true)),
Deferred.await(shutdown).pipe(Effect.as(false)),
).pipe(Effect.flatMap((retry) => (retry ? run(key) : Effect.interrupt)))
}),
restore,
)
}),
)
const wake = (key: Key, seq?: number) => {
return Effect.uninterruptible(
const wake = (key: Key, seq?: number) =>
Effect.uninterruptible(
Effect.suspend(() => {
return transition({ _tag: "Wake", key, seq }).pipe(
Effect.flatMap((result) => execute(result.actions, { _tag: "General" })),
)
return SynchronizedRef.modify(state, (current): readonly [Start<Key, A, E> | undefined, State<Key, A, E>] => {
if (current.closed) return [undefined, current]
const boundary = current.interruptSeq.get(key)
if (boundary !== undefined && (seq === undefined || seq <= boundary)) return [undefined, current]
const lane = current.lanes.get(key)
if (lane === undefined) {
const next = start({ state: current, key, demand: Demand.wake(seq) })
return [next.start, next.state]
}
if (
lane.lifecycle._tag === "Stopping" &&
(lane.lifecycle.boundary === undefined || seq === undefined || seq <= lane.lifecycle.boundary)
)
return [undefined, current]
const pending = lane.pending.combine(Demand.wake(seq))
if (Equal.equals(pending, lane.pending)) return [undefined, current]
return [undefined, updateLane(current, key, { ...lane, pending })]
}).pipe(Effect.flatMap((instruction) => (instruction === undefined ? Effect.void : launch(instruction))))
}),
)
}
const interrupt = (key: Key, seq?: number) => dispatch({ _tag: "Interrupt", key, seq })
const interrupt = (key: Key, seq?: number) =>
Effect.uninterruptible(
SynchronizedRef.modify(state, (current) => {
if (current.closed) return [undefined, current] as const
const latest = current.interruptSeq.get(key)
const lane = current.lanes.get(key)
if (seq !== undefined && latest !== undefined && seq <= latest)
return [lane?.lifecycle._tag === "Stopping" ? lane.lifecycle.owner : undefined, current] as const
const bounded = (() => {
if (seq === undefined) return current
const interruptSeq = new Map(current.interruptSeq)
interruptSeq.set(key, seq)
return { ...current, interruptSeq }
})()
if (lane === undefined) return [undefined, bounded] as const
if (
!lane.current.explicit &&
seq !== undefined &&
lane.current.wakeSeq !== undefined &&
lane.current.wakeSeq > seq
)
return [undefined, bounded] as const
const pending = lane.current.afterBoundary(seq).combine(lane.pending.afterBoundary(seq))
const boundary =
lane.lifecycle._tag === "Stopping" && lane.lifecycle.boundary !== undefined && seq !== undefined
? Math.max(lane.lifecycle.boundary, seq)
: lane.lifecycle._tag === "Stopping" && seq === undefined
? lane.lifecycle.boundary
: seq
return [
lane.lifecycle.owner,
updateLane(bounded, key, {
...lane,
pending,
lifecycle: { _tag: "Stopping", token: lane.lifecycle.token, owner: lane.lifecycle.owner, boundary },
}),
] as const
}).pipe(
Effect.flatMap((owner) =>
owner === undefined ? Effect.void : Deferred.await(owner).pipe(Effect.flatMap(Fiber.interrupt)),
),
),
)
const awaitIdle = (key: Key): Effect.Effect<void, E> =>
Effect.gen(function* () {
let failure: Cause.Cause<E> | undefined
while (true) {
const observation = yield* transition({ _tag: "Observe", key })
if (observation.response._tag !== "Observe") break
const terminal = (yield* SynchronizedRef.get(state)).lanes.get(key)?.terminal
if (terminal === undefined) break
const exit = yield* Effect.raceFirst(
Deferred.await(observation.response.deferred),
Deferred.await(terminal),
Deferred.await(shutdown).pipe(Effect.as(Exit.void)),
)
if (exit._tag === "Failure" && failure === undefined) failure = exit.cause
}
if (failure !== undefined) return yield* Effect.failCause(failure)
return undefined
})
yield* Effect.addFinalizer(() =>
transition({ _tag: "Close" }).pipe(Effect.andThen(Effect.sync(() => Deferred.doneUnsafe(shutdown, Effect.void)))),
SynchronizedRef.modify(state, (_current) => [
undefined,
{ closed: true, lanes: new Map(), interruptSeq: new Map() } satisfies State<Key, A, E>,
]).pipe(Effect.andThen(Deferred.succeed(shutdown, undefined))),
)
return { run, wake, interrupt, awaitIdle }
function awaitResult(deferred: Deferred.Deferred<A, E>) {
return Effect.raceFirst(Deferred.await(deferred), Deferred.await(shutdown).pipe(Effect.andThen(Effect.interrupt)))
}
})
export interface Interface extends Coordinator<SessionSchema.ID, void, SessionRunner.RunError> {}

213
packages/core/test/session-run-coordinator-machine.test.ts
View file

@ -1,213 +0,0 @@
import { describe, expect, test } from "bun:test"
import { SessionRunCoordinatorMachine } from "../src/session/run-coordinator-machine"
const Machine = SessionRunCoordinatorMachine
describe("SessionRunCoordinatorMachine.Demand", () => {
test("empty is the combine identity", () => {
const demand = SessionRunCoordinatorMachine.Demand.combine(
SessionRunCoordinatorMachine.Demand.explicit,
SessionRunCoordinatorMachine.Demand.wake(3),
)
expect(SessionRunCoordinatorMachine.Demand.combine(SessionRunCoordinatorMachine.Demand.empty, demand)).toEqual(
demand,
)
expect(SessionRunCoordinatorMachine.Demand.combine(demand, SessionRunCoordinatorMachine.Demand.empty)).toEqual(
demand,
)
})
test("combine is associative, commutative, and idempotent", () => {
const left = Machine.Demand.explicit
const middle = Machine.Demand.wake()
const right = Machine.Demand.wake(3)
expect(Machine.Demand.combine(left, right)).toEqual(Machine.Demand.combine(right, left))
expect(Machine.Demand.combine(left, left)).toEqual(left)
expect(Machine.Demand.combine(Machine.Demand.combine(left, middle), right)).toEqual(
Machine.Demand.combine(left, Machine.Demand.combine(middle, right)),
)
})
test("afterBoundary removes explicit and stale wake components", () => {
const combined = Machine.Demand.combine(Machine.Demand.explicit, Machine.Demand.wake(3))
expect(Machine.Demand.afterBoundary(combined, 2)).toEqual(Machine.Demand.wake(3))
expect(Machine.Demand.nonEmpty(Machine.Demand.afterBoundary(combined, 3))).toBeFalse()
expect(Machine.Demand.nonEmpty(Machine.Demand.afterBoundary(combined))).toBeFalse()
})
test("mode follows only the explicit component", () => {
expect(Machine.Demand.mode(Machine.Demand.explicit)).toBe("run")
expect(Machine.Demand.mode(Machine.Demand.wake(1))).toBe("wake")
expect(Machine.Demand.mode(Machine.Demand.combine(Machine.Demand.explicit, Machine.Demand.wake(1)))).toBe("run")
})
})
describe("SessionRunCoordinatorMachine.reduce", () => {
test("ignores a stale attempt from the active chain", () => {
const active = combinedActive(3)
const result = Machine.reduce(active, {
_tag: "Settled",
key: "session",
chain: 1,
attempt: 2,
outcome: "Success",
})
expect(result.state).toBe(active)
expect(result.actions).toEqual([])
expect(result.response).toEqual({ _tag: "None" })
})
test("ignores duplicate and foreign settlements", () => {
const active = combinedActive(3)
const foreign = Machine.reduce(active, {
_tag: "Settled",
key: "session",
chain: 99,
attempt: 100,
outcome: "Failure",
})
const idle = Machine.reduce(Machine.initial<string>(), {
_tag: "Settled",
key: "session",
chain: 1,
attempt: 2,
outcome: "Success",
})
expect(foreign).toEqual({ state: active, actions: [], response: { _tag: "None" } })
expect(idle.actions).toEqual([])
expect(idle.state).toEqual(Machine.initial<string>())
})
test("completes the superseded chain when creating its successor", () => {
const interrupted = Machine.reduce(combinedActive(3), { _tag: "Interrupt", key: "session", seq: 2 })
const settled = Machine.reduce(interrupted.state, {
_tag: "Settled",
key: "session",
chain: 1,
attempt: 4,
outcome: "Interrupted",
})
expect(settled.actions).toContainEqual({ _tag: "CompleteChain", chain: 1 })
expect(settled.state.lanes.get("session")?.chain).not.toBe(1)
})
test("returns caller observation separately from executable actions", () => {
const result = Machine.reduce(Machine.initial<string>(), {
_tag: "Run",
key: "session",
})
expect(result.response).toEqual({ _tag: "AwaitChain", chain: 1 })
expect(result.actions).toEqual([
{
_tag: "Start",
key: "session",
chain: 1,
attempt: 2,
demand: Machine.Demand.explicit,
successor: false,
},
])
expect(result.state.nextID).toBe(3)
})
test("allocates only identities selected by each transition", () => {
const woken = Machine.reduce(Machine.initial<string>(), { _tag: "Wake", key: "session", seq: 1 })
expect(woken.state.nextID).toBe(3)
const coalesced = Machine.reduce(woken.state, { _tag: "Wake", key: "session", seq: 2 })
expect(coalesced.state.nextID).toBe(3)
const explicit = Machine.reduce(coalesced.state, { _tag: "Run", key: "session" })
expect(explicit.state.nextID).toBe(4)
const joined = Machine.reduce(explicit.state, { _tag: "Run", key: "session" })
expect(joined.state.nextID).toBe(4)
const continued = Machine.reduce(joined.state, {
_tag: "Settled",
key: "session",
chain: 1,
attempt: 2,
outcome: "Success",
})
expect(continued.state.nextID).toBe(5)
expect(continued.state.lanes.get("session")?.attempt).toBe(4)
})
test("interrupting an active combined demand preserves its newer wake as an advisory successor", () => {
const active = combinedActive(3)
const interrupted = Machine.reduce(active, { _tag: "Interrupt", key: "session", seq: 2 })
expect(interrupted.actions).toEqual([{ _tag: "Interrupt", attempt: 4 }])
expect(interrupted.state.lanes.get("session")?.pending).toEqual(Machine.Demand.wake(3))
const settled = Machine.reduce(interrupted.state, {
_tag: "Settled",
key: "session",
chain: 1,
attempt: 4,
outcome: "Interrupted",
})
expect(settled.state.lanes.get("session")?.current).toEqual(Machine.Demand.wake(3))
expect(settled.actions).toContainEqual({
_tag: "Start",
key: "session",
chain: 5,
attempt: 6,
demand: Machine.Demand.wake(3),
successor: true,
})
})
test("interrupting an active combined demand suppresses its wake at the boundary", () => {
const active = combinedActive(2)
const interrupted = Machine.reduce(active, { _tag: "Interrupt", key: "session", seq: 2 })
const pending = interrupted.state.lanes.get("session")?.pending
expect(pending).toBeDefined()
if (pending === undefined) throw new Error("Missing stopping lane")
expect(Machine.Demand.nonEmpty(pending)).toBeFalse()
const settled = Machine.reduce(interrupted.state, {
_tag: "Settled",
key: "session",
chain: 1,
attempt: 4,
outcome: "Interrupted",
})
expect(settled.state.lanes.has("session")).toBeFalse()
expect(settled.actions.some((action) => action._tag === "Start")).toBeFalse()
})
})
function combinedActive(seq: number) {
const woken = Machine.reduce(Machine.initial<string>(), {
_tag: "Wake",
key: "session",
seq: 1,
})
const explicit = Machine.reduce(woken.state, {
_tag: "Run",
key: "session",
})
const pending = Machine.reduce(explicit.state, {
_tag: "Wake",
key: "session",
seq,
})
const active = Machine.reduce(pending.state, {
_tag: "Settled",
key: "session",
chain: 1,
attempt: 2,
outcome: "Success",
})
return active.state
}

75
packages/core/test/session-run-coordinator.test.ts
View file

@ -1,10 +1,34 @@
import { describe, expect } from "bun:test"
import { Cause, Deferred, Effect, Exit, Fiber, Layer, Scope } from "effect"
import { describe, expect, test } from "bun:test"
import { Cause, Deferred, Effect, Equal, Exit, Fiber, Layer, Scope } from "effect"
import { SessionRunCoordinator } from "@opencode-ai/core/session/run-coordinator"
import { testEffect } from "./lib/effect"
const it = testEffect(Layer.empty)
describe("SessionRunCoordinator.Demand", () => {
const Demand = SessionRunCoordinator.Demand
test("combines associatively with an identity", () => {
const left = Demand.run.combine(Demand.wake(1))
const right = Demand.wake().combine(Demand.wake(3))
expect(Equal.equals(Demand.empty.combine(left), left)).toBeTrue()
expect(Equal.equals(left.combine(Demand.empty), left)).toBeTrue()
expect(Equal.equals(left.combine(right), right.combine(left))).toBeTrue()
expect(
Equal.equals(left.combine(right).combine(Demand.wake(2)), left.combine(right.combine(Demand.wake(2)))),
).toBeTrue()
})
test("keeps only sequenced wakes newer than an interrupt boundary", () => {
const demand = Demand.run.combine(Demand.wake()).combine(Demand.wake(3))
expect(Equal.equals(demand.afterBoundary(2), Demand.wake(3))).toBeTrue()
expect(Equal.equals(demand.afterBoundary(3), Demand.empty)).toBeTrue()
expect(Equal.equals(demand.afterBoundary(), Demand.empty)).toBeTrue()
})
})
describe("SessionRunCoordinator", () => {
it.effect("joins concurrent resumes for one key", () =>
Effect.scoped(
@ -461,6 +485,53 @@ describe("SessionRunCoordinator", () => {
),
)
it.effect("does not let an interrupted attempt completion settle its successor", () =>
Effect.scoped(
Effect.gen(function* () {
const firstStarted = yield* Deferred.make<void>()
const cleanupStarted = yield* Deferred.make<void>()
const cleanupGate = yield* Deferred.make<void>()
const secondStarted = yield* Deferred.make<void>()
const secondGate = yield* Deferred.make<void>()
const idleSettled = yield* Deferred.make<void>()
let runs = 0
const coordinator = yield* SessionRunCoordinator.make({
drain: () =>
Effect.sync(() => ++runs).pipe(
Effect.flatMap((run) =>
run === 1
? Deferred.succeed(firstStarted, undefined).pipe(
Effect.andThen(Effect.never),
Effect.onInterrupt(() =>
Deferred.succeed(cleanupStarted, undefined).pipe(Effect.andThen(Deferred.await(cleanupGate))),
),
)
: Deferred.succeed(secondStarted, undefined).pipe(Effect.andThen(Deferred.await(secondGate))),
),
),
})
yield* coordinator.wake("session", 1)
yield* Deferred.await(firstStarted)
const interrupt = yield* coordinator.interrupt("session", 2).pipe(Effect.forkChild)
yield* Deferred.await(cleanupStarted)
yield* coordinator.wake("session", 3)
yield* Deferred.succeed(cleanupGate, undefined)
yield* Fiber.join(interrupt)
yield* Deferred.await(secondStarted)
const idle = yield* coordinator
.awaitIdle("session")
.pipe(Effect.ensuring(Deferred.succeed(idleSettled, undefined)), Effect.forkChild)
yield* Effect.yieldNow
expect(yield* Deferred.isDone(idleSettled)).toBeFalse()
yield* Deferred.succeed(secondGate, undefined)
yield* Fiber.join(idle)
expect(runs).toBe(2)
}),
),
)
it.effect("interrupts an explicit run queued before the interruption request", () =>
Effect.scoped(
Effect.gen(function* () {