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

refactor(core): prove event publication branches

Browse source
This commit is contained in:
Kit Langton 2026-06-25 12:04:05 -04:00
commit 45071a8d8e
4 changed files with 132 additions and 123 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -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<string, unknown>)[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<D extends Definition>(
definition: D,
event: UncommittedPayload<D>,
commit?: PublishOptions["commit"],
): Effect.Effect<PublishedPayload<D>> {
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<DurableDefinition>,
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<D>
}
const published = event as PublishedPayload<D>
yield* notify(published, false)
return published
})
}
const observe = (event: Payload, observer: (event: Payload) => Effect.Effect<void>) =>
Effect.suspend(() => observer(event)).pipe(
Effect.catchCauseIf(
@ -358,7 +329,17 @@ export const layerWith = (options?: LayerOptions) =>
})
}
function publish<D extends Definition>(definition: D, data: Data<D>, options?: PublishOptions) {
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>,
options?: PublishOptions,
): Effect.Effect<Payload<D>>
function publish(definition: Definition, data: unknown, options?: PublishOptions): Effect.Effect<Payload> {
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<DurableDefinition> = {
id: options?.id ?? ID.create(),
...(options?.metadata ? { metadata: options.metadata } : {}),
type: definition.type,
...(location ? { location } : {}),
data,
} as UncommittedPayload<D>,
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 = <D extends Definition>(definition: D): Stream.Stream<Payload<D>> =>
Stream.unwrap(getOrCreate(definition).pipe(Effect.map((pubsub) => Stream.fromPubSub(pubsub)))).pipe(
Stream.map((event) => event as Payload<D>),
Stream.filter(isPayload(definition)),
)
const streamAll = (): Stream.Stream<Payload> => Stream.fromPubSub(pubsub.all)
@ -563,7 +565,11 @@ export const layerWith = (options?: LayerOptions) =>
const project = <D extends DurableDefinition>(definition: D, projector: Subscriber<D>): Effect.Effect<void> =>
Effect.sync(() => {
const list = projectors.get(definition.type) ?? []
list.push((event) => projector(event as Payload<D>))
list.push((event) =>
isPayload(definition)(event)
? projector(event)
: Effect.die(`Published event ${event.type} does not match its definition`),
)
projectors.set(definition.type, list)
})

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

@ -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<SessionMessage.Assistant | undefined>
readonly getAssistant: (messageID: SessionMessage.ID) => Effect.Effect<SessionMessage.Assistant | undefined>
@ -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<SessionMessage.Assistant>
type DraftTool = WritableDraft<SessionMessage.AssistantTool>
type DraftText = WritableDraft<SessionMessage.AssistantText>
@ -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"

44
packages/opencode/test/v2/session-message-updater.test.ts
View file

@ -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)

90
packages/schema/src/event.ts
View file

@ -48,14 +48,6 @@ export type DurableDefinition<
export type Definition = LiveDefinition | DurableDefinition
type Defined<
Type extends string,
DataSchema extends Schema.Codec<unknown, unknown>,
Durability extends DurableOptions | undefined,
> = Durability extends DurableOptions
? DurableDefinition<Type, DataSchema, Durability>
: LiveDefinition<Type, DataSchema>
export type Data<D extends Definition> = Schema.Schema.Type<D["data"]>
export type UncommittedPayload<D extends Definition = Definition> = D extends Definition
@ -68,47 +60,73 @@ export type UncommittedPayload<D extends Definition = Definition> = D extends De
}
: never
export type PublishedPayload<D extends Definition = Definition> = D extends Definition
? UncommittedPayload<D> &
(D extends { readonly durable: infer Durability extends DurableOptions }
? { readonly durable: DurableEnvelope<Durability["version"]> }
: { readonly durable?: never })
: never
export type DurablePublishedPayload<D extends DurableDefinition = DurableDefinition> = UncommittedPayload<D> & {
readonly durable: DurableEnvelope<D["durable"]["version"]>
}
export type LivePublishedPayload<D extends LiveDefinition = LiveDefinition> = UncommittedPayload<D> & {
readonly durable?: never
}
export type PublishedPayload<D extends Definition = Definition> = D extends DurableDefinition
? DurablePublishedPayload<D>
: D extends LiveDefinition
? LivePublishedPayload<D>
: never
export type Payload<D extends Definition = Definition> = PublishedPayload<D>
type EventSchema<
type LiveEventSchema<
Type extends string,
Fields extends Readonly<Record<PropertyKey, Schema.Codec<unknown, unknown>>>,
Durability extends DurableOptions | undefined,
> = Schema.Schema<PublishedPayload<Defined<Type, Schema.Struct<Fields>, Durability>>> &
Defined<Type, Schema.Struct<Fields>, Durability>
> = Schema.Schema<PublishedPayload<LiveDefinition<Type, Schema.Struct<Fields>>>> &
LiveDefinition<Type, Schema.Struct<Fields>>
type DurableEventSchema<
Type extends string,
Fields extends Readonly<Record<PropertyKey, Schema.Codec<unknown, unknown>>>,
Durability extends DurableOptions,
> = Schema.Schema<PublishedPayload<DurableDefinition<Type, Schema.Struct<Fields>, Durability>>> &
DurableDefinition<Type, Schema.Struct<Fields>, Durability>
export function define<
const Type extends string,
Fields extends Readonly<Record<PropertyKey, Schema.Codec<unknown, unknown>>>,
const Durability extends DurableOptions | undefined = undefined,
>(input: { readonly type: Type; readonly durable?: never; readonly schema: Fields }): LiveEventSchema<Type, Fields>
export function define<
const Type extends string,
Fields extends Readonly<Record<PropertyKey, Schema.Codec<unknown, unknown>>>,
const Durability extends DurableOptions,
>(input: {
readonly type: Type
readonly durable?: Durability
readonly durable: Durability
readonly schema: Fields
}): EventSchema<Type, Fields, Durability> {
}): DurableEventSchema<Type, Fields, Durability>
export function define(input: {
readonly type: string
readonly durable?: DurableOptions
readonly schema: Readonly<Record<PropertyKey, Schema.Codec<unknown, unknown>>>
}): 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<Type, Fields, Durability>
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<const Definitions extends ReadonlyArray<Definition>>(...definitions: Definitions) {