From 45071a8d8e293b26284ac14bd21a248fa0616c99 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Thu, 25 Jun 2026 12:04:05 -0400 Subject: [PATCH] refactor(core): prove event publication branches --- packages/core/src/event.ts | 106 +++++++++--------- packages/core/src/session/message-updater.ts | 15 ++- .../test/v2/session-message-updater.test.ts | 44 +++----- packages/schema/src/event.ts | 90 +++++++++------ 4 files changed, 132 insertions(+), 123 deletions(-) diff --git a/packages/core/src/event.ts b/packages/core/src/event.ts index d8ec020627..0b0eae26d7 100644 --- a/packages/core/src/event.ts +++ b/packages/core/src/event.ts @@ -1,13 +1,13 @@ export * as EventV2 from "./event" -import { Cause, Context, Effect, Layer, Option, PubSub, Schema, Stream } from "effect" +import { Cause, Context, Effect, Layer, Option, Predicate, PubSub, Schema, Stream } from "effect" import { Event } from "@opencode-ai/schema/event" import type { Data, Definition, DurableDefinition, + LivePublishedPayload, Payload, - PublishedPayload, UncommittedPayload, } from "@opencode-ai/schema/event" import { and, asc, eq, gt } from "drizzle-orm" @@ -24,8 +24,8 @@ export type { Data, Definition, DurableDefinition, + LivePublishedPayload, Payload, - PublishedPayload, UncommittedPayload, } from "@opencode-ai/schema/event" @@ -154,7 +154,7 @@ export const layerWith = (options?: LayerOptions) => return Effect.gen(function* () { const durable = definition.durable if (durable) { - const aggregateID = (event.data as Record)[durable.aggregate] + const aggregateID = Predicate.isReadonlyObject(event.data) ? event.data[durable.aggregate] : undefined if (typeof aggregateID !== "string") { yield* Effect.die( new InvalidDurableEventError({ @@ -185,10 +185,14 @@ export const layerWith = (options?: LayerOptions) => .get() .pipe(Effect.orDie) const latest = row?.seq ?? -1 - const encoded = Schema.encodeUnknownSync(definition.data)(event.data) as Record< - string, - unknown - > + 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", + }), + ) if (input?.strictOwner && row?.ownerID && row.ownerID !== input.ownerID) { yield* Effect.die( new InvalidDurableEventError({ @@ -304,39 +308,6 @@ export const layerWith = (options?: LayerOptions) => }) } - function publishEvent( - definition: D, - event: UncommittedPayload, - commit?: PublishOptions["commit"], - ): Effect.Effect> { - return Effect.gen(function* () { - if (!definition?.durable && commit) - return yield* Effect.die( - new InvalidDurableEventError({ - type: event.type, - message: "Local commit hooks require a durable event", - }), - ) - if (definition?.durable) { - const committed = yield* commitDurableEvent( - definition, - event as UncommittedPayload, - undefined, - 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 as PublishedPayload - } - const published = event as PublishedPayload - yield* notify(published, false) - return published - }) - } - const observe = (event: Payload, observer: (event: Payload) => Effect.Effect) => Effect.suspend(() => observer(event)).pipe( Effect.catchCauseIf( @@ -358,7 +329,17 @@ export const layerWith = (options?: LayerOptions) => }) } - function publish(definition: D, data: Data, options?: PublishOptions) { + const isPayload = + (definition: D) => + (event: Payload): event is Payload => + Schema.is(definition)(event) + + function publish( + definition: D, + data: Data, + options?: PublishOptions, + ): Effect.Effect> + function publish(definition: Definition, data: unknown, options?: PublishOptions): Effect.Effect { return Effect.gen(function* () { const serviceLocation = Option.getOrUndefined(yield* Effect.serviceOption(Location.Service)) const location = @@ -366,17 +347,38 @@ export const layerWith = (options?: LayerOptions) => (serviceLocation ? { directory: serviceLocation.directory, workspaceID: serviceLocation.workspaceID } : undefined) - return yield* publishEvent( - definition, - { + if (definition.durable) { + const event: UncommittedPayload = { id: options?.id ?? ID.create(), ...(options?.metadata ? { metadata: options.metadata } : {}), type: definition.type, ...(location ? { location } : {}), data, - } as UncommittedPayload, - options?.commit, - ) + } + 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 + } + if (options?.commit) + return yield* Effect.die( + new InvalidDurableEventError({ + type: definition.type, + message: "Local commit hooks require a durable event", + }), + ) + const event: LivePublishedPayload = { + id: options?.id ?? ID.create(), + ...(options?.metadata ? { metadata: options.metadata } : {}), + type: definition.type, + ...(location ? { location } : {}), + data, + } + yield* notify(event, false) + return event }) } @@ -465,7 +467,7 @@ export const layerWith = (options?: LayerOptions) => const subscribe = (definition: D): Stream.Stream> => Stream.unwrap(getOrCreate(definition).pipe(Effect.map((pubsub) => Stream.fromPubSub(pubsub)))).pipe( - Stream.map((event) => event as Payload), + Stream.filter(isPayload(definition)), ) const streamAll = (): Stream.Stream => Stream.fromPubSub(pubsub.all) @@ -563,7 +565,11 @@ export const layerWith = (options?: LayerOptions) => const project = (definition: D, projector: Subscriber): Effect.Effect => Effect.sync(() => { const list = projectors.get(definition.type) ?? [] - list.push((event) => projector(event as Payload)) + list.push((event) => + isPayload(definition)(event) + ? projector(event) + : Effect.die(`Published event ${event.type} does not match its definition`), + ) projectors.set(definition.type, list) }) diff --git a/packages/core/src/session/message-updater.ts b/packages/core/src/session/message-updater.ts index a97afb58b6..8f856d2288 100644 --- a/packages/core/src/session/message-updater.ts +++ b/packages/core/src/session/message-updater.ts @@ -1,5 +1,6 @@ import { castDraft, produce, type WritableDraft } from "immer" -import { Effect } from "effect" +import { Effect, Match } from "effect" +import type { UncommittedPayload } from "@opencode-ai/schema/event" import { SessionEvent } from "./event" import { SessionMessage } from "./message" @@ -7,6 +8,8 @@ export type MemoryState = { messages: SessionMessage.Message[] } +export type Input = UncommittedPayload<(typeof SessionEvent.Definitions)[number]> + export interface Adapter { readonly getCurrentAssistant: () => Effect.Effect readonly getAssistant: (messageID: SessionMessage.ID) => Effect.Effect @@ -75,7 +78,7 @@ export function memory(state: MemoryState): Adapter { } } -export function update(adapter: Adapter, event: SessionEvent.Event) { +export function update(adapter: Adapter, event: Input) { type DraftAssistant = WritableDraft type DraftTool = WritableDraft type DraftText = WritableDraft @@ -98,8 +101,8 @@ export function update(adapter: Adapter, event: SessionEvent.Event) { if (assistant) yield* adapter.updateAssistant(produce(assistant, recipe)) }) - return Effect.gen(function* () { - yield* SessionEvent.All.match(event, { + return Match.value(event).pipe( + Match.discriminatorsExhaustive("type")({ "session.next.agent.switched": (event) => { return adapter.appendMessage( SessionMessage.AgentSwitched.make({ @@ -388,8 +391,8 @@ export function update(adapter: Adapter, event: SessionEvent.Event) { "session.next.revert.staged": () => Effect.void, "session.next.revert.cleared": () => Effect.void, "session.next.revert.committed": () => Effect.void, - }) - }) + }), + ) } export * as SessionMessageUpdater from "./message-updater" diff --git a/packages/opencode/test/v2/session-message-updater.test.ts b/packages/opencode/test/v2/session-message-updater.test.ts index 56478a4d1d..5611f6ed70 100644 --- a/packages/opencode/test/v2/session-message-updater.test.ts +++ b/packages/opencode/test/v2/session-message-updater.test.ts @@ -5,16 +5,9 @@ import { SessionID } from "../../src/session/schema" import { EventV2 } from "@opencode-ai/core/event" import { ModelV2 } from "@opencode-ai/core/model" import { ProviderV2 } from "@opencode-ai/core/provider" -import { SessionEvent } from "@opencode-ai/core/session/event" import { SessionMessageUpdater } from "@opencode-ai/core/session/message-updater" import { SessionMessage } from "@opencode-ai/core/session/message" -function durable(sessionID: SessionID, seq?: number): { aggregateID: SessionID; seq: number; version: 1 } -function durable(sessionID: SessionID, seq: number, version: 2): { aggregateID: SessionID; seq: number; version: 2 } -function durable(sessionID: SessionID, seq = 0, version: 1 | 2 = 1) { - return { aggregateID: sessionID, seq, version } -} - test.skip("step snapshots carry over to assistant messages", () => { const state: SessionMessageUpdater.MemoryState = { messages: [] } const sessionID = SessionID.make("session") @@ -23,7 +16,6 @@ test.skip("step snapshots carry over to assistant messages", () => { Effect.runSync( SessionMessageUpdater.update(SessionMessageUpdater.memory(state), { id: EventV2.ID.create(), - durable: durable(sessionID), type: "session.next.step.started", data: { sessionID, @@ -37,7 +29,7 @@ test.skip("step snapshots carry over to assistant messages", () => { }, snapshot: "before", }, - } satisfies SessionEvent.Event), + } satisfies SessionMessageUpdater.Input), ) expect(state.messages).toEqual([]) @@ -45,7 +37,6 @@ test.skip("step snapshots carry over to assistant messages", () => { Effect.runSync( SessionMessageUpdater.update(SessionMessageUpdater.memory(state), { id: EventV2.ID.create(), - durable: durable(sessionID, 1, 2), type: "session.next.step.ended", data: { sessionID, @@ -61,7 +52,7 @@ test.skip("step snapshots carry over to assistant messages", () => { }, snapshot: "after", }, - } satisfies SessionEvent.Event), + } satisfies SessionMessageUpdater.Input), ) expect(state.messages[0]?.type).toBe("assistant") @@ -78,7 +69,6 @@ test.skip("text ended populates assistant text content", () => { Effect.runSync( SessionMessageUpdater.update(SessionMessageUpdater.memory(state), { id: EventV2.ID.create(), - durable: durable(sessionID), type: "session.next.step.started", data: { sessionID, @@ -91,13 +81,12 @@ test.skip("text ended populates assistant text content", () => { variant: ModelV2.VariantID.make("default"), }, }, - } satisfies SessionEvent.Event), + } satisfies SessionMessageUpdater.Input), ) Effect.runSync( SessionMessageUpdater.update(SessionMessageUpdater.memory(state), { id: EventV2.ID.create(), - durable: durable(sessionID, 1), type: "session.next.text.started", data: { sessionID, @@ -105,13 +94,12 @@ test.skip("text ended populates assistant text content", () => { timestamp: DateTime.makeUnsafe(2), textID: "text-1", }, - } satisfies SessionEvent.Event), + } satisfies SessionMessageUpdater.Input), ) Effect.runSync( SessionMessageUpdater.update(SessionMessageUpdater.memory(state), { id: EventV2.ID.create(), - durable: durable(sessionID, 2), type: "session.next.text.ended", data: { sessionID, @@ -120,7 +108,7 @@ test.skip("text ended populates assistant text content", () => { textID: "text-1", text: "hello assistant", }, - } satisfies SessionEvent.Event), + } satisfies SessionMessageUpdater.Input), ) expect(state.messages[0]?.type).toBe("assistant") @@ -137,7 +125,6 @@ test.skip("tool completion stores completed timestamp", () => { Effect.runSync( SessionMessageUpdater.update(SessionMessageUpdater.memory(state), { id: EventV2.ID.create(), - durable: durable(sessionID), type: "session.next.step.started", data: { sessionID, @@ -150,13 +137,12 @@ test.skip("tool completion stores completed timestamp", () => { variant: ModelV2.VariantID.make("default"), }, }, - } satisfies SessionEvent.Event), + } satisfies SessionMessageUpdater.Input), ) Effect.runSync( SessionMessageUpdater.update(SessionMessageUpdater.memory(state), { id: EventV2.ID.create(), - durable: durable(sessionID, 1), type: "session.next.tool.input.started", data: { sessionID, @@ -165,13 +151,12 @@ test.skip("tool completion stores completed timestamp", () => { callID, name: "bash", }, - } satisfies SessionEvent.Event), + } satisfies SessionMessageUpdater.Input), ) Effect.runSync( SessionMessageUpdater.update(SessionMessageUpdater.memory(state), { id: EventV2.ID.create(), - durable: durable(sessionID, 2), type: "session.next.tool.called", data: { sessionID, @@ -182,13 +167,12 @@ test.skip("tool completion stores completed timestamp", () => { input: { command: "pwd" }, provider: { executed: true, metadata: { fake: { source: "provider" } } }, }, - } satisfies SessionEvent.Event), + } satisfies SessionMessageUpdater.Input), ) Effect.runSync( SessionMessageUpdater.update(SessionMessageUpdater.memory(state), { id: EventV2.ID.create(), - durable: durable(sessionID, 3), type: "session.next.tool.success", data: { sessionID, @@ -199,7 +183,7 @@ test.skip("tool completion stores completed timestamp", () => { content: [{ type: "text", text: "/tmp" }], provider: { executed: true, metadata: { fake: { status: "done" } } }, }, - } satisfies SessionEvent.Event), + } satisfies SessionMessageUpdater.Input), ) expect(state.messages[0]?.type).toBe("assistant") @@ -219,7 +203,6 @@ test("compaction events reduce to compaction message only when completed", () => Effect.runSync( SessionMessageUpdater.update(SessionMessageUpdater.memory(state), { id, - durable: durable(sessionID), type: "session.next.compaction.started", data: { sessionID, @@ -227,7 +210,7 @@ test("compaction events reduce to compaction message only when completed", () => timestamp: DateTime.makeUnsafe(1), reason: "auto", }, - } satisfies SessionEvent.Event), + } satisfies SessionMessageUpdater.Input), ) expect(state.messages).toEqual([]) @@ -242,7 +225,7 @@ test("compaction events reduce to compaction message only when completed", () => timestamp: DateTime.makeUnsafe(2), text: "hello ", }, - } satisfies SessionEvent.Event), + } satisfies SessionMessageUpdater.Input), ) Effect.runSync( @@ -255,13 +238,12 @@ test("compaction events reduce to compaction message only when completed", () => timestamp: DateTime.makeUnsafe(3), text: "summary", }, - } satisfies SessionEvent.Event), + } satisfies SessionMessageUpdater.Input), ) Effect.runSync( SessionMessageUpdater.update(SessionMessageUpdater.memory(state), { id: EventV2.ID.create(), - durable: durable(sessionID, 3), type: "session.next.compaction.ended", data: { sessionID, @@ -271,7 +253,7 @@ test("compaction events reduce to compaction message only when completed", () => text: "final summary", recent: "recent context", }, - } satisfies SessionEvent.Event), + } satisfies SessionMessageUpdater.Input), ) expect(state.messages).toHaveLength(1) diff --git a/packages/schema/src/event.ts b/packages/schema/src/event.ts index ba116b05e5..168eb1eaf7 100644 --- a/packages/schema/src/event.ts +++ b/packages/schema/src/event.ts @@ -48,14 +48,6 @@ export type DurableDefinition< export type Definition = LiveDefinition | DurableDefinition -type Defined< - Type extends string, - DataSchema extends Schema.Codec, - Durability extends DurableOptions | undefined, -> = Durability extends DurableOptions - ? DurableDefinition - : LiveDefinition - export type Data = Schema.Schema.Type export type UncommittedPayload = D extends Definition @@ -68,47 +60,73 @@ export type UncommittedPayload = D extends De } : never -export type PublishedPayload = D extends Definition - ? UncommittedPayload & - (D extends { readonly durable: infer Durability extends DurableOptions } - ? { readonly durable: DurableEnvelope } - : { readonly durable?: never }) - : never +export type DurablePublishedPayload = UncommittedPayload & { + readonly durable: DurableEnvelope +} + +export type LivePublishedPayload = UncommittedPayload & { + readonly durable?: never +} + +export type PublishedPayload = D extends DurableDefinition + ? DurablePublishedPayload + : D extends LiveDefinition + ? LivePublishedPayload + : never export type Payload = PublishedPayload -type EventSchema< +type LiveEventSchema< Type extends string, Fields extends Readonly>>, - Durability extends DurableOptions | undefined, -> = Schema.Schema, Durability>>> & - Defined, Durability> +> = Schema.Schema>>> & + LiveDefinition> + +type DurableEventSchema< + Type extends string, + Fields extends Readonly>>, + Durability extends DurableOptions, +> = Schema.Schema, Durability>>> & + DurableDefinition, Durability> export function define< const Type extends string, Fields extends Readonly>>, - const Durability extends DurableOptions | undefined = undefined, +>(input: { readonly type: Type; readonly durable?: never; readonly schema: Fields }): LiveEventSchema +export function define< + const Type extends string, + Fields extends Readonly>>, + const Durability extends DurableOptions, >(input: { readonly type: Type - readonly durable?: Durability + readonly durable: Durability readonly schema: Fields -}): EventSchema { +}): DurableEventSchema +export function define(input: { + readonly type: string + readonly durable?: DurableOptions + readonly schema: Readonly>> +}): Schema.Top { const data = Schema.Struct(input.schema) - return Object.assign( - Schema.Struct({ - id: ID, - metadata: Schema.optional(Schema.Record(Schema.String, Schema.Unknown)), - type: Schema.Literal(input.type), - durable: input.durable === undefined ? NoDurableEnvelope : durableEnvelope(input.durable.version), - location: Schema.optional(Location.Ref), - data, - }).annotate({ identifier: input.type }), - { - type: input.type, - ...(input.durable === undefined ? {} : { durable: input.durable }), - data, - }, - ) as unknown as EventSchema + const fields = { + id: ID, + metadata: Schema.optional(Schema.Record(Schema.String, Schema.Unknown)), + type: Schema.Literal(input.type), + location: Schema.optional(Location.Ref), + data, + } + if (input.durable) { + return Object.assign( + Schema.Struct({ ...fields, durable: durableEnvelope(input.durable.version) }).annotate({ + identifier: input.type, + }), + { type: input.type, durable: input.durable, data }, + ) + } + return Object.assign(Schema.Struct({ ...fields, durable: NoDurableEnvelope }).annotate({ identifier: input.type }), { + type: input.type, + data, + }) } export function inventory>(...definitions: Definitions) {