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

feat(sdk): unify session event stream

Browse source
This commit is contained in:
Kit Langton 2026-06-25 23:26:07 -04:00
commit 20975b9434
20 changed files with 583 additions and 95 deletions
Download patch file Download diff file Expand all files Collapse all files

60
packages/core/src/event.ts
View file

@ -1,9 +1,9 @@
export * as EventV2 from "./event"
import { Cause, Context, Effect, Layer, Option, PubSub, Schema, Stream } from "effect"
import { Cause, Context, Effect, Layer, Option, PubSub, Queue, Schema, Scope, Stream } from "effect"
import { Event } from "@opencode-ai/schema/event"
import type { Data, Definition, Payload } from "@opencode-ai/schema/event"
import { and, asc, eq, gt } from "drizzle-orm"
import { and, asc, eq, gt, lte } from "drizzle-orm"
import { Database } from "./database/database"
import { EventSequenceTable, EventTable } from "./event/sql"
import { Location } from "./location"
@ -67,6 +67,15 @@ export interface Interface {
readonly subscribe: <D extends Definition>(definition: D) => Stream.Stream<Payload<D>>
readonly all: () => Stream.Stream<Payload>
readonly durable: (input: { readonly aggregateID: string; readonly after?: number }) => Stream.Stream<Payload>
readonly observeAggregate: (input: {
readonly aggregateID: string
readonly after?: number
readonly live: (event: Payload) => boolean
}) => Effect.Effect<
{ readonly replay: ReadonlyArray<Payload>; readonly updates: Stream.Stream<Payload> },
never,
Scope.Scope
>
/** @deprecated Use `all()` and consume the returned stream. */
readonly listen: (listener: Subscriber) => Effect.Effect<Unsubscribe>
readonly project: <D extends Definition>(definition: D, projector: Subscriber<D>) => Effect.Effect<void>
@ -472,13 +481,19 @@ export const layerWith = (options?: LayerOptions) =>
}
}
const readAfter = (aggregateID: string, after: number) =>
const readAfter = (aggregateID: string, after: number, through?: number) =>
(options?.beforeAggregateRead?.(aggregateID) ?? Effect.void).pipe(
Effect.andThen(
db
.select()
.from(EventTable)
.where(and(eq(EventTable.aggregate_id, aggregateID), gt(EventTable.seq, after)))
.where(
and(
eq(EventTable.aggregate_id, aggregateID),
gt(EventTable.seq, after),
through === undefined ? undefined : lte(EventTable.seq, through),
),
)
.orderBy(asc(EventTable.seq))
.all(),
),
@ -537,6 +552,42 @@ export const layerWith = (options?: LayerOptions) =>
}),
)
const observeAggregate: Interface["observeAggregate"] = (input) =>
Effect.gen(function* () {
type Signal = { readonly _tag: "durable" } | { readonly _tag: "live"; readonly event: Payload }
const signals = yield* Queue.dropping<Signal, Cause.Done>(256)
const unsubscribe = yield* listen((event) =>
Effect.sync(() => {
if (event.durable?.aggregateID === input.aggregateID) {
// Durable payloads are only wakeups; a full queue already contains work that will drain the database.
Queue.offerUnsafe(signals, { _tag: "durable" })
return
}
if (!input.live(event)) return
if (!Queue.offerUnsafe(signals, { _tag: "live", event })) Queue.endUnsafe(signals)
}),
)
yield* Effect.addFinalizer(() => unsubscribe.pipe(Effect.andThen(Queue.shutdown(signals))))
const cutoff = yield* latestSequence(db, input.aggregateID)
const replay = yield* readAfter(input.aggregateID, input.after ?? -1, cutoff)
let sequence = cutoff
const drain = Effect.suspend(() => readAfter(input.aggregateID, sequence)).pipe(
Effect.tap((events) =>
Effect.sync(() => {
sequence = events.at(-1)?.durable?.seq ?? sequence
}),
),
)
const updates = Stream.fromQueue(signals).pipe(
Stream.mapEffect((signal) =>
drain.pipe(Effect.map((events) => (signal._tag === "live" ? [...events, signal.event] : events))),
),
Stream.flattenIterable,
)
return { replay, updates }
})
const listen = (listener: Subscriber): Effect.Effect<Unsubscribe> =>
Effect.sync(() => {
listeners.push(listener)
@ -558,6 +609,7 @@ export const layerWith = (options?: LayerOptions) =>
subscribe,
all: streamAll,
durable,
observeAggregate,
listen,
project,
replay,

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

@ -106,6 +106,7 @@ export type MessageNotFoundError = SessionRevert.MessageNotFoundError
export type Error = NotFoundError | MessageDecodeError | OperationUnavailableError | PromptConflictError
export interface Interface {
readonly active: Effect.Effect<ReadonlySet<SessionSchema.ID>>
readonly list: (input?: ListInput) => Effect.Effect<SessionSchema.Info[]>
readonly create: (input: CreateInput) => Effect.Effect<SessionSchema.Info>
readonly get: (sessionID: SessionSchema.ID) => Effect.Effect<SessionSchema.Info, NotFoundError>
@ -128,7 +129,7 @@ export interface Interface {
readonly events: (input: {
sessionID: SessionSchema.ID
after?: number
}) => Stream.Stream<SessionEvent.DurableEvent, NotFoundError>
}) => Stream.Stream<SessionEvent.StreamEvent, NotFoundError>
readonly switchAgent: (input: { sessionID: SessionSchema.ID; agent: string }) => Effect.Effect<void, NotFoundError>
readonly switchModel: (input: {
sessionID: SessionSchema.ID
@ -186,6 +187,7 @@ export const layer = Layer.unwrap(
const locations = yield* LocationServiceMap
const decodeMessage = Schema.decodeUnknownEffect(SessionMessage.Message)
const isDurableSessionEvent = Schema.is(SessionEvent.Durable)
const sessionEventTypes = new Set<string>(SessionEvent.Definitions.map((definition) => definition.type))
const decode = (row: typeof SessionMessageTable.$inferSelect) =>
decodeMessage({ ...row.data, id: row.id, type: row.type }).pipe(
Effect.mapError(
@ -198,6 +200,7 @@ export const layer = Layer.unwrap(
)
const result = Service.of({
active: execution.active,
create: Effect.fn("V2Session.create")(function* (input) {
const sessionID = input.id ?? SessionSchema.ID.create()
const recorded = yield* store.get(sessionID)
@ -341,10 +344,46 @@ export const layer = Layer.unwrap(
}),
events: (input) =>
Stream.unwrap(
result
.get(input.sessionID)
.pipe(Effect.as(events.durable({ aggregateID: input.sessionID, after: input.after }))),
).pipe(Stream.filter((event): event is SessionEvent.DurableEvent => isDurableSessionEvent(event))),
Effect.gen(function* () {
yield* result.get(input.sessionID)
const activity = yield* execution.activity(input.sessionID)
const observed = yield* events.observeAggregate({
aggregateID: input.sessionID,
after: input.after,
live: (event) =>
event.durable === undefined &&
sessionEventTypes.has(event.type) &&
typeof event.data === "object" &&
event.data !== null &&
"sessionID" in event.data &&
event.data.sessionID === input.sessionID,
})
const initialActivity: SessionEvent.Activity = {
id: EventV2.ID.create(),
type: SessionEvent.Activity.type,
data: { sessionID: input.sessionID, active: yield* activity.snapshot },
}
const replay = observed.replay.filter((event): event is SessionEvent.DurableEvent =>
isDurableSessionEvent(event),
)
const updates = observed.updates.pipe(
Stream.filter((event): event is SessionEvent.Event => Schema.is(SessionEvent.All)(event)),
)
const activityChanges = activity.changes.pipe(
Stream.map(
(active): SessionEvent.Activity => ({
id: EventV2.ID.create(),
type: SessionEvent.Activity.type,
data: { sessionID: input.sessionID, active },
}),
),
)
return Stream.fromIterable(replay).pipe(
Stream.concat(Stream.make(initialActivity)),
Stream.concat(Stream.merge(updates, activityChanges)),
)
}),
),
prompt: Effect.fn("V2Session.prompt")((input) =>
Effect.uninterruptible(
Effect.gen(function* () {

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

@ -1,12 +1,20 @@
export * as SessionExecution from "./execution"
import { Context, Effect, Layer } from "effect"
import { Context, Effect, Layer, Scope, Stream } from "effect"
import { SessionRunner } from "./runner/index"
import { SessionSchema } from "./schema"
export interface Interface {
/** Snapshots active execution owned by this process. */
readonly active: Effect.Effect<ReadonlySet<SessionSchema.ID>>
/** Observes foreground ownership with an authoritative initial snapshot. */
readonly activity: (
sessionID: SessionSchema.ID,
) => Effect.Effect<
{ readonly snapshot: Effect.Effect<boolean>; readonly changes: Stream.Stream<boolean> },
never,
Scope.Scope
>
/** Starts execution while idle or joins the active execution. */
readonly resume: (sessionID: SessionSchema.ID) => Effect.Effect<void, SessionRunner.RunError>
/** Registers newly recorded work. Repeated wakeups may coalesce. */
@ -23,6 +31,7 @@ export const noopLayer = Layer.succeed(
Service,
Service.of({
active: Effect.succeed(new Set()),
activity: () => Effect.succeed({ snapshot: Effect.succeed(false), changes: Stream.never }),
resume: () => Effect.void,
wake: () => Effect.void,
interrupt: () => Effect.void,

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

@ -29,6 +29,7 @@ export const layer = Layer.effect(
return SessionExecution.Service.of({
active: coordinator.active,
activity: coordinator.activity,
interrupt: coordinator.interrupt,
resume: coordinator.run,
wake: coordinator.wake,

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

@ -1,11 +1,19 @@
export * as SessionRunCoordinator from "./run-coordinator"
import { Deferred, Effect, Exit, Fiber, FiberSet, Scope } from "effect"
import { Cause, Deferred, Effect, Exit, Fiber, FiberSet, Queue, Scope, Stream } from "effect"
export interface Activity {
/** Discards earlier transitions and snapshots current ownership. */
readonly snapshot: Effect.Effect<boolean>
readonly changes: Stream.Stream<boolean>
}
/** Serializes execution for each key while allowing different keys to run concurrently. */
export interface Coordinator<Key, E> {
/** Snapshots keys with an execution owned by this coordinator. */
readonly active: Effect.Effect<ReadonlySet<Key>>
/** Registers transition observation before taking its authoritative snapshot. */
readonly activity: (key: Key) => Effect.Effect<Activity, never, Scope.Scope>
/** 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. */
@ -26,6 +34,7 @@ export const make = <Key, E>(options: {
}): Effect.Effect<Coordinator<Key, E>, never, Scope.Scope> =>
Effect.gen(function* () {
const active = new Map<Key, Entry<E>>()
const activityObservers = new Map<Key, Set<(active: boolean) => void>>()
const fork = yield* FiberSet.makeRuntime<never, void, never>()
const makeEntry = (): Entry<E> => ({
@ -34,6 +43,10 @@ export const make = <Key, E>(options: {
stopping: false,
})
const notifyActivity = (key: Key, value: boolean) => {
for (const observer of activityObservers.get(key) ?? []) observer(value)
}
const start = (key: Key, entry: Entry<E>, force: boolean, successor = false) => {
const ready = Deferred.makeUnsafe<void>()
const owner = fork(
@ -56,8 +69,10 @@ export const make = <Key, E>(options: {
}
const successor = entry.pendingWake ? makeEntry() : undefined
if (successor === undefined) active.delete(key)
else {
if (successor === undefined) {
active.delete(key)
notifyActivity(key, false)
} else {
active.set(key, successor)
start(key, successor, false, true)
}
@ -74,6 +89,7 @@ export const make = <Key, E>(options: {
const next = makeEntry()
active.set(key, next)
notifyActivity(key, true)
start(key, next, true)
return restore(Deferred.await(next.done))
})
@ -88,6 +104,7 @@ export const make = <Key, E>(options: {
const next = makeEntry()
active.set(key, next)
notifyActivity(key, true)
start(key, next, false)
})
@ -100,5 +117,30 @@ export const make = <Key, E>(options: {
return Fiber.interrupt(entry.owner)
})
return { active: Effect.sync(() => new Set(active.keys())), run, wake, interrupt }
const activity = (key: Key) =>
Effect.gen(function* () {
const queue = yield* Queue.dropping<boolean, Cause.Done>(256)
const observer = (value: boolean) => {
if (!Queue.offerUnsafe(queue, value)) Queue.endUnsafe(queue)
}
yield* Effect.acquireRelease(
Effect.sync(() => {
const observers = activityObservers.get(key) ?? new Set()
observers.add(observer)
activityObservers.set(key, observers)
}),
() =>
Effect.sync(() => {
const observers = activityObservers.get(key)
observers?.delete(observer)
if (observers?.size === 0) activityObservers.delete(key)
}).pipe(Effect.andThen(Queue.shutdown(queue))),
)
return {
snapshot: Queue.clear(queue).pipe(Effect.andThen(Effect.sync(() => active.has(key)))),
changes: Stream.fromQueue(queue),
}
})
return { active: Effect.sync(() => new Set(active.keys())), activity, run, wake, interrupt }
})

70
packages/core/test/event.test.ts
View file

@ -418,6 +418,76 @@ describe("EventV2", () => {
}),
)
it.effect("observes a fixed replay cutoff and delivers commits during replay as updates", () =>
Effect.gen(function* () {
const readStarted = yield* Deferred.make<void>()
const continueRead = yield* Deferred.make<void>()
let pause = true
const eventLayer = EventV2.layerWith({
beforeAggregateRead: () =>
pause
? Deferred.succeed(readStarted, undefined).pipe(Effect.andThen(Deferred.await(continueRead)))
: Effect.void,
}).pipe(Layer.provide(Database.defaultLayer))
yield* Effect.gen(function* () {
const events = yield* EventV2.Service
const aggregateID = Session.ID.create()
const observer = yield* events.observeAggregate({ aggregateID, live: () => false }).pipe(Effect.forkScoped)
yield* Deferred.await(readStarted)
pause = false
yield* events.publish(DurableMessage, durableData(aggregateID, "after cutoff"))
yield* Deferred.succeed(continueRead, undefined)
const observed = yield* Fiber.join(observer)
const update = yield* observed.updates.pipe(Stream.take(1), Stream.runCollect)
expect(observed.replay).toEqual([])
expect(Array.from(update).map((event) => [event.durable?.seq, event.data])).toEqual([
[0, durableData(aggregateID, "after cutoff")],
])
}).pipe(Effect.provide(Layer.mergeAll(Database.defaultLayer, eventLayer)))
}),
)
it.effect("drains causally preceding durable rows before a live payload", () =>
Effect.gen(function* () {
const events = yield* EventV2.Service
const aggregateID = Session.ID.create()
const observed = yield* events.observeAggregate({
aggregateID,
live: (event) => event.type === Message.type,
})
const updates = yield* observed.updates.pipe(Stream.take(2), Stream.runCollect, Effect.forkScoped)
yield* events.publish(DurableMessage, durableData(aggregateID, "start"))
const live = yield* events.publish(Message, { text: "delta" })
expect(Array.from(yield* Fiber.join(updates))).toEqual([
expect.objectContaining({ durable: expect.objectContaining({ seq: 0 }) }),
live,
])
}),
)
it.effect("coalesces saturated observer signals without losing durable rows", () =>
Effect.gen(function* () {
const events = yield* EventV2.Service
const aggregateID = Session.ID.create()
const count = 300
const observed = yield* events.observeAggregate({ aggregateID, live: () => false })
for (let index = 0; index < count; index++) {
yield* events.publish(DurableMessage, durableData(aggregateID, String(index)))
}
const updates = yield* observed.updates.pipe(Stream.take(count), Stream.runCollect)
expect(Array.from(updates, (event) => event.durable?.seq)).toEqual(
Array.from({ length: count }, (_, index) => index),
)
}),
)
it.effect("coalesces durable aggregate wakes while draining every committed event", () =>
Effect.gen(function* () {
const events = yield* EventV2.Service

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

@ -1,5 +1,5 @@
import { describe, expect } from "bun:test"
import { DateTime, Effect, Fiber, Layer, Stream } from "effect"
import { DateTime, Deferred, Effect, Fiber, Layer, Stream } from "effect"
import { eq } from "drizzle-orm"
import { Database } from "@opencode-ai/core/database/database"
import { EventV2 } from "@opencode-ai/core/event"
@ -23,10 +23,29 @@ const executionCalls: SessionV2.ID[] = []
const interruptCalls: SessionV2.ID[] = []
const wakeCalls: SessionV2.ID[] = []
const activeSessions = new Set<SessionV2.ID>()
let activityObserver: ((active: boolean) => void) | undefined
let activityState = false
const setActivity = (active: boolean) =>
Effect.sync(() => {
activityState = active
activityObserver?.(active)
})
const execution = Layer.succeed(
SessionExecution.Service,
SessionExecution.Service.of({
active: Effect.sync(() => new Set(activeSessions)),
activity: () =>
Effect.sync(() => {
activityState = false
activityObserver = undefined
return {
attach: (observer: (active: boolean) => void) =>
Effect.sync(() => {
activityObserver = observer
return activityState
}),
}
}),
resume: (sessionID) =>
Effect.sync(() => {
executionCalls.push(sessionID)
@ -179,7 +198,7 @@ describe("SessionV2.prompt", () => {
const session = yield* SessionV2.Service
const events = yield* EventV2.Service
const { db } = yield* Database.Service
const fiber = yield* session.events({ sessionID }).pipe(Stream.take(4), Stream.runCollect, Effect.forkScoped)
const fiber = yield* session.events({ sessionID }).pipe(Stream.take(5), Stream.runCollect, Effect.forkScoped)
yield* Effect.yieldNow
yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "First" }), resume: false })
@ -188,6 +207,7 @@ describe("SessionV2.prompt", () => {
const streamed = Array.from(yield* Fiber.join(fiber))
expect(streamed.map((event) => [event.durable?.seq, event.type])).toEqual([
[undefined, "session.activity"],
[0, "session.next.prompt.admitted"],
[1, "session.next.prompt.admitted"],
[2, "session.next.prompted"],
@ -196,10 +216,44 @@ describe("SessionV2.prompt", () => {
expect(
Array.from(
yield* session
.events({ sessionID, after: streamed[0]!.durable?.seq })
.pipe(Stream.take(1), Stream.runCollect),
.events({ sessionID, after: streamed[2]?.durable?.seq })
.pipe(Stream.take(3), Stream.runCollect),
).map((event) => [event.durable?.seq, event.type]),
).toEqual([[1, "session.next.prompt.admitted"]])
).toEqual([
[2, "session.next.prompted"],
[3, "session.next.prompted"],
[undefined, "session.activity"],
])
}),
)
it.effect("replays history, emits activity, then fences live deltas behind durable starts", () =>
Effect.gen(function* () {
yield* setup
const session = yield* SessionV2.Service
const events = yield* EventV2.Service
const assistantMessageID = SessionMessage.ID.create()
yield* events.publish(SessionEvent.Text.Started, {
sessionID,
assistantMessageID,
textID: "text-1",
timestamp: yield* DateTime.now,
})
const streamed = yield* session.events({ sessionID }).pipe(Stream.take(3), Stream.runCollect, Effect.forkScoped)
yield* Effect.yieldNow
yield* events.publish(SessionEvent.Text.Delta, {
sessionID,
assistantMessageID,
textID: "text-1",
delta: "hello",
timestamp: yield* DateTime.now,
})
expect(Array.from(yield* Fiber.join(streamed)).map((event) => event.type)).toEqual([
"session.next.text.started",
"session.activity",
"session.next.text.delta",
])
}),
)

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

@ -99,6 +99,27 @@ describe("SessionRunCoordinator", () => {
),
)
it.effect("reflects activity races in the snapshot or subsequent transitions", () =>
Effect.scoped(
Effect.gen(function* () {
const started = yield* Deferred.make<void>()
const gate = yield* Deferred.make<void>()
const coordinator = yield* SessionRunCoordinator.make({
drain: () => Deferred.succeed(started, undefined).pipe(Effect.andThen(Deferred.await(gate))),
})
const activity = yield* coordinator.activity("session")
const run = yield* coordinator.run("session").pipe(Effect.forkChild)
yield* Deferred.await(started)
const transitions = new Array<boolean>()
expect(yield* activity.attach((active) => transitions.push(active))).toBeTrue()
yield* Deferred.succeed(gate, undefined)
yield* Fiber.join(run)
expect(transitions).toEqual([false])
}),
),
)
it.effect("cleans active executions after failure and defect", () =>
Effect.scoped(
Effect.gen(function* () {

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

@ -96,6 +96,7 @@ const execution = Layer.effect(
})
return SessionExecution.Service.of({
active: coordinator.active,
activity: coordinator.activity,
resume: coordinator.run,
wake: coordinator.wake,
interrupt: coordinator.interrupt,

1
packages/core/test/session-runner-tool-events.test.ts
View file

@ -28,6 +28,7 @@ const capture = () => {
}),
subscribe: () => Stream.empty,
all: () => Stream.empty,
observeAggregate: () => Effect.die("not implemented"),
durable: () => Stream.empty,
listen: () => Effect.succeed(Effect.void),
project: () => Effect.void,

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

@ -260,6 +260,7 @@ const execution = Layer.effect(
})
return SessionExecution.Service.of({
active: coordinator.active,
activity: coordinator.activity,
resume: coordinator.run,
wake: coordinator.wake,
interrupt: coordinator.interrupt,