fix(core): preserve event routing semantics

CONTEXT.md

@@ -141,6 +141,7 @@ _Avoid_: Response envelope
 - Promise streaming methods return a lazy `AsyncIterable` directly rather than a Promise-wrapped stream object. Iteration opens the connection, `AbortSignal` cancels it, and ending iteration closes the underlying request; the Effect emitter analogously returns `Stream` directly.
 - Promise SSE connection establishment, declared HTTP failures, and infrastructure failures occur during `AsyncIterable` iteration, beginning with its first `next()` call, rather than during synchronous method construction.
 - Neither generated streaming runtime automatically reconnects after disconnection. Promise `AsyncIterable` and Effect `Stream` fail explicitly; live consumers refresh and resubscribe, while durable sequence-based resume remains explicit composition above the generated client.
+- Event definition durability is authoritative for published payloads. Durable definitions publish and decode only with commit metadata (`aggregateID`, `seq`, and `version`); live definitions forbid that metadata. Core's pre-commit payload without an assigned sequence is a separate internal type and never reaches subscribers or projectors.
 - Promise client construction is synchronous and network-free. It requires `baseUrl`, defaults to `globalThis.fetch`, accepts client-level headers, and merges them with per-call header overrides.
 - Effect client construction accepts an explicit `baseUrl` and obtains `HttpClient.HttpClient` from the Effect environment. It does not install fetch or duplicate per-call transport policy; callers transform/provide the client for headers, tracing, retries, recording, and tests, while fiber interruption owns cancellation.
 - Promise and Effect emitters each own their generated public type modules. The **SDK Contract IR**, not a physically shared generated type package, is the common source; this permits zero-Effect wire types and rich decoded Effect types to evolve independently.

packages/core/src/event.ts

@@ -2,7 +2,14 @@ export * as EventV2 from "./event"
 
 import { Cause, Context, Effect, Layer, Option, PubSub, Schema, Stream } from "effect"
 import { Event } from "@opencode-ai/schema/event"
-import type { Data, Definition, Payload } from "@opencode-ai/schema/event"
+import type {
+  Data,
+  Definition,
+  DurableDefinition,
+  LiveDefinition,
+  Payload,
+  UncommittedPayload,
+} from "@opencode-ai/schema/event"
 import { and, asc, eq, gt } from "drizzle-orm"
 import { Database } from "./database/database"
 import { EventSequenceTable, EventTable } from "./event/sql"
@@ -124,8 +131,25 @@ export const layerWith = (options?: LayerOptions) =>
       )
 
       function commitDurableEvent(
-        definition: Definition,
-        event: Payload,
+        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>,
         input?: {
           readonly seq: number
           readonly aggregateID: string
@@ -135,7 +159,7 @@ export const layerWith = (options?: LayerOptions) =>
         commit?: (seq: number) => Effect.Effect<void>,
       ) {
         return Effect.gen(function* () {
-          const durable = definition?.durable
+          const durable = definition.durable
           if (durable) {
             const aggregateID = (event.data as Record<string, unknown>)[durable.aggregate]
             if (typeof aggregateID !== "string") {
@@ -200,7 +224,7 @@ export const layerWith = (options?: LayerOptions) =>
                                   .run()
                                   .pipe(Effect.orDie)
                               }
-                              return
+                              return undefined
                             }
                             yield* Effect.die(
                               new InvalidDurableEventError({
@@ -210,7 +234,7 @@ export const layerWith = (options?: LayerOptions) =>
                             )
                           }
                           if (input && row?.ownerID && row.ownerID !== input.ownerID) {
-                            return
+                            return undefined
                           }
                           const seq = input?.seq ?? latest + 1
                           if (input && seq !== latest + 1) {
@@ -234,10 +258,10 @@ export const layerWith = (options?: LayerOptions) =>
                                 message: `Event ${event.id} already exists at aggregate ${stored.aggregateID} sequence ${stored.seq}`,
                               }),
                             )
-                          const committed = {
+                          const committed: Payload<DurableDefinition> = {
                             ...event,
                             durable: { aggregateID, seq, version: durable.version },
-                          } as Payload
+                          }
                           for (const projector of list) {
                             yield* projector(committed)
                           }
@@ -267,14 +291,14 @@ export const layerWith = (options?: LayerOptions) =>
                             ])
                             .run()
                             .pipe(Effect.orDie)
-                          return { aggregateID, seq }
+                          return committed
                         }),
                       { behavior: "immediate" },
                     )
                     .pipe(Effect.orDie)
                   if (committed) {
                     yield* Effect.forEach(
-                      pubsub.durable.get(committed.aggregateID) ?? [],
+                      pubsub.durable.get(committed.durable.aggregateID) ?? [],
                       (wake) => PubSub.publish(wake, undefined),
                       { discard: true },
                     )
@@ -287,35 +311,6 @@ export const layerWith = (options?: LayerOptions) =>
         })
       }
 
-      function publishEvent<D extends Definition>(definition: D, event: Payload<D>, commit?: PublishOptions["commit"]) {
-        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 Payload, undefined, commit)
-            if (committed) {
-              event = {
-                ...event,
-                durable: {
-                  aggregateID: committed.aggregateID,
-                  seq: committed.seq,
-                  version: definition.durable.version,
-                },
-              }
-              yield* notify(event as Payload, true)
-              return event
-            }
-          }
-          yield* notify(event as Payload, false)
-          return event
-        })
-      }
-
       const observe = (event: Payload, observer: (event: Payload) => Effect.Effect<void>) =>
         Effect.suspend(() => observer(event)).pipe(
           Effect.catchCauseIf(
@@ -337,7 +332,12 @@ export const layerWith = (options?: LayerOptions) =>
         })
       }
 
-      function publish<D extends Definition>(definition: D, data: Data<D>, options?: PublishOptions) {
+      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 =
@@ -345,17 +345,34 @@ 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 Payload<D>,
-            options?.commit,
-          )
+            }
+            const committed = yield* commitDurableEvent(definition, event, undefined, options?.commit)
+            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: Payload<LiveDefinition> = {
+            id: options?.id ?? ID.create(),
+            ...(options?.metadata ? { metadata: options.metadata } : {}),
+            type: definition.type,
+            ...(location ? { location } : {}),
+            data,
+          }
+          yield* notify(event, false)
+          return event
         })
       }
 
@@ -370,11 +387,11 @@ export const layerWith = (options?: LayerOptions) =>
               new InvalidDurableEventError({ type: event.type, message: `Unknown durable event type ${event.type}` }),
             )
           } else {
-            const payload = {
+            const payload: UncommittedPayload<DurableDefinition> = {
               id: event.id,
               type: definition.type,
               data: Schema.decodeUnknownSync(definition.data)(event.data),
-            } as Payload
+            }
             const committed = yield* commitDurableEvent(definition, payload, {
               seq: event.seq,
               aggregateID: event.aggregateID,
@@ -382,17 +399,7 @@ export const layerWith = (options?: LayerOptions) =>
               strictOwner: options?.strictOwner,
             })
             if (committed && options?.publish) {
-              yield* notify(
-                {
-                  ...payload,
-                  durable: {
-                    aggregateID: committed.aggregateID,
-                    seq: committed.seq,
-                    version: definition.durable.version,
-                  },
-                },
-                true,
-              )
+              yield* notify(committed, true)
             }
           }
         })
@@ -459,7 +466,7 @@ export const layerWith = (options?: LayerOptions) =>
 
       const streamAll = (): Stream.Stream<Payload> => Stream.fromPubSub(pubsub.all)
 
-      const decodeSerializedEvent = (event: SerializedEvent) => {
+      const decodeSerializedEvent = (event: SerializedEvent): Payload<DurableDefinition> => {
         const definition = Durable.get(event.type)
         if (!definition?.durable) {
           throw new InvalidDurableEventError({ type: event.type, message: `Unknown durable event type ${event.type}` })

packages/core/test/event.test.ts

@@ -116,7 +116,7 @@ describe("EventV2", () => {
       const event = yield* events.publish(VersionedMessage, { id: "one", text: "hello" })
 
       expect(event.type).toBe("test.versioned")
-      expect(event.durable?.version).toBe(2)
+      expect(event.durable.version).toBe(2)
     }),
   )
 
@@ -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
@@ -764,7 +808,7 @@ describe("EventV2", () => {
       const replayed = {
         id: published.id,
         type: EventV2.versionedType(DurableMessage.type, 1),
-        seq: published.durable!.seq,
+        seq: published.durable.seq,
         aggregateID,
         data: published.data,
       }

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

packages/opencode/test/server/httpapi-public-openapi.test.ts

@@ -99,6 +99,17 @@ describe("PublicApi OpenAPI v2 errors", () => {
     })
   })
 
+  test("documents durable metadata only on durable events", () => {
+    const spec = OpenApi.fromApi(PublicApi) as OpenApiSpec
+    const durable = spec.components.schemas.V2EventSessionCreated
+    const live = spec.components.schemas.V2EventSessionNextTextDelta
+
+    expect(durable?.required).toContain("durable")
+    expect(durable?.properties?.durable).toBeDefined()
+    expect(live?.required).not.toContain("durable")
+    expect(Reflect.get(live?.properties?.durable ?? {}, "not")).toEqual({})
+  })
+
   test("preserves /api auth responses", () => {
     const spec = OpenApi.fromApi(PublicApi) as OpenApiSpec
 

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) },
     })

packages/opencode/test/v2/session-message-updater.test.ts

@@ -9,6 +9,12 @@ 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")
@@ -17,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,
@@ -38,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,
@@ -70,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,
@@ -88,6 +97,7 @@ test.skip("text ended populates assistant text content", () => {
   Effect.runSync(
     SessionMessageUpdater.update(SessionMessageUpdater.memory(state), {
       id: EventV2.ID.create(),
+      durable: durable(sessionID, 1),
       type: "session.next.text.started",
       data: {
         sessionID,
@@ -101,6 +111,7 @@ test.skip("text ended populates assistant text content", () => {
   Effect.runSync(
     SessionMessageUpdater.update(SessionMessageUpdater.memory(state), {
       id: EventV2.ID.create(),
+      durable: durable(sessionID, 2),
       type: "session.next.text.ended",
       data: {
         sessionID,
@@ -126,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,
@@ -144,6 +156,7 @@ test.skip("tool completion stores completed timestamp", () => {
   Effect.runSync(
     SessionMessageUpdater.update(SessionMessageUpdater.memory(state), {
       id: EventV2.ID.create(),
+      durable: durable(sessionID, 1),
       type: "session.next.tool.input.started",
       data: {
         sessionID,
@@ -158,6 +171,7 @@ test.skip("tool completion stores completed timestamp", () => {
   Effect.runSync(
     SessionMessageUpdater.update(SessionMessageUpdater.memory(state), {
       id: EventV2.ID.create(),
+      durable: durable(sessionID, 2),
       type: "session.next.tool.called",
       data: {
         sessionID,
@@ -174,6 +188,7 @@ test.skip("tool completion stores completed timestamp", () => {
   Effect.runSync(
     SessionMessageUpdater.update(SessionMessageUpdater.memory(state), {
       id: EventV2.ID.create(),
+      durable: durable(sessionID, 3),
       type: "session.next.tool.success",
       data: {
         sessionID,
@@ -204,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,
@@ -245,6 +261,7 @@ test("compaction events reduce to compaction message only when completed", () =>
   Effect.runSync(
     SessionMessageUpdater.update(SessionMessageUpdater.memory(state), {
       id: EventV2.ID.create(),
+      durable: durable(sessionID, 3),
       type: "session.next.compaction.ended",
       data: {
         sessionID,

packages/protocol/src/groups/event.ts

@@ -8,24 +8,34 @@ import { HttpApiEndpoint, HttpApiGroup, OpenApi } from "effect/unstable/httpapi"
 const fields = {
   id: Event.ID,
   metadata: Schema.optional(Schema.Record(Schema.String, Schema.Unknown)),
-  durable: Schema.optional(Schema.Struct({ aggregateID: Schema.String, seq: Schema.Int, version: Schema.Int })),
   location: Schema.optional(Location.Ref),
 }
 
+const DurableEnvelope = Schema.Struct({ aggregateID: Schema.String, seq: Schema.Int, version: Schema.Int })
+
 const schema = (definitions: ReadonlyArray<Definition>) =>
   Schema.Union([
     ...definitions.map((definition) =>
-      Schema.Struct({
-        ...fields,
-        type: Schema.Literal(definition.type),
-        data: definition.data,
-      }).annotate({ identifier: `V2Event.${definition.type}` }),
+      definition.durable
+        ? Schema.Struct({
+            ...fields,
+            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}` }),
     ),
     ...(definitions.some((definition) => definition.type === "server.connected")
       ? []
       : [
           Schema.Struct({
             ...fields,
+            durable: Schema.optional(Schema.Never),
             type: Schema.Literal("server.connected"),
             data: Schema.Struct({}),
           }).annotate({ identifier: "V2Event.server.connected" }),
@@ -56,4 +66,5 @@ export const makeEventGroup = (definitions: ReadonlyArray<Definition>) => 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

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)
+  })
+})

packages/schema/src/event.ts

@@ -11,62 +11,119 @@ export const ID = Schema.String.check(Schema.isStartsWith("evt_")).pipe(
 )
 export type ID = typeof ID.Type
 
-export type Definition<
+export type DurableOptions = {
+  readonly version: number
+  readonly aggregate: string
+}
+
+export type DurableEnvelope = {
+  readonly aggregateID: string
+  readonly seq: number
+  readonly version: number
+}
+
+const PublishedDurableEnvelope = Schema.Struct({
+  aggregateID: Schema.String,
+  seq: Schema.Number,
+  version: Schema.Number,
+})
+const NoDurableEnvelope = Schema.optional(Schema.Never)
+
+export type LiveDefinition<
   Type extends string = string,
   DataSchema extends Schema.Codec<unknown, unknown> = Schema.Codec<unknown, unknown>,
 > = Schema.Top & {
   readonly type: Type
-  readonly durable?: {
-    readonly version: number
-    readonly aggregate: string
-  }
   readonly data: DataSchema
+  readonly durable?: never
 }
 
+export type DurableDefinition<
+  Type extends string = string,
+  DataSchema extends Schema.Codec<unknown, unknown> = Schema.Codec<unknown, unknown>,
+  Durability extends DurableOptions = DurableOptions,
+> = Schema.Top & {
+  readonly type: Type
+  readonly data: DataSchema
+  readonly durable: Durability
+}
+
+export type Definition<
+  Type extends string = string,
+  DataSchema extends Schema.Codec<unknown, unknown> = Schema.Codec<unknown, unknown>,
+> = LiveDefinition<Type, DataSchema> | DurableDefinition<Type, DataSchema>
+
 export type Data<D extends Definition> = Schema.Schema.Type<D["data"]>
 
-export type Payload<D extends Definition = Definition> = {
-  readonly id: ID
-  readonly type: D["type"]
-  readonly data: Data<D>
-  readonly durable?: {
-    readonly aggregateID: string
-    readonly seq: number
-    readonly version: number
-  }
-  readonly location?: Location.Ref
-  readonly metadata?: Record<string, unknown>
-}
+export type UncommittedPayload<D extends Definition = Definition> = D extends Definition
+  ? {
+      readonly id: ID
+      readonly type: D["type"]
+      readonly data: Data<D>
+      readonly location?: Location.Ref
+      readonly metadata?: Record<string, unknown>
+    }
+  : never
+
+export type PublishedPayload<D extends Definition = Definition> = D extends DurableDefinition
+  ? UncommittedPayload<D> & { readonly durable: DurableEnvelope }
+  : D extends LiveDefinition
+    ? UncommittedPayload<D> & { readonly durable?: never }
+    : never
+
+export type Payload<D extends Definition = Definition> = PublishedPayload<D>
+
+type LiveEventSchema<
+  Type extends string,
+  Fields extends Readonly<Record<PropertyKey, Schema.Codec<unknown, unknown>>>,
+> = 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>>>,
+>(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?: {
-    readonly version: number
-    readonly aggregate: string
-  }
+  readonly durable: Durability
   readonly schema: Fields
-}): Schema.Schema<Payload<Definition<Type, Schema.Struct<Fields>>>> & Definition<Type, Schema.Struct<Fields>> {
+}): 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: Schema.optional(
-        Schema.Struct({ aggregateID: Schema.String, seq: Schema.Number, version: Schema.Number }),
-      ),
-      location: Schema.optional(Location.Ref),
-      data,
-    }).annotate({ identifier: input.type }),
-    {
-      type: input.type,
-      ...(input.durable === undefined ? {} : { durable: input.durable }),
-      data,
-    },
-  ) as Schema.Schema<Payload<Definition<Type, Schema.Struct<Fields>>>> & Definition<Type, Schema.Struct<Fields>>
+  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: PublishedDurableEnvelope }).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) {
@@ -103,7 +160,7 @@ export function durable(definitions: ReadonlyArray<Definition>) {
       if (result.has(key)) throw new Error(`Duplicate durable event definition for ${key}`)
       result.set(key, definition)
       return result
-    }, new Map<string, Definition>()),
+    }, new Map<string, DurableDefinition>()),
   )
 }
 

packages/schema/test/event.test.ts

@@ -34,4 +34,96 @@ describe("public event schemas", () => {
 
     expect(Event.durable([definition]).get("test.durable.1")).toBe(definition)
   })
+
+  test("durable definitions require published commit metadata", () => {
+    const definition = Event.define({
+      type: "test.durable",
+      durable: { aggregate: "id", version: 1 },
+      schema: { id: Schema.String },
+    })
+    const payload: typeof definition.Type = {
+      id: Event.ID.create(),
+      type: definition.type,
+      durable: { aggregateID: "aggregate", seq: 0, version: 1 },
+      data: { id: "aggregate" },
+    }
+
+    expect(Schema.is(definition)(payload)).toBe(true)
+    expect(
+      Schema.is(definition)({
+        id: Event.ID.create(),
+        type: definition.type,
+        data: { id: "aggregate" },
+      }),
+    ).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
+  })
+
+  test("live definitions reject durable commit metadata", () => {
+    const definition = Event.define({
+      type: "test.live",
+      schema: { value: Schema.String },
+    })
+    const payload: typeof definition.Type = {
+      id: Event.ID.create(),
+      type: definition.type,
+      data: { value: "value" },
+    }
+
+    expect(Schema.is(definition)(payload)).toBe(true)
+    expect(
+      Schema.is(definition)({
+        ...payload,
+        durable: { aggregateID: "aggregate", seq: 0, version: 1 },
+      }),
+    ).toBe(false)
+
+    const invalid: typeof definition.Type = {
+      ...payload,
+      // @ts-expect-error Live payloads cannot carry durable commit metadata.
+      durable: { aggregateID: "aggregate", seq: 0, version: 1 },
+    }
+    void invalid
+  })
+
+  test("mixed definition payloads preserve durability correlation", () => {
+    const durable = Event.define({
+      type: "test.mixed.durable",
+      durable: { aggregate: "id", version: 2 },
+      schema: { id: Schema.String },
+    })
+    const live = Event.define({
+      type: "test.mixed.live",
+      schema: { value: Schema.String },
+    })
+    type Mixed = Event.Payload<typeof durable | typeof live>
+
+    const committed: Mixed = {
+      id: Event.ID.create(),
+      type: durable.type,
+      durable: { aggregateID: "aggregate", seq: 0, version: 2 },
+      data: { id: "aggregate" },
+    }
+    const ephemeral: Mixed = {
+      id: Event.ID.create(),
+      type: live.type,
+      data: { value: "value" },
+    }
+    void committed
+    void ephemeral
+
+    // @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,
+      durable: { aggregateID: "aggregate", seq: 0, version: 2 },
+      data: { value: "value" },
+    }
+    void uncommitted
+    void falselyCommitted
+  })
 })

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`