From 603b334b7f70620e8e531fadcd1f0ff159f186ab Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Thu, 25 Jun 2026 12:41:44 -0400 Subject: [PATCH] fix(core): preserve event routing semantics --- packages/core/src/event.ts | 79 +++++------ packages/core/src/session/message-updater.ts | 15 +-- packages/core/test/event.test.ts | 44 +++++++ .../test/session-runner-tool-events.test.ts | 9 +- .../server/httpapi-public-openapi.test.ts | 3 +- .../test/server/httpapi-v2-location.test.ts | 4 + .../test/v2/session-message-updater.test.ts | 44 +++++-- packages/protocol/src/groups/event.ts | 9 +- packages/protocol/test/event.test.ts | 27 ++++ packages/schema/src/event.ts | 33 +++-- packages/schema/test/event.test.ts | 5 +- packages/sdk/js/script/build.ts | 11 ++ packages/sdk/js/src/v2/gen/types.gen.ts | 123 +++++++++++++----- 13 files changed, 277 insertions(+), 129 deletions(-) create mode 100644 packages/protocol/test/event.test.ts diff --git a/packages/core/src/event.ts b/packages/core/src/event.ts index 0b0eae26d7..2c0261ebbf 100644 --- a/packages/core/src/event.ts +++ b/packages/core/src/event.ts @@ -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 = (event: Payload) => Effect.Effect export type Unsubscribe = Effect.Effect @@ -80,13 +73,10 @@ export interface Interface { ) => Effect.Effect> readonly subscribe: (definition: D) => Stream.Stream> readonly all: () => Stream.Stream - readonly durable: (input: { - readonly aggregateID: string - readonly after?: number - }) => Stream.Stream> + readonly durable: (input: { readonly aggregateID: string; readonly after?: number }) => Stream.Stream /** @deprecated Use `all()` and consume the returned stream. */ readonly listen: (listener: Subscriber) => Effect.Effect - readonly project: (definition: D, projector: Subscriber) => Effect.Effect + readonly project: (definition: D, projector: Subscriber) => Effect.Effect 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, + input: undefined, + commit?: (seq: number) => Effect.Effect, + ): Effect.Effect> + function commitDurableEvent( + definition: DurableDefinition, + event: UncommittedPayload, + input: { + readonly seq: number + readonly aggregateID: string + readonly ownerID?: string + readonly strictOwner?: boolean + }, + commit?: (seq: number) => Effect.Effect, + ): Effect.Effect | undefined> function commitDurableEvent( definition: DurableDefinition, event: UncommittedPayload, @@ -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)[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 = - (definition: D) => - (event: Payload): event is Payload => - Schema.is(definition)(event) - function publish( definition: D, data: Data, @@ -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 = { id: options?.id ?? ID.create(), ...(options?.metadata ? { metadata: options.metadata } : {}), type: definition.type, @@ -467,7 +461,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.filter(isPayload(definition)), + Stream.map((event) => event as Payload), ) const streamAll = (): Stream.Stream => 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> => + const durable = (input: { readonly aggregateID: string; readonly after?: number }): Stream.Stream => 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 = (definition: D, projector: Subscriber): Effect.Effect => + const project = (definition: D, projector: Subscriber): Effect.Effect => 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)) projectors.set(definition.type, list) }) diff --git a/packages/core/src/session/message-updater.ts b/packages/core/src/session/message-updater.ts index 8f856d2288..a97afb58b6 100644 --- a/packages/core/src/session/message-updater.ts +++ b/packages/core/src/session/message-updater.ts @@ -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 readonly getAssistant: (messageID: SessionMessage.ID) => Effect.Effect @@ -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 type DraftTool = WritableDraft type DraftText = WritableDraft @@ -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" diff --git a/packages/core/test/event.test.ts b/packages/core/test/event.test.ts index 2284bba7e4..26d93b4d99 100644 --- a/packages/core/test/event.test.ts +++ b/packages/core/test/event.test.ts @@ -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() + 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 diff --git a/packages/core/test/session-runner-tool-events.test.ts b/packages/core/test/session-runner-tool-events.test.ts index f96ea4dea2..2fe34c6276 100644 --- a/packages/core/test/session-runner-tool-events.test.ts +++ b/packages/core/test/session-runner-tool-events.test.ts @@ -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 + 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 published.push({ type: definition.durable ? EventV2.versionedType(definition.type, definition.durable.version) diff --git a/packages/opencode/test/server/httpapi-public-openapi.test.ts b/packages/opencode/test/server/httpapi-public-openapi.test.ts index d9eb3643a9..73a0158795 100644 --- a/packages/opencode/test/server/httpapi-public-openapi.test.ts +++ b/packages/opencode/test/server/httpapi-public-openapi.test.ts @@ -106,7 +106,8 @@ describe("PublicApi OpenAPI v2 errors", () => { expect(durable?.required).toContain("durable") expect(durable?.properties?.durable).toBeDefined() - expect(live?.properties?.durable).toBeUndefined() + expect(live?.required).not.toContain("durable") + expect(Reflect.get(live?.properties?.durable ?? {}, "not")).toEqual({}) }) test("preserves /api auth responses", () => { diff --git a/packages/opencode/test/server/httpapi-v2-location.test.ts b/packages/opencode/test/server/httpapi-v2-location.test.ts index e178062e7a..172e0bd4a1 100644 --- a/packages/opencode/test/server/httpapi-v2-location.test.ts +++ b/packages/opencode/test/server/httpapi-v2-location.test.ts @@ -23,6 +23,7 @@ function request(route: string, directory: string, init: RequestInit = {}) { const Event = Schema.Struct({ id: EventV2.ID, type: Schema.String, + durable: Schema.optional(Schema.Struct({ aggregateID: Schema.String, seq: Schema.Int, version: Schema.Int })), location: Schema.optional(Location.Ref), data: Schema.Unknown, }) @@ -81,12 +82,15 @@ describe("v2 location HttpApi", () => { const reader = response.body!.getReader() const connected = await readEvent(reader) expect(connected.type).toBe("server.connected") + expect(connected).not.toHaveProperty("durable") expect(connected.location).toBeUndefined() const created = await request("/session", publisher.path, { method: "POST" }) expect(created.status).toBe(200) + const session = (await created.json()) as { id: string } expect(await readEventType(reader, "session.created")).toMatchObject({ type: "session.created", + durable: { aggregateID: session.id, seq: 0, version: 1 }, location: { directory: publisher.path }, data: { sessionID: expect.any(String) }, }) diff --git a/packages/opencode/test/v2/session-message-updater.test.ts b/packages/opencode/test/v2/session-message-updater.test.ts index 5611f6ed70..56478a4d1d 100644 --- a/packages/opencode/test/v2/session-message-updater.test.ts +++ b/packages/opencode/test/v2/session-message-updater.test.ts @@ -5,9 +5,16 @@ 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") @@ -16,6 +23,7 @@ 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, @@ -29,7 +37,7 @@ test.skip("step snapshots carry over to assistant messages", () => { }, snapshot: "before", }, - } satisfies SessionMessageUpdater.Input), + } satisfies SessionEvent.Event), ) expect(state.messages).toEqual([]) @@ -37,6 +45,7 @@ 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, @@ -52,7 +61,7 @@ test.skip("step snapshots carry over to assistant messages", () => { }, snapshot: "after", }, - } satisfies SessionMessageUpdater.Input), + } satisfies SessionEvent.Event), ) expect(state.messages[0]?.type).toBe("assistant") @@ -69,6 +78,7 @@ 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, @@ -81,12 +91,13 @@ test.skip("text ended populates assistant text content", () => { variant: ModelV2.VariantID.make("default"), }, }, - } satisfies SessionMessageUpdater.Input), + } satisfies SessionEvent.Event), ) Effect.runSync( SessionMessageUpdater.update(SessionMessageUpdater.memory(state), { id: EventV2.ID.create(), + durable: durable(sessionID, 1), type: "session.next.text.started", data: { sessionID, @@ -94,12 +105,13 @@ test.skip("text ended populates assistant text content", () => { timestamp: DateTime.makeUnsafe(2), textID: "text-1", }, - } satisfies SessionMessageUpdater.Input), + } satisfies SessionEvent.Event), ) Effect.runSync( SessionMessageUpdater.update(SessionMessageUpdater.memory(state), { id: EventV2.ID.create(), + durable: durable(sessionID, 2), type: "session.next.text.ended", data: { sessionID, @@ -108,7 +120,7 @@ test.skip("text ended populates assistant text content", () => { textID: "text-1", text: "hello assistant", }, - } satisfies SessionMessageUpdater.Input), + } satisfies SessionEvent.Event), ) expect(state.messages[0]?.type).toBe("assistant") @@ -125,6 +137,7 @@ 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, @@ -137,12 +150,13 @@ test.skip("tool completion stores completed timestamp", () => { variant: ModelV2.VariantID.make("default"), }, }, - } satisfies SessionMessageUpdater.Input), + } satisfies SessionEvent.Event), ) Effect.runSync( SessionMessageUpdater.update(SessionMessageUpdater.memory(state), { id: EventV2.ID.create(), + durable: durable(sessionID, 1), type: "session.next.tool.input.started", data: { sessionID, @@ -151,12 +165,13 @@ test.skip("tool completion stores completed timestamp", () => { callID, name: "bash", }, - } satisfies SessionMessageUpdater.Input), + } satisfies SessionEvent.Event), ) Effect.runSync( SessionMessageUpdater.update(SessionMessageUpdater.memory(state), { id: EventV2.ID.create(), + durable: durable(sessionID, 2), type: "session.next.tool.called", data: { sessionID, @@ -167,12 +182,13 @@ test.skip("tool completion stores completed timestamp", () => { input: { command: "pwd" }, provider: { executed: true, metadata: { fake: { source: "provider" } } }, }, - } satisfies SessionMessageUpdater.Input), + } satisfies SessionEvent.Event), ) Effect.runSync( SessionMessageUpdater.update(SessionMessageUpdater.memory(state), { id: EventV2.ID.create(), + durable: durable(sessionID, 3), type: "session.next.tool.success", data: { sessionID, @@ -183,7 +199,7 @@ test.skip("tool completion stores completed timestamp", () => { content: [{ type: "text", text: "/tmp" }], provider: { executed: true, metadata: { fake: { status: "done" } } }, }, - } satisfies SessionMessageUpdater.Input), + } satisfies SessionEvent.Event), ) expect(state.messages[0]?.type).toBe("assistant") @@ -203,6 +219,7 @@ 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, @@ -210,7 +227,7 @@ test("compaction events reduce to compaction message only when completed", () => timestamp: DateTime.makeUnsafe(1), reason: "auto", }, - } satisfies SessionMessageUpdater.Input), + } satisfies SessionEvent.Event), ) expect(state.messages).toEqual([]) @@ -225,7 +242,7 @@ test("compaction events reduce to compaction message only when completed", () => timestamp: DateTime.makeUnsafe(2), text: "hello ", }, - } satisfies SessionMessageUpdater.Input), + } satisfies SessionEvent.Event), ) Effect.runSync( @@ -238,12 +255,13 @@ test("compaction events reduce to compaction message only when completed", () => timestamp: DateTime.makeUnsafe(3), text: "summary", }, - } satisfies SessionMessageUpdater.Input), + } satisfies SessionEvent.Event), ) Effect.runSync( SessionMessageUpdater.update(SessionMessageUpdater.memory(state), { id: EventV2.ID.create(), + durable: durable(sessionID, 3), type: "session.next.compaction.ended", data: { sessionID, @@ -253,7 +271,7 @@ test("compaction events reduce to compaction message only when completed", () => text: "final summary", recent: "recent context", }, - } satisfies SessionMessageUpdater.Input), + } satisfies SessionEvent.Event), ) expect(state.messages).toHaveLength(1) diff --git a/packages/protocol/src/groups/event.ts b/packages/protocol/src/groups/event.ts index 703db3a3d6..cf6195a264 100644 --- a/packages/protocol/src/groups/event.ts +++ b/packages/protocol/src/groups/event.ts @@ -11,18 +11,21 @@ const fields = { location: Schema.optional(Location.Ref), } +const DurableEnvelope = Schema.Struct({ aggregateID: Schema.String, seq: Schema.Int, version: Schema.Int }) + const schema = (definitions: ReadonlyArray) => Schema.Union([ ...definitions.map((definition) => definition.durable ? Schema.Struct({ ...fields, - durable: Event.durableEnvelope(definition.durable.version), + durable: DurableEnvelope, type: Schema.Literal(definition.type), data: definition.data, }).annotate({ identifier: `V2Event.${definition.type}` }) : Schema.Struct({ ...fields, + durable: Schema.optional(Schema.Never), type: Schema.Literal(definition.type), data: definition.data, }).annotate({ identifier: `V2Event.${definition.type}` }), @@ -32,6 +35,7 @@ const schema = (definitions: ReadonlyArray) => : [ Schema.Struct({ ...fields, + durable: Schema.optional(Schema.Never), type: Schema.Literal("server.connected"), data: Schema.Struct({}), }).annotate({ identifier: "V2Event.server.connected" }), @@ -62,4 +66,5 @@ export const makeEventGroup = (definitions: ReadonlyArray) => make(d const event = make(EventManifest.ServerDefinitions) export const EventGroup = event.group -export type Event = typeof event.schema.Type +export const EventSchema = event.schema +export type Event = typeof EventSchema.Type diff --git a/packages/protocol/test/event.test.ts b/packages/protocol/test/event.test.ts new file mode 100644 index 0000000000..e254087bd0 --- /dev/null +++ b/packages/protocol/test/event.test.ts @@ -0,0 +1,27 @@ +import { describe, expect, test } from "bun:test" +import { Event } from "@opencode-ai/schema/event" +import { Schema } from "effect" +import { EventSchema } from "../src/groups/event" + +describe("EventSchema", () => { + test("requires durable metadata on durable events", () => { + expect( + Schema.is(EventSchema)({ + id: Event.ID.create(), + type: "session.created", + data: { sessionID: "session" }, + }), + ).toBe(false) + }) + + test("rejects durable metadata on live events", () => { + expect( + Schema.is(EventSchema)({ + id: Event.ID.create(), + type: "server.connected", + durable: { aggregateID: "aggregate", seq: 0, version: 1 }, + data: {}, + }), + ).toBe(false) + }) +}) diff --git a/packages/schema/src/event.ts b/packages/schema/src/event.ts index 168eb1eaf7..243667e543 100644 --- a/packages/schema/src/event.ts +++ b/packages/schema/src/event.ts @@ -3,7 +3,7 @@ export * as Event from "./event" import { Schema } from "effect" import { ascending } from "./identifier" import { Location } from "./location" -import { NonNegativeInt, statics } from "./schema" +import { statics } from "./schema" export const ID = Schema.String.check(Schema.isStartsWith("evt_")).pipe( Schema.brand("Event.ID"), @@ -16,15 +16,17 @@ export type DurableOptions = { readonly aggregate: string } -export type DurableEnvelope = { +export type DurableEnvelope = { readonly aggregateID: string readonly seq: number - readonly version: Version + readonly version: number } -export const durableEnvelope = (version: Version) => - Schema.Struct({ aggregateID: Schema.String, seq: NonNegativeInt, version: Schema.Literal(version) }) - +const PublishedDurableEnvelope = Schema.Struct({ + aggregateID: Schema.String, + seq: Schema.Number, + version: Schema.Number, +}) const NoDurableEnvelope = Schema.optional(Schema.Never) export type LiveDefinition< @@ -46,7 +48,10 @@ export type DurableDefinition< readonly durable: Durability } -export type Definition = LiveDefinition | DurableDefinition +export type Definition< + Type extends string = string, + DataSchema extends Schema.Codec = Schema.Codec, +> = LiveDefinition | DurableDefinition export type Data = Schema.Schema.Type @@ -60,18 +65,10 @@ export type UncommittedPayload = D extends De } : never -export type DurablePublishedPayload = UncommittedPayload & { - readonly durable: DurableEnvelope -} - -export type LivePublishedPayload = UncommittedPayload & { - readonly durable?: never -} - export type PublishedPayload = D extends DurableDefinition - ? DurablePublishedPayload + ? UncommittedPayload & { readonly durable: DurableEnvelope } : D extends LiveDefinition - ? LivePublishedPayload + ? UncommittedPayload & { readonly durable?: never } : never export type Payload = PublishedPayload @@ -117,7 +114,7 @@ export function define(input: { } if (input.durable) { return Object.assign( - Schema.Struct({ ...fields, durable: durableEnvelope(input.durable.version) }).annotate({ + Schema.Struct({ ...fields, durable: PublishedDurableEnvelope }).annotate({ identifier: input.type, }), { type: input.type, durable: input.durable, data }, diff --git a/packages/schema/test/event.test.ts b/packages/schema/test/event.test.ts index 3b119368cc..02b1feda76 100644 --- a/packages/schema/test/event.test.ts +++ b/packages/schema/test/event.test.ts @@ -56,9 +56,6 @@ describe("public event schemas", () => { data: { id: "aggregate" }, }), ).toBe(false) - expect(Schema.is(definition)({ ...payload, durable: { ...payload.durable, seq: -1 } })).toBe(false) - expect(Schema.is(definition)({ ...payload, durable: { ...payload.durable, version: 2 } })).toBe(false) - // @ts-expect-error Published durable payloads require commit metadata. const missing: typeof definition.Type = { id: Event.ID.create(), type: definition.type, data: { id: "aggregate" } } void missing @@ -119,10 +116,10 @@ describe("public event schemas", () => { // @ts-expect-error Durable union members require commit metadata. const uncommitted: Mixed = { id: Event.ID.create(), type: durable.type, data: { id: "aggregate" } } + // @ts-expect-error Live union members cannot carry durable commit metadata. const falselyCommitted: Mixed = { id: Event.ID.create(), type: live.type, - // @ts-expect-error Live union members cannot carry durable commit metadata. durable: { aggregateID: "aggregate", seq: 0, version: 2 }, data: { value: "value" }, } diff --git a/packages/sdk/js/script/build.ts b/packages/sdk/js/script/build.ts index 72f4e3f3e9..2b2032f714 100755 --- a/packages/sdk/js/script/build.ts +++ b/packages/sdk/js/script/build.ts @@ -58,6 +58,17 @@ if (sseTypesPatched === sseTypesSource) { } await Bun.write(sseTypesPath, sseTypesPatched) +// OpenAPI represents Schema.Never as `not: {}`, which @hey-api currently +// widens to unknown. Preserve impossible optional event fields as never. +const eventTypesPath = "./src/v2/gen/types.gen.ts" +const eventTypesFile = Bun.file(eventTypesPath) +const eventTypesSource = await eventTypesFile.text() +const eventTypesPatched = eventTypesSource.replaceAll(" durable?: unknown", " durable?: never") +if (eventTypesPatched === eventTypesSource) { + throw new Error(`Event never patch did not apply; @hey-api/openapi-ts output may have changed (${eventTypesPath})`) +} +await Bun.write(eventTypesPath, eventTypesPatched) + await $`bun prettier --write src/gen` await $`bun prettier --write src/v2` await $`rm -rf dist` diff --git a/packages/sdk/js/src/v2/gen/types.gen.ts b/packages/sdk/js/src/v2/gen/types.gen.ts index b653c75e1b..ce480c6ae6 100644 --- a/packages/sdk/js/src/v2/gen/types.gen.ts +++ b/packages/sdk/js/src/v2/gen/types.gen.ts @@ -4329,6 +4329,7 @@ export type V2EventModelsDevRefreshed = { [key: string]: unknown } location?: LocationRef + durable?: never type: "models-dev.refreshed" data: { [key: string]: unknown @@ -4341,6 +4342,7 @@ export type V2EventIntegrationUpdated = { [key: string]: unknown } location?: LocationRef + durable?: never type: "integration.updated" data: { [key: string]: unknown @@ -4353,6 +4355,7 @@ export type V2EventIntegrationConnectionUpdated = { [key: string]: unknown } location?: LocationRef + durable?: never type: "integration.connection.updated" data: { integrationID: string @@ -4365,6 +4368,7 @@ export type V2EventCatalogUpdated = { [key: string]: unknown } location?: LocationRef + durable?: never type: "catalog.updated" data: { [key: string]: unknown @@ -4380,7 +4384,7 @@ export type V2EventSessionCreated = { durable: { aggregateID: string seq: number - version: 1 + version: number } type: "session.created" data: { @@ -4398,7 +4402,7 @@ export type V2EventSessionUpdated = { durable: { aggregateID: string seq: number - version: 1 + version: number } type: "session.updated" data: { @@ -4416,7 +4420,7 @@ export type V2EventSessionDeleted = { durable: { aggregateID: string seq: number - version: 1 + version: number } type: "session.deleted" data: { @@ -4434,7 +4438,7 @@ export type V2EventMessageUpdated = { durable: { aggregateID: string seq: number - version: 1 + version: number } type: "message.updated" data: { @@ -4452,7 +4456,7 @@ export type V2EventMessageRemoved = { durable: { aggregateID: string seq: number - version: 1 + version: number } type: "message.removed" data: { @@ -4470,7 +4474,7 @@ export type V2EventMessagePartUpdated = { durable: { aggregateID: string seq: number - version: 1 + version: number } type: "message.part.updated" data: { @@ -4489,7 +4493,7 @@ export type V2EventMessagePartRemoved = { durable: { aggregateID: string seq: number - version: 1 + version: number } type: "message.part.removed" data: { @@ -4508,7 +4512,7 @@ export type V2EventSessionNextAgentSwitched = { durable: { aggregateID: string seq: number - version: 1 + version: number } type: "session.next.agent.switched" data: { @@ -4528,7 +4532,7 @@ export type V2EventSessionNextModelSwitched = { durable: { aggregateID: string seq: number - version: 1 + version: number } type: "session.next.model.switched" data: { @@ -4552,7 +4556,7 @@ export type V2EventSessionNextMoved = { durable: { aggregateID: string seq: number - version: 1 + version: number } type: "session.next.moved" data: { @@ -4572,7 +4576,7 @@ export type V2EventSessionNextPrompted = { durable: { aggregateID: string seq: number - version: 1 + version: number } type: "session.next.prompted" data: { @@ -4593,7 +4597,7 @@ export type V2EventSessionNextPromptAdmitted = { durable: { aggregateID: string seq: number - version: 1 + version: number } type: "session.next.prompt.admitted" data: { @@ -4614,7 +4618,7 @@ export type V2EventSessionNextContextUpdated = { durable: { aggregateID: string seq: number - version: 1 + version: number } type: "session.next.context.updated" data: { @@ -4634,7 +4638,7 @@ export type V2EventSessionNextSynthetic = { durable: { aggregateID: string seq: number - version: 1 + version: number } type: "session.next.synthetic" data: { @@ -4654,7 +4658,7 @@ export type V2EventSessionNextShellStarted = { durable: { aggregateID: string seq: number - version: 1 + version: number } type: "session.next.shell.started" data: { @@ -4675,7 +4679,7 @@ export type V2EventSessionNextShellEnded = { durable: { aggregateID: string seq: number - version: 1 + version: number } type: "session.next.shell.ended" data: { @@ -4695,7 +4699,7 @@ export type V2EventSessionNextStepStarted = { durable: { aggregateID: string seq: number - version: 1 + version: number } type: "session.next.step.started" data: { @@ -4721,7 +4725,7 @@ export type V2EventSessionNextStepEnded = { durable: { aggregateID: string seq: number - version: 2 + version: number } type: "session.next.step.ended" data: { @@ -4753,7 +4757,7 @@ export type V2EventSessionNextStepFailed = { durable: { aggregateID: string seq: number - version: 2 + version: number } type: "session.next.step.failed" data: { @@ -4773,7 +4777,7 @@ export type V2EventSessionNextTextStarted = { durable: { aggregateID: string seq: number - version: 1 + version: number } type: "session.next.text.started" data: { @@ -4790,6 +4794,7 @@ export type V2EventSessionNextTextDelta = { [key: string]: unknown } location?: LocationRef + durable?: never type: "session.next.text.delta" data: { timestamp: number @@ -4809,7 +4814,7 @@ export type V2EventSessionNextTextEnded = { durable: { aggregateID: string seq: number - version: 1 + version: number } type: "session.next.text.ended" data: { @@ -4830,7 +4835,7 @@ export type V2EventSessionNextReasoningStarted = { durable: { aggregateID: string seq: number - version: 1 + version: number } type: "session.next.reasoning.started" data: { @@ -4852,6 +4857,7 @@ export type V2EventSessionNextReasoningDelta = { [key: string]: unknown } location?: LocationRef + durable?: never type: "session.next.reasoning.delta" data: { timestamp: number @@ -4871,7 +4877,7 @@ export type V2EventSessionNextReasoningEnded = { durable: { aggregateID: string seq: number - version: 1 + version: number } type: "session.next.reasoning.ended" data: { @@ -4897,7 +4903,7 @@ export type V2EventSessionNextToolInputStarted = { durable: { aggregateID: string seq: number - version: 1 + version: number } type: "session.next.tool.input.started" data: { @@ -4915,6 +4921,7 @@ export type V2EventSessionNextToolInputDelta = { [key: string]: unknown } location?: LocationRef + durable?: never type: "session.next.tool.input.delta" data: { timestamp: number @@ -4934,7 +4941,7 @@ export type V2EventSessionNextToolInputEnded = { durable: { aggregateID: string seq: number - version: 1 + version: number } type: "session.next.tool.input.ended" data: { @@ -4955,7 +4962,7 @@ export type V2EventSessionNextToolCalled = { durable: { aggregateID: string seq: number - version: 1 + version: number } type: "session.next.tool.called" data: { @@ -4987,7 +4994,7 @@ export type V2EventSessionNextToolProgress = { durable: { aggregateID: string seq: number - version: 1 + version: number } type: "session.next.tool.progress" data: { @@ -5011,7 +5018,7 @@ export type V2EventSessionNextToolSuccess = { durable: { aggregateID: string seq: number - version: 1 + version: number } type: "session.next.tool.success" data: { @@ -5045,7 +5052,7 @@ export type V2EventSessionNextToolFailed = { durable: { aggregateID: string seq: number - version: 1 + version: number } type: "session.next.tool.failed" data: { @@ -5075,7 +5082,7 @@ export type V2EventSessionNextRetried = { durable: { aggregateID: string seq: number - version: 1 + version: number } type: "session.next.retried" data: { @@ -5095,7 +5102,7 @@ export type V2EventSessionNextCompactionStarted = { durable: { aggregateID: string seq: number - version: 1 + version: number } type: "session.next.compaction.started" data: { @@ -5112,6 +5119,7 @@ export type V2EventSessionNextCompactionDelta = { [key: string]: unknown } location?: LocationRef + durable?: never type: "session.next.compaction.delta" data: { timestamp: number @@ -5130,7 +5138,7 @@ export type V2EventSessionNextCompactionEnded = { durable: { aggregateID: string seq: number - version: 1 + version: number } type: "session.next.compaction.ended" data: { @@ -5152,7 +5160,7 @@ export type V2EventSessionNextRevertStaged = { durable: { aggregateID: string seq: number - version: 1 + version: number } type: "session.next.revert.staged" data: { @@ -5177,7 +5185,7 @@ export type V2EventSessionNextRevertCleared = { durable: { aggregateID: string seq: number - version: 1 + version: number } type: "session.next.revert.cleared" data: { @@ -5195,7 +5203,7 @@ export type V2EventSessionNextRevertCommitted = { durable: { aggregateID: string seq: number - version: 1 + version: number } type: "session.next.revert.committed" data: { @@ -5211,6 +5219,7 @@ export type V2EventMessagePartDelta = { [key: string]: unknown } location?: LocationRef + durable?: never type: "message.part.delta" data: { sessionID: string @@ -5227,6 +5236,7 @@ export type V2EventSessionDiff = { [key: string]: unknown } location?: LocationRef + durable?: never type: "session.diff" data: { sessionID: string @@ -5240,6 +5250,7 @@ export type V2EventSessionError = { [key: string]: unknown } location?: LocationRef + durable?: never type: "session.error" data: { sessionID?: string @@ -5261,6 +5272,7 @@ export type V2EventInstallationUpdated = { [key: string]: unknown } location?: LocationRef + durable?: never type: "installation.updated" data: { version: string @@ -5273,6 +5285,7 @@ export type V2EventInstallationUpdateAvailable = { [key: string]: unknown } location?: LocationRef + durable?: never type: "installation.update-available" data: { version: string @@ -5285,6 +5298,7 @@ export type V2EventFileEdited = { [key: string]: unknown } location?: LocationRef + durable?: never type: "file.edited" data: { file: string @@ -5297,6 +5311,7 @@ export type V2EventReferenceUpdated = { [key: string]: unknown } location?: LocationRef + durable?: never type: "reference.updated" data: { [key: string]: unknown @@ -5309,6 +5324,7 @@ export type V2EventPermissionV2Asked = { [key: string]: unknown } location?: LocationRef + durable?: never type: "permission.v2.asked" data: { id: string @@ -5329,6 +5345,7 @@ export type V2EventPermissionV2Replied = { [key: string]: unknown } location?: LocationRef + durable?: never type: "permission.v2.replied" data: { sessionID: string @@ -5343,6 +5360,7 @@ export type V2EventPluginAdded = { [key: string]: unknown } location?: LocationRef + durable?: never type: "plugin.added" data: { id: string @@ -5355,6 +5373,7 @@ export type V2EventProjectDirectoriesUpdated = { [key: string]: unknown } location?: LocationRef + durable?: never type: "project.directories.updated" data: { projectID: string @@ -5367,6 +5386,7 @@ export type V2EventFileWatcherUpdated = { [key: string]: unknown } location?: LocationRef + durable?: never type: "file.watcher.updated" data: { file: string @@ -5380,6 +5400,7 @@ export type V2EventPtyCreated = { [key: string]: unknown } location?: LocationRef + durable?: never type: "pty.created" data: { info: Pty @@ -5392,6 +5413,7 @@ export type V2EventPtyUpdated = { [key: string]: unknown } location?: LocationRef + durable?: never type: "pty.updated" data: { info: Pty @@ -5404,6 +5426,7 @@ export type V2EventPtyExited = { [key: string]: unknown } location?: LocationRef + durable?: never type: "pty.exited" data: { id: string @@ -5417,6 +5440,7 @@ export type V2EventPtyDeleted = { [key: string]: unknown } location?: LocationRef + durable?: never type: "pty.deleted" data: { id: string @@ -5429,6 +5453,7 @@ export type V2EventQuestionV2Asked = { [key: string]: unknown } location?: LocationRef + durable?: never type: "question.v2.asked" data: { id: string @@ -5447,6 +5472,7 @@ export type V2EventQuestionV2Replied = { [key: string]: unknown } location?: LocationRef + durable?: never type: "question.v2.replied" data: { sessionID: string @@ -5461,6 +5487,7 @@ export type V2EventQuestionV2Rejected = { [key: string]: unknown } location?: LocationRef + durable?: never type: "question.v2.rejected" data: { sessionID: string @@ -5474,6 +5501,7 @@ export type V2EventTodoUpdated = { [key: string]: unknown } location?: LocationRef + durable?: never type: "todo.updated" data: { sessionID: string @@ -5487,6 +5515,7 @@ export type V2EventLspUpdated = { [key: string]: unknown } location?: LocationRef + durable?: never type: "lsp.updated" data: { [key: string]: unknown @@ -5499,6 +5528,7 @@ export type V2EventPermissionAsked = { [key: string]: unknown } location?: LocationRef + durable?: never type: "permission.asked" data: { id: string @@ -5522,6 +5552,7 @@ export type V2EventPermissionReplied = { [key: string]: unknown } location?: LocationRef + durable?: never type: "permission.replied" data: { sessionID: string @@ -5536,6 +5567,7 @@ export type V2EventTuiPromptAppend = { [key: string]: unknown } location?: LocationRef + durable?: never type: "tui.prompt.append" data: { text: string @@ -5548,6 +5580,7 @@ export type V2EventTuiCommandExecute = { [key: string]: unknown } location?: LocationRef + durable?: never type: "tui.command.execute" data: { command: @@ -5577,6 +5610,7 @@ export type V2EventTuiToastShow = { [key: string]: unknown } location?: LocationRef + durable?: never type: "tui.toast.show" data: { title?: string @@ -5592,6 +5626,7 @@ export type V2EventTuiSessionSelect = { [key: string]: unknown } location?: LocationRef + durable?: never type: "tui.session.select" data: { /** @@ -5607,6 +5642,7 @@ export type V2EventMcpToolsChanged = { [key: string]: unknown } location?: LocationRef + durable?: never type: "mcp.tools.changed" data: { server: string @@ -5619,6 +5655,7 @@ export type V2EventMcpBrowserOpenFailed = { [key: string]: unknown } location?: LocationRef + durable?: never type: "mcp.browser.open.failed" data: { mcpName: string @@ -5632,6 +5669,7 @@ export type V2EventCommandExecuted = { [key: string]: unknown } location?: LocationRef + durable?: never type: "command.executed" data: { name: string @@ -5647,6 +5685,7 @@ export type V2EventProjectUpdated = { [key: string]: unknown } location?: LocationRef + durable?: never type: "project.updated" data: { id: string @@ -5679,6 +5718,7 @@ export type V2EventSessionStatus = { [key: string]: unknown } location?: LocationRef + durable?: never type: "session.status" data: { sessionID: string @@ -5692,6 +5732,7 @@ export type V2EventSessionIdle = { [key: string]: unknown } location?: LocationRef + durable?: never type: "session.idle" data: { sessionID: string @@ -5704,6 +5745,7 @@ export type V2EventQuestionAsked = { [key: string]: unknown } location?: LocationRef + durable?: never type: "question.asked" data: { id: string @@ -5722,6 +5764,7 @@ export type V2EventQuestionReplied = { [key: string]: unknown } location?: LocationRef + durable?: never type: "question.replied" data: { sessionID: string @@ -5736,6 +5779,7 @@ export type V2EventQuestionRejected = { [key: string]: unknown } location?: LocationRef + durable?: never type: "question.rejected" data: { sessionID: string @@ -5749,6 +5793,7 @@ export type V2EventSessionCompacted = { [key: string]: unknown } location?: LocationRef + durable?: never type: "session.compacted" data: { sessionID: string @@ -5761,6 +5806,7 @@ export type V2EventVcsBranchUpdated = { [key: string]: unknown } location?: LocationRef + durable?: never type: "vcs.branch.updated" data: { branch?: string @@ -5773,6 +5819,7 @@ export type V2EventWorkspaceReady = { [key: string]: unknown } location?: LocationRef + durable?: never type: "workspace.ready" data: { name: string @@ -5785,6 +5832,7 @@ export type V2EventWorkspaceFailed = { [key: string]: unknown } location?: LocationRef + durable?: never type: "workspace.failed" data: { message: string @@ -5797,6 +5845,7 @@ export type V2EventWorkspaceStatus = { [key: string]: unknown } location?: LocationRef + durable?: never type: "workspace.status" data: { workspaceID: string @@ -5810,6 +5859,7 @@ export type V2EventWorktreeReady = { [key: string]: unknown } location?: LocationRef + durable?: never type: "worktree.ready" data: { name: string @@ -5823,6 +5873,7 @@ export type V2EventWorktreeFailed = { [key: string]: unknown } location?: LocationRef + durable?: never type: "worktree.failed" data: { message: string @@ -5835,6 +5886,7 @@ export type V2EventServerConnected = { [key: string]: unknown } location?: LocationRef + durable?: never type: "server.connected" data: { [key: string]: unknown @@ -5847,6 +5899,7 @@ export type V2EventGlobalDisposed = { [key: string]: unknown } location?: LocationRef + durable?: never type: "global.disposed" data: { [key: string]: unknown