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

fix(core): preserve event routing semantics

Browse source
This commit is contained in:
Kit Langton 2026-06-25 12:41:44 -04:00
commit 603b334b7f
13 changed files with 277 additions and 129 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -1,12 +1,12 @@
export * as EventV2 from "./event"
import { Cause, Context, Effect, Layer, Option, Predicate, PubSub, Schema, Stream } from "effect"
import { Cause, Context, Effect, Layer, Option, PubSub, Schema, Stream } from "effect"
import { Event } from "@opencode-ai/schema/event"
import type {
Data,
Definition,
DurableDefinition,
LivePublishedPayload,
LiveDefinition,
Payload,
UncommittedPayload,
} from "@opencode-ai/schema/event"
@ -20,14 +20,7 @@ import { Durable } from "@opencode-ai/schema/durable-event-manifest"
export const ID = Event.ID
export type ID = import("@opencode-ai/schema/event").ID
export type {
Data,
Definition,
DurableDefinition,
LivePublishedPayload,
Payload,
UncommittedPayload,
} from "@opencode-ai/schema/event"
export type { Data, Definition, Payload } from "@opencode-ai/schema/event"
export type Subscriber<D extends Definition = Definition> = (event: Payload<D>) => Effect.Effect<void>
export type Unsubscribe = Effect.Effect<void>
@ -80,13 +73,10 @@ export interface Interface {
) => Effect.Effect<Payload<D>>
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<DurableDefinition>>
readonly durable: (input: { readonly aggregateID: string; readonly after?: number }) => Stream.Stream<Payload>
/** @deprecated Use `all()` and consume the returned stream. */
readonly listen: (listener: Subscriber) => Effect.Effect<Unsubscribe>
readonly project: <D extends DurableDefinition>(definition: D, projector: Subscriber<D>) => Effect.Effect<void>
readonly project: <D extends Definition>(definition: D, projector: Subscriber<D>) => Effect.Effect<void>
readonly replay: (
event: SerializedEvent,
options?: { readonly publish?: boolean; readonly ownerID?: string; readonly strictOwner?: boolean },
@ -140,6 +130,23 @@ export const layerWith = (options?: LayerOptions) =>
}),
)
function commitDurableEvent(
definition: DurableDefinition,
event: UncommittedPayload<DurableDefinition>,
input: undefined,
commit?: (seq: number) => Effect.Effect<void>,
): Effect.Effect<Payload<DurableDefinition>>
function commitDurableEvent(
definition: DurableDefinition,
event: UncommittedPayload<DurableDefinition>,
input: {
readonly seq: number
readonly aggregateID: string
readonly ownerID?: string
readonly strictOwner?: boolean
},
commit?: (seq: number) => Effect.Effect<void>,
): Effect.Effect<Payload<DurableDefinition> | undefined>
function commitDurableEvent(
definition: DurableDefinition,
event: UncommittedPayload<DurableDefinition>,
@ -154,7 +161,7 @@ export const layerWith = (options?: LayerOptions) =>
return Effect.gen(function* () {
const durable = definition.durable
if (durable) {
const aggregateID = Predicate.isReadonlyObject(event.data) ? event.data[durable.aggregate] : undefined
const aggregateID = (event.data as Record<string, unknown>)[durable.aggregate]
if (typeof aggregateID !== "string") {
yield* Effect.die(
new InvalidDurableEventError({
@ -185,14 +192,10 @@ export const layerWith = (options?: LayerOptions) =>
.get()
.pipe(Effect.orDie)
const latest = row?.seq ?? -1
const encoded = Schema.encodeUnknownSync(definition.data)(event.data)
if (!Predicate.isReadonlyObject(encoded))
return yield* Effect.die(
new InvalidDurableEventError({
type: event.type,
message: "Expected durable event data to encode as an object",
}),
)
const encoded = Schema.encodeUnknownSync(definition.data)(event.data) as Record<
string,
unknown
>
if (input?.strictOwner && row?.ownerID && row.ownerID !== input.ownerID) {
yield* Effect.die(
new InvalidDurableEventError({
@ -329,11 +332,6 @@ export const layerWith = (options?: LayerOptions) =>
})
}
const isPayload =
<D extends Definition>(definition: D) =>
(event: Payload): event is Payload<D> =>
Schema.is(definition)(event)
function publish<D extends Definition>(
definition: D,
data: Data<D>,
@ -356,10 +354,6 @@ export const layerWith = (options?: LayerOptions) =>
data,
}
const committed = yield* commitDurableEvent(definition, event, undefined, options?.commit)
if (!committed)
return yield* Effect.die(
new InvalidDurableEventError({ type: event.type, message: "New durable event was not committed" }),
)
yield* notify(committed, true)
return committed
}
@ -370,7 +364,7 @@ export const layerWith = (options?: LayerOptions) =>
message: "Local commit hooks require a durable event",
}),
)
const event: LivePublishedPayload = {
const event: Payload<LiveDefinition> = {
id: options?.id ?? ID.create(),
...(options?.metadata ? { metadata: options.metadata } : {}),
type: definition.type,
@ -467,7 +461,7 @@ export const layerWith = (options?: LayerOptions) =>
const subscribe = <D extends Definition>(definition: D): Stream.Stream<Payload<D>> =>
Stream.unwrap(getOrCreate(definition).pipe(Effect.map((pubsub) => Stream.fromPubSub(pubsub)))).pipe(
Stream.filter(isPayload(definition)),
Stream.map((event) => event as Payload<D>),
)
const streamAll = (): Stream.Stream<Payload> => Stream.fromPubSub(pubsub.all)
@ -529,10 +523,7 @@ export const layerWith = (options?: LayerOptions) =>
return subscription
})
const durable = (input: {
readonly aggregateID: string
readonly after?: number
}): Stream.Stream<Payload<DurableDefinition>> =>
const durable = (input: { readonly aggregateID: string; readonly after?: number }): Stream.Stream<Payload> =>
Stream.unwrap(
Effect.gen(function* () {
const wakes = yield* subscribeDurable(input.aggregateID)
@ -540,7 +531,7 @@ export const layerWith = (options?: LayerOptions) =>
const read = Effect.suspend(() => readAfter(input.aggregateID, sequence)).pipe(
Effect.tap((events) =>
Effect.sync(() => {
sequence = events.at(-1)?.durable.seq ?? sequence
sequence = events.at(-1)?.durable?.seq ?? sequence
}),
),
)
@ -562,14 +553,10 @@ export const layerWith = (options?: LayerOptions) =>
})
})
const project = <D extends DurableDefinition>(definition: D, projector: Subscriber<D>): Effect.Effect<void> =>
const project = <D extends Definition>(definition: D, projector: Subscriber<D>): Effect.Effect<void> =>
Effect.sync(() => {
const list = projectors.get(definition.type) ?? []
list.push((event) =>
isPayload(definition)(event)
? projector(event)
: Effect.die(`Published event ${event.type} does not match its definition`),
)
list.push((event) => projector(event as Payload<D>))
projectors.set(definition.type, list)
})

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

@ -1,6 +1,5 @@
import { castDraft, produce, type WritableDraft } from "immer"
import { Effect, Match } from "effect"
import type { UncommittedPayload } from "@opencode-ai/schema/event"
import { Effect } from "effect"
import { SessionEvent } from "./event"
import { SessionMessage } from "./message"
@ -8,8 +7,6 @@ export type MemoryState = {
messages: SessionMessage.Message[]
}
export type Input = UncommittedPayload<(typeof SessionEvent.Definitions)[number]>
export interface Adapter {
readonly getCurrentAssistant: () => Effect.Effect<SessionMessage.Assistant | undefined>
readonly getAssistant: (messageID: SessionMessage.ID) => Effect.Effect<SessionMessage.Assistant | undefined>
@ -78,7 +75,7 @@ export function memory(state: MemoryState): Adapter {
}
}
export function update(adapter: Adapter, event: Input) {
export function update(adapter: Adapter, event: SessionEvent.Event) {
type DraftAssistant = WritableDraft<SessionMessage.Assistant>
type DraftTool = WritableDraft<SessionMessage.AssistantTool>
type DraftText = WritableDraft<SessionMessage.AssistantText>
@ -101,8 +98,8 @@ export function update(adapter: Adapter, event: Input) {
if (assistant) yield* adapter.updateAssistant(produce(assistant, recipe))
})
return Match.value(event).pipe(
Match.discriminatorsExhaustive("type")({
return Effect.gen(function* () {
yield* SessionEvent.All.match(event, {
"session.next.agent.switched": (event) => {
return adapter.appendMessage(
SessionMessage.AgentSwitched.make({
@ -391,8 +388,8 @@ export function update(adapter: Adapter, event: Input) {
"session.next.revert.staged": () => Effect.void,
"session.next.revert.cleared": () => Effect.void,
"session.next.revert.committed": () => Effect.void,
}),
)
})
})
}
export * as SessionMessageUpdater from "./message-updater"

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

@ -138,6 +138,50 @@ describe("EventV2", () => {
}),
)
it.effect("preserves same-type projector routing across durable versions", () =>
Effect.gen(function* () {
const events = yield* EventV2.Service
const historical = EventV2.define({
type: "test.projector-version",
durable: { version: 1, aggregate: "id" },
schema: { id: Schema.String },
})
const current = EventV2.define({
type: "test.projector-version",
durable: { version: 2, aggregate: "id" },
schema: { id: Schema.String },
})
const received = new Array<EventV2.Payload>()
yield* events.project(historical, (event) => Effect.sync(() => received.push(event)))
const published = yield* events.publish(current, { id: "aggregate" })
expect(received).toEqual([published])
}),
)
it.effect("preserves same-type subscription routing across durable versions", () =>
Effect.gen(function* () {
const events = yield* EventV2.Service
const historical = EventV2.define({
type: "test.subscription-version",
durable: { version: 1, aggregate: "id" },
schema: { id: Schema.String },
})
const current = EventV2.define({
type: "test.subscription-version",
durable: { version: 2, aggregate: "id" },
schema: { id: Schema.String },
})
const fiber = yield* events.subscribe(historical).pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped)
yield* Effect.yieldNow
const published = yield* events.publish(current, { id: "aggregate" })
expect(Array.from(yield* Fiber.join(fiber))).toEqual([published])
}),
)
it.effect("publishes to typed and wildcard subscriptions", () =>
Effect.gen(function* () {
const events = yield* EventV2.Service

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

@ -17,7 +17,14 @@ const capture = () => {
const events = EventV2.Service.of({
publish: (definition, data) =>
Effect.sync(() => {
const event = { id: EventV2.ID.create(), type: definition.type, data } as EventV2.Payload<typeof definition>
const event = {
id: EventV2.ID.create(),
type: definition.type,
...(definition.durable
? { durable: { aggregateID: sessionID, seq: published.length, version: definition.durable.version } }
: {}),
data,
} as EventV2.Payload<typeof definition>
published.push({
type: definition.durable
? EventV2.versionedType(definition.type, definition.durable.version)