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 59c1f89d39
21 changed files with 618 additions and 98 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
@ -185,6 +186,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(
@ -197,6 +199,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)
@ -340,10 +343,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 }
})