ceos/opencode
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

chore(core): merge dev into session subpath fix

Browse source
This commit is contained in:
Kit Langton 2026-06-22 13:19:06 -04:00
commit 2d550aac46
22 changed files with 220 additions and 1384 deletions
Download patch file Download diff file Expand all files Collapse all files

38
packages/core/src/session.ts
View file

@ -1,7 +1,7 @@
export * as SessionV2 from "./session"
export * from "./session/schema"
import { Cause, DateTime, Effect, Layer, Schema, Context, Stream } from "effect"
import { DateTime, Effect, Layer, Schema, Context, Stream } from "effect"
import { and, asc, desc, eq, gt, gte, like, lt, or, type SQL } from "drizzle-orm"
import { ProjectV2 } from "./project"
import { WorkspaceV2 } from "./workspace"
@ -25,7 +25,6 @@ import { fromRow } from "./session/info"
import { SessionRunner } from "./session/runner/index"
import { SessionStore } from "./session/store"
import { SessionExecution } from "./session/execution"
import { logFailure } from "./session/logging"
import { MessageDecodeError } from "./session/error"
import { SessionEvent } from "./session/event"
import { SessionInput } from "./session/input"
@ -168,20 +167,6 @@ export const layer = Layer.effect(
const store = yield* SessionStore.Service
const decodeMessage = Schema.decodeUnknownEffect(SessionMessage.Message)
const isDurableSessionEvent = Schema.is(SessionEvent.Durable)
const scope = yield* Effect.scope
const enqueueWake = (admitted: SessionInput.Admitted) =>
execution.wake(admitted.sessionID, admitted.admittedSeq).pipe(
Effect.tapCause((cause) =>
Cause.hasInterruptsOnly(cause)
? Effect.void
: logFailure("Failed to wake Session", admitted.sessionID, cause),
),
Effect.ignore,
Effect.forkIn(scope, { startImmediately: true }),
Effect.asVoid,
)
const decode = (row: typeof SessionMessageTable.$inferSelect) =>
decodeMessage({ ...row.data, id: row.id, type: row.type }).pipe(
Effect.mapError(
@ -349,10 +334,6 @@ export const layer = Layer.effect(
Effect.uninterruptible(
Effect.gen(function* () {
yield* result.get(input.sessionID)
const returnPrompt = Effect.fnUntraced(function* (admitted: SessionInput.Admitted) {
if (input.resume !== false) yield* enqueueWake(admitted)
return admitted
}, Effect.uninterruptible)
const messageID = input.id ?? SessionMessage.ID.create()
const delivery = input.delivery ?? "steer"
const expected = { sessionID: input.sessionID, messageID, prompt: input.prompt, delivery }
@ -370,7 +351,8 @@ export const layer = Layer.effect(
)
if (!SessionInput.equivalent(admitted, expected))
return yield* new PromptConflictError({ sessionID: input.sessionID, messageID })
return yield* returnPrompt(admitted)
if (input.resume !== false) yield* execution.wake(admitted.sessionID)
return admitted
}),
),
),
@ -411,19 +393,7 @@ export const layer = Layer.effect(
yield* execution.resume(sessionID)
}),
interrupt: Effect.fn("V2Session.interrupt")((sessionID) =>
Effect.uninterruptible(
Effect.gen(function* () {
const session = yield* store.get(sessionID)
if (!session) return yield* execution.interrupt(sessionID)
const event = yield* events.publish(SessionEvent.InterruptRequested, {
sessionID,
timestamp: yield* DateTime.now,
})
if (event.durable === undefined)
return yield* Effect.die("Interrupt request event is missing aggregate sequence")
yield* execution.interrupt(sessionID, event.durable.seq)
}),
),
Effect.uninterruptible(execution.interrupt(sessionID)),
),
})

8
packages/core/src/session/event.ts
View file

@ -118,13 +118,6 @@ export namespace PromptLifecycle {
export type Promoted = typeof Promoted.Type
}
export const InterruptRequested = EventV2.define({
type: "session.next.interrupt.requested",
...options,
schema: Base,
})
export type InterruptRequested = typeof InterruptRequested.Type
export const ContextUpdated = EventV2.define({
type: "session.next.context.updated",
...options,
@ -475,7 +468,6 @@ const DurableDefinitions = [
Prompted,
PromptLifecycle.Admitted,
PromptLifecycle.Promoted,
InterruptRequested,
ContextUpdated,
Synthetic,
Shell.Started,

8
packages/core/src/session/execution.ts
View file

@ -5,12 +5,12 @@ import { SessionRunner } from "./runner/index"
import { SessionSchema } from "./schema"
export interface Interface {
/** Explicitly drain one Session, making at least one provider attempt. */
/** Starts execution while idle or joins the active execution. */
readonly resume: (sessionID: SessionSchema.ID) => Effect.Effect<void, SessionRunner.RunError>
/** Schedule a drain after durable work is recorded. Repeated wakeups may coalesce. */
readonly wake: (sessionID: SessionSchema.ID, seq?: number) => Effect.Effect<void, SessionRunner.RunError>
/** Registers newly recorded work. Repeated wakeups may coalesce. */
readonly wake: (sessionID: SessionSchema.ID) => Effect.Effect<void>
/** Interrupt active work owned by this process. Idle interruption is a no-op. */
readonly interrupt: (sessionID: SessionSchema.ID, seq?: number) => Effect.Effect<void>
readonly interrupt: (sessionID: SessionSchema.ID) => Effect.Effect<void>
}
/** Routes execution from a Session ID to the runner owned by that Session's Location. */

15
packages/core/src/session/execution/local.ts
View file

@ -1,11 +1,10 @@
import { Effect, Layer } from "effect"
import { Cause, Effect, Layer } from "effect"
import { LocationServiceMap } from "../../location-layer"
import { SessionRunCoordinator } from "../run-coordinator"
import { SessionRunner } from "../runner"
import { SessionSchema } from "../schema"
import { SessionStore } from "../store"
import { SessionExecution } from "../execution"
import { logFailure } from "../logging"
/** Current-process routing for implicit-local Locations. Future remote placement belongs here. */
export const layer = Layer.effect(
@ -13,15 +12,19 @@ export const layer = Layer.effect(
Effect.gen(function* () {
const store = yield* SessionStore.Service
const locations = yield* LocationServiceMap
const coordinator = yield* SessionRunCoordinator.make<SessionSchema.ID, void, SessionRunner.RunError>({
drain: Effect.fnUntraced(function* (sessionID: SessionSchema.ID, mode) {
const coordinator = yield* SessionRunCoordinator.make<SessionSchema.ID, SessionRunner.RunError>({
drain: Effect.fnUntraced(function* (sessionID: SessionSchema.ID, force) {
const session = yield* store.get(sessionID)
if (!session) return yield* Effect.die(`Session not found: ${sessionID}`)
return yield* SessionRunner.Service.use((runner) => runner.run({ sessionID, force: mode === "run" })).pipe(
return yield* SessionRunner.Service.use((runner) => runner.run({ sessionID, force })).pipe(
Effect.provide(locations.get(session.location)),
Effect.tapCause((cause) =>
Cause.hasInterruptsOnly(cause)
? Effect.void
: Effect.logError("Failed to drain Session", cause).pipe(Effect.annotateLogs({ sessionID })),
),
)
}),
onFailure: (sessionID, cause) => logFailure("Failed to drain Session", sessionID, cause),
})
return SessionExecution.Service.of({

8
packages/core/src/session/logging.ts
View file

@ -1,8 +0,0 @@
import { Cause, Effect } from "effect"
import { SessionSchema } from "./schema"
export const logFailure = (
message: "Failed to drain Session" | "Failed to wake Session",
sessionID: SessionSchema.ID,
cause: Cause.Cause<unknown>,
) => Effect.logError(message, cause).pipe(Effect.annotateLogs({ sessionID }))

1
packages/core/src/session/message-updater.ts
View file

@ -138,7 +138,6 @@ export function update(adapter: Adapter, event: SessionEvent.Event) {
},
"session.next.prompt.admitted": () => Effect.void,
"session.next.prompt.promoted": () => Effect.void,
"session.next.interrupt.requested": () => Effect.void,
"session.next.context.updated": (event) =>
adapter.appendMessage(
new SessionMessage.System({

1
packages/core/src/session/projector.ts
View file

@ -399,7 +399,6 @@ export const layer = Layer.effectDiscard(
)
}),
)
yield* events.project(SessionEvent.InterruptRequested, () => Effect.void)
yield* events.project(SessionEvent.ContextUpdated, (event) => run(db, event))
yield* events.project(SessionEvent.Synthetic, (event) => run(db, event))
yield* events.project(SessionEvent.Shell.Started, (event) => run(db, event))

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

@ -1,106 +1,43 @@
export * as SessionRunCoordinator from "./run-coordinator"
import { Cause, Context, Deferred, Effect, Exit, Fiber, FiberSet, Layer, Scope } from "effect"
import { SessionRunner } from "./runner"
import { logFailure } from "./logging"
import { SessionSchema } from "./schema"
import { Deferred, Effect, Exit, Fiber, FiberSet, Scope } from "effect"
export type Mode = "run" | "wake"
/** Why one drain generation should run. Explicit runs dominate advisory wakes when demands coalesce. */
type Demand = { readonly _tag: "run" } | { readonly _tag: "wake"; readonly seq?: number }
/**
* Runs at most one drain chain per key while allowing different keys to drain concurrently.
*
* For each key:
*
* idle --run/wake--> draining --run/wake--> draining + one coalesced rerun --> idle
*
* `run` is an explicit drain request. It starts a chain or joins the current chain and
* upgrades a pending follow-up so the caller receives explicit-run semantics.
*
* `wake` reports that durable work may now be available. It starts a chain while idle or
* requests one coalesced follow-up while draining. Repeated wakes collapse together.
*
* `interrupt` stops the current ownership chain. Advisory wakes from before the interrupt
* boundary are suppressed; advisory wakes after the boundary run after cleanup.
*/
export interface Coordinator<Key, A, E> {
/** Starts or joins one explicit drain generation. */
readonly run: (key: Key) => Effect.Effect<A, E>
/** Coalesces one wake-up after durable work is recorded. */
readonly wake: (key: Key, seq?: number) => Effect.Effect<void>
/** Waits until the current ownership chain settles. */
readonly awaitIdle: (key: Key) => Effect.Effect<void, E>
/** Interrupts the active ownership chain without automatically draining pending wakes. */
readonly interrupt: (key: Key, seq?: number) => Effect.Effect<void>
/** Serializes execution for each key while allowing different keys to run concurrently. */
export interface Coordinator<Key, E> {
/** Starts execution while idle or joins the active execution. */
readonly run: (key: Key) => Effect.Effect<void, E>
/** Registers one coalesced follow-up after newly recorded work. */
readonly wake: (key: Key) => Effect.Effect<void>
/** Stops active execution and waits for its cleanup. */
readonly interrupt: (key: Key) => Effect.Effect<void>
}
/** One Session's process-local execution lane: one active demand and at most one coalesced follow-up. */
type Entry<A, E> = {
readonly done: Deferred.Deferred<A, E>
readonly settled: Deferred.Deferred<Exit.Exit<A, E>>
current: Demand
pending?: Demand
explicitWaiter?: Deferred.Deferred<A, E>
interruptSeq?: number
type Entry<E> = {
readonly done: Deferred.Deferred<void, E>
owner?: Fiber.Fiber<void, never>
pendingWake: boolean
stopping: boolean
}
/** Combines follow-up demand: runs dominate, while wakes retain the newest durable admission sequence. */
const coalesce = (left: Demand | undefined, right: Demand): Demand => {
if (left?._tag === "run" || right._tag === "run") return { _tag: "run" }
return { _tag: "wake", seq: maxSeq(left?.seq, right.seq) }
}
const maxSeq = (left: number | undefined, right: number | undefined) => {
if (left === undefined) return right
if (right === undefined) return left
return Math.max(left, right)
}
/** Constructs a scoped coordinator. Every in-memory transition is synchronous. */
export const make = <Key, A, E>(options: {
readonly drain: (key: Key, mode: Mode) => Effect.Effect<A, E>
readonly onFailure?: (key: Key, cause: Cause.Cause<E>) => Effect.Effect<void>
}): Effect.Effect<Coordinator<Key, A, E>, never, Scope.Scope> =>
export const make = <Key, E>(options: {
readonly drain: (key: Key, force: boolean) => Effect.Effect<void, E>
}): Effect.Effect<Coordinator<Key, E>, never, Scope.Scope> =>
Effect.gen(function* () {
const active = new Map<Key, Entry<A, E>>()
const interruptSeq = new Map<Key, number>()
const report = yield* FiberSet.makeRuntime<never, void, never>()
const active = new Map<Key, Entry<E>>()
const fork = yield* FiberSet.makeRuntime<never, void, never>()
const shutdown = Deferred.makeUnsafe<void>()
let closed = false
yield* Effect.addFinalizer(() =>
Effect.sync(() => {
closed = true
Deferred.doneUnsafe(shutdown, Effect.void)
active.clear()
interruptSeq.clear()
}),
)
const makeEntry = (current: Demand, explicitWaiter?: Deferred.Deferred<A, E>): Entry<A, E> => ({
done: Deferred.makeUnsafe<A, E>(),
settled: Deferred.makeUnsafe<Exit.Exit<A, E>>(),
current,
explicitWaiter,
const makeEntry = (): Entry<E> => ({
done: Deferred.makeUnsafe<void, E>(),
pendingWake: false,
stopping: false,
})
const start = (key: Key, entry: Entry<A, E>, demand: Demand, successor = false) => {
const start = (key: Key, entry: Entry<E>, force: boolean, successor = false) => {
const ready = Deferred.makeUnsafe<void>()
const drain = Effect.suspend(() => options.drain(key, demand._tag))
// Initial work retains immediate-start behavior but cannot run before ownership is published.
// Observer-started successors yield once so synchronous drains cannot recurse on the JS stack.
const owner = fork(
(successor
? Effect.yieldNow.pipe(Effect.andThen(drain))
: Deferred.await(ready).pipe(Effect.andThen(drain))
).pipe(
Effect.onExit((exit) => Effect.sync(() => settle(key, entry, demand, exit))),
(successor ? Effect.yieldNow : Deferred.await(ready)).pipe(
Effect.andThen(Effect.suspend(() => options.drain(key, force))),
Effect.onExit((exit) => Effect.sync(() => settle(key, entry, exit))),
Effect.exit,
Effect.asVoid,
),
@ -109,176 +46,57 @@ export const make = <Key, A, E>(options: {
if (!successor) Deferred.doneUnsafe(ready, Effect.void)
}
const settle = (key: Key, entry: Entry<A, E>, demand: Demand, exit: Exit.Exit<A, E>) => {
if (closed) {
Deferred.doneUnsafe(entry.done, exit)
Deferred.doneUnsafe(entry.settled, Effect.succeed(exit))
return
}
if (demand._tag === "run" && entry.explicitWaiter !== undefined) {
Deferred.doneUnsafe(entry.explicitWaiter, exit)
entry.explicitWaiter = undefined
}
if (entry.stopping && demand._tag === "wake" && entry.explicitWaiter !== undefined) {
Deferred.doneUnsafe(entry.explicitWaiter, exit)
entry.explicitWaiter = undefined
}
if (active.get(key) !== entry) {
Deferred.doneUnsafe(entry.done, exit)
Deferred.doneUnsafe(entry.settled, Effect.succeed(exit))
return
}
if (exit._tag === "Success" && !entry.stopping) {
if (entry.pending !== undefined) {
const pending = entry.pending
entry.pending = undefined
entry.current = pending
start(key, entry, pending, true)
return
}
active.delete(key)
Deferred.doneUnsafe(entry.done, exit)
Deferred.doneUnsafe(entry.settled, Effect.succeed(exit))
const settle = (key: Key, entry: Entry<E>, exit: Exit.Exit<void, E>) => {
if (Exit.isSuccess(exit) && !entry.stopping && entry.pendingWake) {
entry.pendingWake = false
start(key, entry, false, true)
return
}
const successor = entry.pending !== undefined ? makeEntry(entry.pending, entry.explicitWaiter) : undefined
const successor = entry.pendingWake ? makeEntry() : undefined
if (successor === undefined) active.delete(key)
else active.set(key, successor)
if (successor !== undefined) start(key, successor, successor.current, true)
Deferred.doneUnsafe(entry.done, exit)
Deferred.doneUnsafe(entry.settled, Effect.succeed(exit))
if (
exit._tag === "Failure" &&
!(entry.stopping && Cause.hasInterruptsOnly(exit.cause)) &&
demand._tag === "wake" &&
options.onFailure !== undefined
) {
report(Effect.suspend(() => options.onFailure!(key, exit.cause)))
else {
active.set(key, successor)
start(key, successor, false, true)
}
Deferred.doneUnsafe(entry.done, exit)
}
const wake = (key: Key, seq?: number) =>
Effect.sync(() => {
if (closed) return
if (!isAfterInterrupt(key, seq)) return
const run = (key: Key): Effect.Effect<void, E> =>
Effect.uninterruptibleMask((restore) => {
const entry = active.get(key)
if (entry !== undefined) {
if (!acceptsWake(entry, seq)) return
entry.pending = coalesce(entry.pending, { _tag: "wake", seq })
if (entry.stopping) return restore(Deferred.await(entry.done).pipe(Effect.andThen(run(key))))
return restore(Deferred.await(entry.done))
}
const next = makeEntry()
active.set(key, next)
start(key, next, true)
return restore(Deferred.await(next.done))
})
const wake = (key: Key) =>
Effect.sync(() => {
const entry = active.get(key)
if (entry !== undefined) {
entry.pendingWake = true
return
}
const next = makeEntry({ _tag: "wake", seq })
const next = makeEntry()
active.set(key, next)
start(key, next, next.current)
start(key, next, false)
})
const awaitIdle = (key: Key): Effect.Effect<void, E> =>
Effect.gen(function* () {
let firstFailure: Cause.Cause<E> | undefined
while (!closed) {
const entry = active.get(key)
if (entry === undefined) break
const exit = yield* Effect.raceFirst(
Deferred.await(entry.settled),
Deferred.await(shutdown).pipe(Effect.as(Exit.void)),
)
if (closed) break
if (exit._tag === "Failure" && firstFailure === undefined) firstFailure = exit.cause
}
if (firstFailure !== undefined) return yield* Effect.failCause(firstFailure)
})
const interrupt = (key: Key, seq?: number): Effect.Effect<void> =>
const interrupt = (key: Key): Effect.Effect<void> =>
Effect.suspend(() => {
const entry = active.get(key)
const latest = interruptSeq.get(key)
if (seq !== undefined && latest !== undefined && seq <= latest)
return entry?.stopping && entry.owner !== undefined ? Fiber.interrupt(entry.owner) : Effect.void
if (seq !== undefined) interruptSeq.set(key, seq)
if (entry?.owner === undefined) return Effect.void
if (
seq !== undefined &&
entry.current._tag === "wake" &&
entry.current.seq !== undefined &&
entry.current.seq > seq
)
return Effect.void
if (entry.stopping) {
entry.interruptSeq = maxSeq(entry.interruptSeq, seq)
suppressPendingAtOrBefore(entry, seq)
return Fiber.interrupt(entry.owner)
}
entry.stopping = true
entry.interruptSeq = seq
suppressPendingAtOrBefore(entry, seq)
entry.pendingWake = false
return Fiber.interrupt(entry.owner)
})
return { run, wake, awaitIdle, interrupt }
function run(key: Key): Effect.Effect<A, E> {
return Effect.uninterruptibleMask((restore) => {
if (closed) return Effect.interrupt
const entry = active.get(key)
if (entry !== undefined) {
if (entry.stopping) {
return restore(Deferred.await(entry.settled).pipe(Effect.andThen(run(key))))
}
if (entry.current._tag === "wake") {
entry.pending = coalesce(entry.pending, { _tag: "run" })
entry.explicitWaiter ??= Deferred.makeUnsafe<A, E>()
return restore(awaitRun(entry.explicitWaiter))
}
return restore(awaitRun(entry.done))
}
const next = makeEntry({ _tag: "run" })
active.set(key, next)
start(key, next, next.current)
return restore(awaitRun(next.done))
})
}
function awaitRun(done: Deferred.Deferred<A, E>): Effect.Effect<A, E> {
return Effect.raceFirst(Deferred.await(done), Deferred.await(shutdown).pipe(Effect.andThen(Effect.interrupt)))
}
function acceptsWake(entry: Entry<A, E>, seq: number | undefined) {
return !entry.stopping || (entry.interruptSeq !== undefined && seq !== undefined && seq > entry.interruptSeq)
}
function isAfterInterrupt(key: Key, seq: number | undefined) {
const latest = interruptSeq.get(key)
return latest === undefined || (seq !== undefined && seq > latest)
}
function suppressPendingAtOrBefore(entry: Entry<A, E>, seq: number | undefined) {
if (
entry.pending?._tag === "wake" &&
seq !== undefined &&
entry.pending.seq !== undefined &&
entry.pending.seq > seq
)
return
entry.pending = undefined
}
return { run, wake, interrupt }
})
export interface Interface extends Coordinator<SessionSchema.ID, void, SessionRunner.RunError> {}
export class Service extends Context.Service<Service, Interface>()("@opencode/v2/SessionRunCoordinator") {}
export const layer = Layer.effect(
Service,
SessionRunner.Service.pipe(
Effect.flatMap((runner) =>
make<SessionSchema.ID, void, SessionRunner.RunError>({
drain: (sessionID, mode) => runner.run({ sessionID, force: mode === "run" }),
onFailure: (sessionID, cause) => logFailure("Failed to drain Session", sessionID, cause),
}),
),
Effect.map(Service.of),
),
)

2
packages/core/src/session/runner/index.ts
View file

@ -21,7 +21,7 @@ export interface Interface {
/** Drains eligible durable work. Explicit runs perform one provider attempt even when no work is eligible. */
readonly run: (input: {
readonly sessionID: SessionSchema.ID
readonly force?: boolean
readonly force: boolean
}) => Effect.Effect<void, RunError>
}

6
packages/core/src/session/runner/llm.ts
View file

@ -346,14 +346,14 @@ export const layer = Layer.effect(
const run = Effect.fn("SessionRunner.run")(function* (input: {
readonly sessionID: SessionSchema.ID
readonly force?: boolean
readonly force: boolean
}) {
const hasSteer = yield* SessionInput.hasPending(db, input.sessionID, "steer")
const hasQueue = hasSteer ? false : yield* SessionInput.hasPending(db, input.sessionID, "queue")
if (input.force !== true && !hasSteer && !hasQueue) return
if (!input.force && !hasSteer && !hasQueue) return
yield* failInterruptedTools(input.sessionID)
let promotion: SessionInput.Delivery | undefined = hasSteer ? "steer" : hasQueue ? "queue" : undefined
let openActivity = input.force === true || hasSteer || hasQueue
let openActivity = input.force || hasSteer || hasQueue
while (openActivity) {
let needsContinuation = true
for (let step = 1; needsContinuation; step++) {

30
packages/core/test/session-logging.test.ts
View file

@ -1,30 +0,0 @@
import { describe, expect, test } from "bun:test"
import { Cause, Effect, Logger } from "effect"
import { logFailure } from "@opencode-ai/core/session/logging"
import { SessionSchema } from "@opencode-ai/core/session/schema"
describe("Session logging", () => {
for (const message of ["Failed to drain Session", "Failed to wake Session"] as const) {
test(`renders the cause for ${message}`, async () => {
const entries: Array<ReturnType<typeof Logger.formatStructured.log>> = []
const logger = Logger.formatStructured.pipe(
Logger.map((entry): void => {
entries.push(entry)
}),
)
await logFailure(
message,
SessionSchema.ID.make("session-123"),
Cause.fail({ _tag: "SessionFailure", detail: { code: "nested-code" } }),
).pipe(Effect.provide(Logger.layer([logger])), Effect.runPromise)
expect(entries).toHaveLength(1)
expect(entries[0]?.message).toBe(message)
expect(entries[0]?.annotations).toEqual({ sessionID: "session-123" })
expect(entries[0]?.cause).toContain("SessionFailure")
expect(entries[0]?.cause).toContain("nested-code")
expect(entries[0]?.cause).not.toContain("[Object")
})
}
})

36
packages/core/test/session-prompt.test.ts
View file

@ -20,9 +20,7 @@ import { testEffect } from "./lib/effect"
const executionCalls: SessionV2.ID[] = []
const interruptCalls: SessionV2.ID[] = []
const interruptSeqs: Array<number | undefined> = []
const wakeCalls: SessionV2.ID[] = []
const wakeSeqs: Array<number | undefined> = []
const execution = Layer.succeed(
SessionExecution.Service,
SessionExecution.Service.of({
@ -30,15 +28,13 @@ const execution = Layer.succeed(
Effect.sync(() => {
executionCalls.push(sessionID)
}),
interrupt: (sessionID, seq) =>
interrupt: (sessionID) =>
Effect.sync(() => {
interruptCalls.push(sessionID)
interruptSeqs.push(seq)
}),
wake: (sessionID, seq) =>
wake: (sessionID) =>
Effect.sync(() => {
wakeCalls.push(sessionID)
wakeSeqs.push(seq)
}),
}),
)
@ -109,15 +105,6 @@ const eventCount = (type: string) =>
),
)
const interruptEvent = Database.Service.use(({ db }) =>
db
.select()
.from(EventTable)
.where(eq(EventTable.type, "session.next.interrupt.requested.1"))
.get()
.pipe(Effect.orDie),
)
describe("SessionV2.prompt", () => {
it.effect("delegates execution continuation through SessionExecution", () =>
Effect.gen(function* () {
@ -131,19 +118,14 @@ describe("SessionV2.prompt", () => {
}),
)
it.effect("delegates interruption through SessionExecution", () =>
it.effect("delegates process-local interruption through SessionExecution", () =>
Effect.gen(function* () {
yield* setup
const session = yield* SessionV2.Service
interruptCalls.length = 0
interruptSeqs.length = 0
yield* session.interrupt(sessionID)
expect(interruptCalls).toEqual([sessionID])
expect(interruptSeqs).toHaveLength(1)
expect(typeof interruptSeqs[0]).toBe("number")
expect(yield* eventCount("session.next.interrupt.requested.1")).toBe(1)
expect(yield* interruptEvent).toMatchObject({ aggregate_id: sessionID, seq: interruptSeqs[0] })
expect(yield* session.messages({ sessionID })).toEqual([])
}),
)
@ -152,11 +134,9 @@ describe("SessionV2.prompt", () => {
Effect.gen(function* () {
const session = yield* SessionV2.Service
interruptCalls.length = 0
interruptSeqs.length = 0
yield* session.interrupt(SessionV2.ID.make("ses_missing"))
expect(interruptCalls).toEqual([SessionV2.ID.make("ses_missing")])
expect(interruptSeqs).toEqual([undefined])
}),
)
@ -515,13 +495,11 @@ describe("SessionV2.prompt", () => {
const session = yield* SessionV2.Service
executionCalls.length = 0
wakeCalls.length = 0
wakeSeqs.length = 0
const admitted = yield* session.prompt({ sessionID, prompt: new Prompt({ text: "Run by default" }) })
yield* session.prompt({ sessionID, prompt: new Prompt({ text: "Run by default" }) })
expect(executionCalls).toEqual([])
expect(wakeCalls).toEqual([sessionID])
expect(wakeSeqs).toEqual([admitted.admittedSeq])
}),
)
@ -531,9 +509,8 @@ describe("SessionV2.prompt", () => {
const session = yield* SessionV2.Service
executionCalls.length = 0
wakeCalls.length = 0
wakeSeqs.length = 0
const admitted = yield* session.prompt({
yield* session.prompt({
sessionID,
prompt: new Prompt({ text: "Run explicitly" }),
resume: true,
@ -541,7 +518,6 @@ describe("SessionV2.prompt", () => {
expect(executionCalls).toEqual([])
expect(wakeCalls).toEqual([sessionID])
expect(wakeSeqs).toEqual([admitted.admittedSeq])
}),
)
@ -551,13 +527,11 @@ describe("SessionV2.prompt", () => {
const session = yield* SessionV2.Service
executionCalls.length = 0
wakeCalls.length = 0
wakeSeqs.length = 0
yield* session.prompt({ sessionID, prompt: new Prompt({ text: "Do not run" }), resume: false })
expect(executionCalls).toEqual([])
expect(wakeCalls).toEqual([])
expect(wakeSeqs).toEqual([])
}),
)
})

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

File diff suppressed because it is too large Load diff

25
packages/core/test/session-runner-recorded.test.ts
View file

@ -16,6 +16,7 @@ import { Prompt } from "@opencode-ai/core/session/prompt"
import { SessionProjector } from "@opencode-ai/core/session/projector"
import { SessionExecution } from "@opencode-ai/core/session/execution"
import { SessionRunCoordinator } from "@opencode-ai/core/session/run-coordinator"
import { SessionRunner } from "@opencode-ai/core/session/runner"
import * as SessionRunnerLLM from "@opencode-ai/core/session/runner/llm"
import { SessionRunnerModel } from "@opencode-ai/core/session/runner/model"
import { ToolRegistry } from "@opencode-ai/core/tool/registry"
@ -83,19 +84,20 @@ const runner = SessionRunnerLLM.defaultLayer.pipe(
Layer.provide(referenceGuidance),
Layer.provide(config),
)
const coordinator = SessionRunCoordinator.layer.pipe(Layer.provide(runner))
const execution = Layer.effect(
SessionExecution.Service,
SessionRunCoordinator.Service.pipe(
Effect.map((coordinator) =>
SessionExecution.Service.of({
resume: coordinator.run,
wake: coordinator.wake,
interrupt: coordinator.interrupt,
}),
),
),
).pipe(Layer.provide(coordinator))
Effect.gen(function* () {
const sessionRunner = yield* SessionRunner.Service
const coordinator = yield* SessionRunCoordinator.make<SessionV2.ID, SessionRunner.RunError>({
drain: (sessionID, force) => sessionRunner.run({ sessionID, force }),
})
return SessionExecution.Service.of({
resume: coordinator.run,
wake: coordinator.wake,
interrupt: coordinator.interrupt,
})
}),
).pipe(Layer.provide(runner))
const sessions = SessionV2.layer.pipe(
Layer.provide(EventV2.defaultLayer),
Layer.provide(Database.defaultLayer),
@ -120,7 +122,6 @@ const it = testEffect(
skillGuidance,
config,
runner,
coordinator,
execution,
sessions,
),

31
packages/core/test/session-runner.test.ts
View file

@ -244,19 +244,20 @@ const runner = SessionRunnerLLM.layer.pipe(
Layer.provide(referenceGuidance),
Layer.provide(config),
)
const coordinator = SessionRunCoordinator.layer.pipe(Layer.provide(runner))
const execution = Layer.effect(
SessionExecution.Service,
SessionRunCoordinator.Service.pipe(
Effect.map((coordinator) =>
SessionExecution.Service.of({
resume: coordinator.run,
wake: coordinator.wake,
interrupt: coordinator.interrupt,
}),
),
),
).pipe(Layer.provide(coordinator))
Effect.gen(function* () {
const sessionRunner = yield* SessionRunner.Service
const coordinator = yield* SessionRunCoordinator.make<SessionV2.ID, SessionRunner.RunError>({
drain: (sessionID, force) => sessionRunner.run({ sessionID, force }),
})
return SessionExecution.Service.of({
resume: coordinator.run,
wake: coordinator.wake,
interrupt: coordinator.interrupt,
})
}),
).pipe(Layer.provide(runner))
const sessions = SessionV2.layer.pipe(
Layer.provide(EventV2.defaultLayer),
Layer.provide(Database.defaultLayer),
@ -283,7 +284,6 @@ const it = testEffect(
skillGuidance,
config,
runner,
coordinator,
execution,
sessions,
),
@ -681,7 +681,6 @@ describe("SessionRunnerLLM", () => {
systemUnavailable = false
yield* session.prompt({ id: messageID, sessionID, prompt: new Prompt({ text: "First" }) })
yield* (yield* SessionRunCoordinator.Service).awaitIdle(sessionID)
expect(requests).toHaveLength(1)
expect(requests[0]?.messages.map((message) => message.role)).toEqual(["user"])
@ -2161,7 +2160,7 @@ describe("SessionRunnerLLM", () => {
expect(requests).toHaveLength(2)
expect(userTexts(requests[1]!)).toEqual(["Start working", "First steer", "Second steer"])
yield* (yield* SessionRunCoordinator.Service).wake(sessionID)
yield* (yield* SessionExecution.Service).wake(sessionID)
yield* Effect.yieldNow
expect(requests).toHaveLength(2)
}),
@ -2367,7 +2366,7 @@ describe("SessionRunnerLLM", () => {
})
requests.length = 0
yield* (yield* SessionRunCoordinator.Service).wake(sessionID)
yield* (yield* SessionExecution.Service).wake(sessionID)
yield* Effect.yieldNow
expect(requests).toHaveLength(1)
@ -2417,7 +2416,7 @@ describe("SessionRunnerLLM", () => {
LLMEvent.finish({ reason: "stop" }),
]
yield* (yield* SessionRunCoordinator.Service).wake(sessionID)
yield* (yield* SessionExecution.Service).wake(sessionID)
while (requests.length === 0) yield* Effect.yieldNow
expect(userTexts(requests[0]!)).toEqual(["Recover promoted input"])