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 605fb0b6c5
20 changed files with 583 additions and 95 deletions
Download patch file Download diff file Expand all files Collapse all files

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

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