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

refactor(core): simplify event model (#33238)

Browse source
This commit is contained in:
Dax 2026-06-21 16:34:57 +02:00 • committed by GitHub
commit fb43c15f88
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
15 changed files with 231 additions and 569 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -5,7 +5,7 @@ import { and, asc, eq, gt } from "drizzle-orm"
import { Database } from "./database/database"
import { EventSequenceTable, EventTable } from "./event/sql"
import { Location } from "./location"
import { externalID, type ExternalID, NonNegativeInt, withStatics } from "./schema"
import { externalID, type ExternalID, withStatics } from "./schema"
import { Identifier } from "./util/identifier"
import { LayerNode } from "./effect/layer-node"
import { isDeepStrictEqual } from "node:util"
@ -19,16 +19,9 @@ export const ID = Schema.String.check(Schema.isStartsWith("evt_")).pipe(
)
export type ID = typeof ID.Type
/**
* Durable aggregate continuation position for embedded replay streams.
* TODO: Decide whether a future HTTP / SDK surface should expose an opaque cursor instead.
*/
export const Cursor = NonNegativeInt.pipe(Schema.brand("EventV2.Cursor"))
export type Cursor = typeof Cursor.Type
export type Definition<Type extends string = string, DataSchema extends Schema.Top = Schema.Top> = {
readonly type: Type
readonly sync?: {
readonly durable?: {
readonly version: number
readonly aggregate: string
}
@ -41,20 +34,16 @@ export type Payload<D extends Definition = Definition> = {
readonly id: ID
readonly type: D["type"]
readonly data: Data<D>
/** Durable aggregate order, populated while synchronized events are projected. */
readonly seq?: number
readonly version?: number
readonly durable?: {
readonly aggregateID: string
readonly seq: number
readonly version: number
}
readonly location?: Location.Ref
readonly metadata?: Record<string, unknown>
/** Internal replay marker for projectors that own non-replicated operational state. */
readonly replay?: boolean
}
export type Projector<D extends Definition = Definition> = (event: Payload<D>) => Effect.Effect<void>
type AnyProjector = (event: Payload) => Effect.Effect<void>
export type CommitGuard = (event: Payload) => Effect.Effect<void>
export type Listener = (event: Payload) => Effect.Effect<void>
export type Sync = (event: Payload) => Effect.Effect<void>
export type Subscriber<D extends Definition = Definition> = (event: Payload<D>) => Effect.Effect<void>
export type Unsubscribe = Effect.Effect<void>
export type SerializedEvent = {
@ -65,13 +54,8 @@ export type SerializedEvent = {
readonly data: Record<string, unknown>
}
export type CursorEvent<E extends Payload = Payload> = {
readonly cursor: Cursor
readonly event: E
}
export class InvalidSyncEventError extends Schema.TaggedErrorClass<InvalidSyncEventError>()(
"EventV2.InvalidSyncEvent",
export class InvalidDurableEventError extends Schema.TaggedErrorClass<InvalidDurableEventError>()(
"EventV2.InvalidDurableEvent",
{
type: Schema.String,
message: Schema.String,
@ -83,19 +67,11 @@ export function versionedType(type: string, version: number) {
}
export const registry = new Map<string, Definition>()
type SyncDefinition = Definition & {
readonly sync: NonNullable<Definition["sync"]>
readonly encode: (data: unknown) => unknown
readonly decode: (data: unknown) => unknown
}
const syncRegistry = new Map<string, SyncDefinition>()
// Synchronized events cross a JSON boundary, so their data schemas must encode and decode without services.
const syncCodec = (definition: Definition) => definition.data as Schema.Codec<unknown, unknown, never, never>
const durableRegistry = new Map<string, Definition>()
export function define<const Type extends string, Fields extends Schema.Struct.Fields>(input: {
readonly type: Type
readonly sync?: {
readonly durable?: {
readonly version: number
readonly aggregate: string
}
@ -106,28 +82,26 @@ export function define<const Type extends string, Fields extends Schema.Struct.F
id: ID,
metadata: Schema.optional(Schema.Record(Schema.String, Schema.Unknown)),
type: Schema.Literal(input.type),
version: Schema.optional(Schema.Number),
durable: Schema.optional(Schema.Struct({ aggregateID: Schema.String, seq: Schema.Number, version: Schema.Number })),
location: Schema.optional(Location.Ref),
data: Data,
}).annotate({ identifier: input.type })
const definition = Object.assign(Payload, {
type: input.type,
...(input.sync === undefined ? {} : { sync: input.sync }),
...(input.durable === undefined ? {} : { durable: input.durable }),
data: Data,
})
const existing = registry.get(input.type)
if (input.sync === undefined || existing?.sync === undefined || input.sync.version >= existing.sync.version) {
if (
input.durable === undefined ||
existing?.durable === undefined ||
input.durable.version >= existing.durable.version
) {
registry.set(input.type, definition)
}
if (input.sync)
syncRegistry.set(
versionedType(input.type, input.sync.version),
Object.assign(definition, {
encode: Schema.encodeUnknownSync(syncCodec(definition)),
decode: Schema.decodeUnknownSync(syncCodec(definition)),
}) as SyncDefinition,
)
if (input.durable)
durableRegistry.set(versionedType(input.type, input.durable.version), definition)
return definition as Schema.Schema<Payload<Definition<Type, Schema.Struct<Fields>>>> &
Definition<Type, Schema.Struct<Fields>>
}
@ -140,7 +114,7 @@ export interface PublishOptions {
readonly id?: ID
readonly metadata?: Record<string, unknown>
readonly location?: Location.Ref
/** Local operational projection committed atomically with a new synchronized event. Not replayed or serialized. */
/** Local operational projection committed atomically with a new durable event. Not replayed or serialized. */
readonly commit?: (seq: number) => Effect.Effect<void>
}
@ -152,14 +126,13 @@ export interface Interface {
) => Effect.Effect<Payload<D>>
readonly subscribe: <D extends Definition>(definition: D) => Stream.Stream<Payload<D>>
readonly all: () => Stream.Stream<Payload>
readonly aggregateEvents: (input: {
readonly durable: (input: {
readonly aggregateID: string
readonly after?: Cursor
}) => Stream.Stream<CursorEvent>
readonly sync: (handler: Sync) => Effect.Effect<Unsubscribe>
readonly listen: (listener: Listener) => Effect.Effect<Unsubscribe>
readonly beforeCommit: (guard: CommitGuard) => Effect.Effect<void>
readonly project: <D extends Definition>(definition: D, projector: Projector<D>) => Effect.Effect<void>
readonly after?: number
}) => Stream.Stream<Payload>
/** @deprecated Use `all()` and consume the returned stream. */
readonly listen: (listener: Subscriber) => Effect.Effect<Unsubscribe>
readonly project: <D extends Definition>(definition: D, projector: Subscriber<D>) => Effect.Effect<void>
readonly replay: (
event: SerializedEvent,
options?: { readonly publish?: boolean; readonly ownerID?: string; readonly strictOwner?: boolean },
@ -182,37 +155,37 @@ export const layerWith = (options?: LayerOptions) =>
Layer.effect(
Service,
Effect.gen(function* () {
const all = yield* PubSub.unbounded<Payload>()
const synchronized = new Map<string, Set<PubSub.PubSub<void>>>()
const typed = new Map<string, PubSub.PubSub<Payload>>()
const projectors = new Map<string, AnyProjector[]>()
const commitGuards = new Array<CommitGuard>()
const listeners = new Array<Listener>()
const syncHandlers = new Array<Sync>()
const pubsub = {
all: yield* PubSub.unbounded<Payload>(),
durable: new Map<string, Set<PubSub.PubSub<void>>>(),
typed: new Map<string, PubSub.PubSub<Payload>>(),
}
const projectors = new Map<string, Subscriber[]>()
const listeners = new Array<Subscriber>()
const { db } = yield* Database.Service
const getOrCreate = (definition: Definition) =>
Effect.gen(function* () {
const existing = typed.get(definition.type)
const existing = pubsub.typed.get(definition.type)
if (existing) return existing
const pubsub = yield* PubSub.unbounded<Payload>()
typed.set(definition.type, pubsub)
return pubsub
const created = yield* PubSub.unbounded<Payload>()
pubsub.typed.set(definition.type, created)
return created
})
yield* Effect.addFinalizer(() =>
Effect.gen(function* () {
yield* PubSub.shutdown(all)
yield* PubSub.shutdown(pubsub.all)
yield* Effect.forEach(
synchronized.values(),
pubsub.durable.values(),
(pubsubs) => Effect.forEach(pubsubs, PubSub.shutdown, { discard: true }),
{ discard: true },
)
yield* Effect.forEach(typed.values(), PubSub.shutdown, { discard: true })
yield* Effect.forEach(pubsub.typed.values(), PubSub.shutdown, { discard: true })
}),
)
function commitSyncEvent(
function commitDurableEvent(
event: Payload,
input?: {
readonly seq: number
@ -224,28 +197,20 @@ export const layerWith = (options?: LayerOptions) =>
) {
return Effect.gen(function* () {
const definition = registry.get(event.type)
const sync = definition?.sync
if (sync) {
if (event.version !== sync.version) {
yield* Effect.die(
new InvalidSyncEventError({
type: event.type,
message: `Expected event version ${sync.version}, got ${event.version}`,
}),
)
}
const aggregateID = (event.data as Record<string, unknown>)[sync.aggregate]
const durable = definition?.durable
if (durable) {
const aggregateID = (event.data as Record<string, unknown>)[durable.aggregate]
if (typeof aggregateID !== "string") {
yield* Effect.die(
new InvalidSyncEventError({
new InvalidDurableEventError({
type: event.type,
message: `Expected string aggregate field ${sync.aggregate}`,
message: `Expected string aggregate field ${durable.aggregate}`,
}),
)
} else {
if (input && input.aggregateID !== aggregateID) {
yield* Effect.die(
new InvalidSyncEventError({
new InvalidDurableEventError({
type: event.type,
message: `Aggregate mismatch: expected ${input.aggregateID}, got ${aggregateID}`,
}),
@ -265,12 +230,12 @@ export const layerWith = (options?: LayerOptions) =>
.get()
.pipe(Effect.orDie)
const latest = row?.seq ?? -1
const encoded = syncRegistry
.get(versionedType(definition.type, sync.version))!
.encode(event.data) as Record<string, unknown>
const encoded = Schema.encodeUnknownSync(
definition.data as Schema.Codec<unknown, unknown, never, never>,
)(event.data) as Record<string, unknown>
if (input?.strictOwner && row?.ownerID && row.ownerID !== input.ownerID) {
yield* Effect.die(
new InvalidSyncEventError({
new InvalidDurableEventError({
type: event.type,
message: `Replay owner mismatch for aggregate ${aggregateID}: expected ${row.ownerID}, got ${input.ownerID ?? "none"}`,
}),
@ -285,7 +250,7 @@ export const layerWith = (options?: LayerOptions) =>
.pipe(Effect.orDie)
if (
stored?.id === event.id &&
stored.type === versionedType(definition.type, sync.version) &&
stored.type === versionedType(definition.type, durable.version) &&
isDeepStrictEqual(stored.data, encoded)
) {
if (input.ownerID && row?.ownerID == null) {
@ -299,7 +264,7 @@ export const layerWith = (options?: LayerOptions) =>
return
}
yield* Effect.die(
new InvalidSyncEventError({
new InvalidDurableEventError({
type: event.type,
message: `Replay diverged at aggregate ${aggregateID} sequence ${input.seq}`,
}),
@ -311,7 +276,7 @@ export const layerWith = (options?: LayerOptions) =>
const seq = input?.seq ?? latest + 1
if (input && seq !== latest + 1) {
yield* Effect.die(
new InvalidSyncEventError({
new InvalidDurableEventError({
type: event.type,
message: `Sequence mismatch for aggregate ${aggregateID}: expected ${latest + 1}, got ${seq}`,
}),
@ -325,16 +290,17 @@ export const layerWith = (options?: LayerOptions) =>
.pipe(Effect.orDie)
if (stored)
yield* Effect.die(
new InvalidSyncEventError({
new InvalidDurableEventError({
type: event.type,
message: `Event ${event.id} already exists at aggregate ${stored.aggregateID} sequence ${stored.seq}`,
}),
)
for (const guard of commitGuards) {
yield* guard(event)
}
const committed = {
...event,
durable: { aggregateID, seq, version: durable.version },
} as Payload
for (const projector of list) {
yield* projector({ ...event, seq } as Payload)
yield* projector(committed)
}
if (commit) yield* commit(seq)
yield* db
@ -356,7 +322,7 @@ export const layerWith = (options?: LayerOptions) =>
id: event.id,
aggregate_id: aggregateID,
seq,
type: versionedType(definition.type, sync.version),
type: versionedType(definition.type, durable.version),
data: encoded,
},
])
@ -369,8 +335,8 @@ export const layerWith = (options?: LayerOptions) =>
.pipe(Effect.orDie)
if (committed) {
yield* Effect.forEach(
synchronized.get(committed.aggregateID) ?? [],
(pubsub) => PubSub.publish(pubsub, undefined),
pubsub.durable.get(committed.aggregateID) ?? [],
(wake) => PubSub.publish(wake, undefined),
{ discard: true },
)
}
@ -384,19 +350,25 @@ export const layerWith = (options?: LayerOptions) =>
function publishEvent<D extends Definition>(event: Payload<D>, commit?: PublishOptions["commit"]) {
return Effect.gen(function* () {
const durable = registry.get(event.type)?.sync !== undefined
if (!durable && commit)
const definition = registry.get(event.type)
if (!definition?.durable && commit)
return yield* Effect.die(
new InvalidSyncEventError({
new InvalidDurableEventError({
type: event.type,
message: "Local commit hooks require a synchronized event",
message: "Local commit hooks require a durable event",
}),
)
if (durable) {
const committed = yield* commitSyncEvent(event as Payload, undefined, commit)
if (definition?.durable) {
const committed = yield* commitDurableEvent(event as Payload, undefined, commit)
if (committed) {
event = { ...event, seq: committed.seq }
yield* Effect.forEach(syncHandlers, (sync) => observe(event as Payload, "sync", sync), { discard: true })
event = {
...event,
durable: {
aggregateID: committed.aggregateID,
seq: committed.seq,
version: definition.durable.version,
},
}
yield* notify(event as Payload, true)
return event
}
@ -406,12 +378,12 @@ export const layerWith = (options?: LayerOptions) =>
})
}
const observe = (event: Payload, kind: "sync" | "listener", observer: (event: Payload) => Effect.Effect<void>) =>
const observe = (event: Payload, observer: (event: Payload) => Effect.Effect<void>) =>
Effect.suspend(() => observer(event)).pipe(
Effect.catchCauseIf(
(cause) => !Cause.hasInterrupts(cause),
(cause) =>
Effect.logError("Event observer failed", { eventID: event.id, eventType: event.type, kind, cause }),
Effect.logError("Event listener failed", { eventID: event.id, eventType: event.type, cause }),
),
)
@ -419,12 +391,12 @@ export const layerWith = (options?: LayerOptions) =>
return Effect.gen(function* () {
yield* Effect.forEach(
listeners,
(listener) => (isolateListeners ? observe(event, "listener", listener) : listener(event)),
(listener) => (isolateListeners ? observe(event, listener) : listener(event)),
{ discard: true },
)
const pubsub = typed.get(event.type)
if (pubsub) yield* PubSub.publish(pubsub, event)
yield* PubSub.publish(all, event)
const typed = pubsub.typed.get(event.type)
if (typed) yield* PubSub.publish(typed, event)
yield* PubSub.publish(pubsub.all, event)
})
}
@ -441,7 +413,6 @@ export const layerWith = (options?: LayerOptions) =>
id: options?.id ?? ID.create(),
...(options?.metadata ? { metadata: options.metadata } : {}),
type: definition.type,
...(definition.sync === undefined ? {} : { version: definition.sync.version }),
...(location ? { location } : {}),
data,
} as Payload<D>,
@ -455,27 +426,37 @@ export const layerWith = (options?: LayerOptions) =>
options?: { readonly publish?: boolean; readonly ownerID?: string; readonly strictOwner?: boolean },
) {
return Effect.gen(function* () {
const definition = syncRegistry.get(event.type)
if (!definition) {
const definition = durableRegistry.get(event.type)
if (!definition?.durable) {
yield* Effect.die(
new InvalidSyncEventError({ type: event.type, message: `Unknown sync event type ${event.type}` }),
new InvalidDurableEventError({ type: event.type, message: `Unknown durable event type ${event.type}` }),
)
} else {
const payload = {
id: event.id,
type: definition.type,
version: definition.sync.version,
data: definition.decode(event.data),
replay: true,
data: Schema.decodeUnknownSync(
definition.data as Schema.Codec<unknown, unknown, never, never>,
)(event.data),
} as Payload
const committed = yield* commitSyncEvent(payload, {
const committed = yield* commitDurableEvent(payload, {
seq: event.seq,
aggregateID: event.aggregateID,
ownerID: options?.ownerID,
strictOwner: options?.strictOwner,
})
if (committed && options?.publish) {
yield* notify({ ...payload, seq: committed.seq }, true)
yield* notify(
{
...payload,
durable: {
aggregateID: committed.aggregateID,
seq: committed.seq,
version: definition.durable.version,
},
},
true,
)
}
}
})
@ -490,7 +471,7 @@ export const layerWith = (options?: LayerOptions) =>
if (!source) return undefined
if (events.some((event) => event.aggregateID !== source)) {
yield* Effect.die(
new InvalidSyncEventError({
new InvalidDurableEventError({
type: events[0]?.type ?? "unknown",
message: "Replay events must belong to the same aggregate",
}),
@ -501,7 +482,7 @@ export const layerWith = (options?: LayerOptions) =>
const seq = start + index
if (event.seq !== seq) {
yield* Effect.die(
new InvalidSyncEventError({
new InvalidDurableEventError({
type: event.type,
message: `Replay sequence mismatch at index ${index}: expected ${seq}, got ${event.seq}`,
}),
@ -540,22 +521,18 @@ export const layerWith = (options?: LayerOptions) =>
Stream.map((event) => event as Payload<D>),
)
const streamAll = (): Stream.Stream<Payload> => Stream.fromPubSub(all)
const streamAll = (): Stream.Stream<Payload> => Stream.fromPubSub(pubsub.all)
const decodeSerializedEvent = (event: SerializedEvent): CursorEvent => {
const definition = syncRegistry.get(event.type)
if (!definition) {
throw new InvalidSyncEventError({ type: event.type, message: `Unknown sync event type ${event.type}` })
const decodeSerializedEvent = (event: SerializedEvent): Payload => {
const definition = durableRegistry.get(event.type)
if (!definition?.durable) {
throw new InvalidDurableEventError({ type: event.type, message: `Unknown durable event type ${event.type}` })
}
return {
cursor: Cursor.make(event.seq),
event: {
id: event.id,
type: definition.type,
version: definition.sync.version,
seq: event.seq,
data: definition.decode(event.data),
},
id: event.id,
type: definition.type,
durable: { aggregateID: event.aggregateID, seq: event.seq, version: definition.durable.version },
data: Schema.decodeUnknownSync(definition.data as Schema.Codec<unknown, unknown, never, never>)(event.data),
}
}
@ -583,43 +560,43 @@ export const layerWith = (options?: LayerOptions) =>
),
)
const subscribeSynchronized = (aggregateID: string) =>
const subscribeDurable = (aggregateID: string) =>
Effect.gen(function* () {
const pubsub = yield* PubSub.sliding<void>(1)
const subscription = yield* PubSub.subscribe(pubsub)
const wake = yield* PubSub.sliding<void>(1)
const subscription = yield* PubSub.subscribe(wake)
yield* Effect.acquireRelease(
Effect.sync(() => {
const pubsubs = synchronized.get(aggregateID) ?? new Set()
pubsubs.add(pubsub)
synchronized.set(aggregateID, pubsubs)
const wakes = pubsub.durable.get(aggregateID) ?? new Set()
wakes.add(wake)
pubsub.durable.set(aggregateID, wakes)
}),
() =>
Effect.sync(() => {
const pubsubs = synchronized.get(aggregateID)
pubsubs?.delete(pubsub)
if (pubsubs?.size === 0) synchronized.delete(aggregateID)
}).pipe(Effect.andThen(PubSub.shutdown(pubsub))),
const wakes = pubsub.durable.get(aggregateID)
wakes?.delete(wake)
if (wakes?.size === 0) pubsub.durable.delete(aggregateID)
}).pipe(Effect.andThen(PubSub.shutdown(wake))),
)
return subscription
})
const streamEvents = (input: {
const durable = (input: {
readonly aggregateID: string
readonly after?: Cursor
}): Stream.Stream<CursorEvent> =>
readonly after?: number
}): Stream.Stream<Payload> =>
Stream.unwrap(
Effect.gen(function* () {
const synchronized = yield* subscribeSynchronized(input.aggregateID)
let cursor = input.after ?? -1
const read = Effect.suspend(() => readAfter(input.aggregateID, cursor)).pipe(
const wakes = yield* subscribeDurable(input.aggregateID)
let sequence = input.after ?? -1
const read = Effect.suspend(() => readAfter(input.aggregateID, sequence)).pipe(
Effect.tap((events) =>
Effect.sync(() => {
cursor = events.at(-1)?.cursor ?? cursor
sequence = events.at(-1)?.durable?.seq ?? sequence
}),
),
)
const historical = yield* read
const live = Stream.fromSubscription(synchronized).pipe(
const live = Stream.fromSubscription(wakes).pipe(
Stream.mapEffect(() => read),
Stream.flattenIterable,
)
@ -627,7 +604,7 @@ export const layerWith = (options?: LayerOptions) =>
}),
)
const listen = (listener: Listener): Effect.Effect<Unsubscribe> =>
const listen = (listener: Subscriber): Effect.Effect<Unsubscribe> =>
Effect.sync(() => {
listeners.push(listener)
return Effect.sync(() => {
@ -636,21 +613,7 @@ export const layerWith = (options?: LayerOptions) =>
})
})
const sync = (handler: Sync): Effect.Effect<Unsubscribe> =>
Effect.sync(() => {
syncHandlers.push(handler)
return Effect.sync(() => {
const index = syncHandlers.indexOf(handler)
if (index >= 0) syncHandlers.splice(index, 1)
})
})
const beforeCommit = (guard: CommitGuard): Effect.Effect<void> =>
Effect.sync(() => {
commitGuards.push(guard)
})
const project = <D extends Definition>(definition: D, projector: Projector<D>): Effect.Effect<void> =>
const project = <D extends Definition>(definition: D, projector: Subscriber<D>): Effect.Effect<void> =>
Effect.sync(() => {
const list = projectors.get(definition.type) ?? []
list.push((event) => projector(event as Payload<D>))
@ -661,10 +624,8 @@ export const layerWith = (options?: LayerOptions) =>
publish,
subscribe,
all: streamAll,
aggregateEvents: streamEvents,
sync,
durable,
listen,
beforeCommit,
project,
replay,
replayAll,

7
packages/core/src/public/session.ts
View file

@ -1,7 +1,6 @@
export * as Session from "./session"
import { Effect, Schema, Stream } from "effect"
import { EventV2 } from "../event"
import { ModelV2 } from "../model"
import { SessionV2 } from "../session"
import { MessageDecodeError } from "../session/error"
@ -34,9 +33,7 @@ export type Delivery = SessionInput.Delivery
export const ListInput = SessionV2.ListInput
export type ListInput = SessionV2.ListInput
export const EventCursor = EventV2.Cursor
export type EventCursor = EventV2.Cursor
export type Event = EventV2.CursorEvent<SessionEvent.DurableEvent>
export type Event = SessionEvent.DurableEvent
export const NotFoundError = SessionV2.NotFoundError
export type NotFoundError = SessionV2.NotFoundError
@ -99,7 +96,7 @@ export interface MessageInput {
export interface EventsInput {
readonly sessionID: ID
readonly after?: EventCursor
readonly after?: number
}
export interface Interface {

14
packages/core/src/session.ts
View file

@ -124,8 +124,8 @@ export interface Interface {
) => Effect.Effect<SessionMessage.Message[], NotFoundError | MessageDecodeError>
readonly events: (input: {
sessionID: SessionSchema.ID
after?: EventV2.Cursor
}) => Stream.Stream<EventV2.CursorEvent<SessionEvent.DurableEvent>, NotFoundError>
after?: number
}) => Stream.Stream<SessionEvent.DurableEvent, NotFoundError>
readonly switchAgent: (input: {
sessionID: SessionSchema.ID
agent: string
@ -339,11 +339,9 @@ export const layer = Layer.effect(
Stream.unwrap(
result
.get(input.sessionID)
.pipe(Effect.as(events.aggregateEvents({ aggregateID: input.sessionID, after: input.after }))),
.pipe(Effect.as(events.durable({ aggregateID: input.sessionID, after: input.after }))),
).pipe(
Stream.filter((event): event is EventV2.CursorEvent<SessionEvent.DurableEvent> =>
isDurableSessionEvent(event.event),
),
Stream.filter((event): event is SessionEvent.DurableEvent => isDurableSessionEvent(event)),
),
prompt: Effect.fn("V2Session.prompt")((input) =>
Effect.uninterruptible(
@ -413,9 +411,9 @@ export const layer = Layer.effect(
sessionID,
timestamp: yield* DateTime.now,
})
if (event.seq === undefined)
if (event.durable === undefined)
return yield* Effect.die("Interrupt request event is missing aggregate sequence")
yield* execution.interrupt(sessionID, event.seq)
yield* execution.interrupt(sessionID, event.durable.seq)
}),
),
),

6
packages/core/src/session/event.ts
View file

@ -27,13 +27,13 @@ const Base = {
}
const options = {
sync: {
durable: {
aggregate: "sessionID",
version: 1,
},
} as const
const stepSettlementOptions = {
sync: {
durable: {
aggregate: "sessionID",
version: 2,
},
@ -456,7 +456,7 @@ export namespace Compaction {
export const Ended = EventV2.define({
type: "session.next.compaction.ended",
sync: { aggregate: "sessionID", version: 2 },
durable: { aggregate: "sessionID", version: 2 },
schema: {
...Base,
messageID: SessionMessageID.ID,

42
packages/core/src/session/input.ts
View file

@ -74,11 +74,11 @@ export const admit = Effect.fn("SessionInput.admit")(function* (
})
.pipe(
Effect.flatMap((event) =>
event.seq === undefined
event.durable === undefined
? Effect.die("Prompt admission event is missing aggregate sequence")
: Effect.succeed(
new Admitted({
admittedSeq: event.seq,
admittedSeq: event.durable.seq,
id: input.id,
sessionID: input.sessionID,
prompt: input.prompt,
@ -117,13 +117,6 @@ export const projectAdmitted = Effect.fn("SessionInput.projectAdmitted")(functio
readonly timeCreated: DateTime.Utc
},
) {
const message = yield* db
.select({ id: SessionMessageTable.id })
.from(SessionMessageTable)
.where(eq(SessionMessageTable.id, input.id))
.get()
.pipe(Effect.orDie)
if (message) return yield* Effect.die(new LifecycleConflict({ id: input.id }))
const stored = yield* db
.insert(SessionInputTable)
.values({
@ -208,37 +201,6 @@ const matchesPrompt = (input: Admitted, expected: { readonly sessionID: SessionS
input.sessionID === expected.sessionID &&
JSON.stringify(encodePrompt(input.prompt)) === JSON.stringify(encodePrompt(expected.prompt))
export const guardReservedID = Effect.fn("SessionInput.guardReservedID")(function* (
db: DatabaseService,
event: EventV2.Payload,
) {
if (
Schema.is(SessionEvent.PromptLifecycle.Admitted)(event) ||
Schema.is(SessionEvent.PromptLifecycle.Promoted)(event)
)
return
const id = reservedID(event)
if (id === undefined) return
const admitted = yield* db
.select({ id: SessionInputTable.id })
.from(SessionInputTable)
.where(eq(SessionInputTable.id, id))
.get()
.pipe(Effect.orDie)
if (admitted === undefined) return
return yield* Effect.die(new LifecycleConflict({ id }))
})
const reservedID = (event: EventV2.Payload) => {
if (Schema.is(SessionEvent.Step.Started)(event)) return event.data.assistantMessageID
if (Schema.is(SessionEvent.AgentSwitched)(event)) return event.data.messageID
if (Schema.is(SessionEvent.ModelSwitched)(event)) return event.data.messageID
if (Schema.is(SessionEvent.Prompted)(event)) return event.data.messageID
if (Schema.is(SessionEvent.Synthetic)(event)) return event.data.messageID
if (Schema.is(SessionEvent.Shell.Started)(event)) return event.data.messageID
if (Schema.is(SessionEvent.Compaction.Started)(event)) return event.data.messageID
}
export const projectLegacyPrompted = Effect.fn("SessionInput.projectLegacyPrompted")(function* (
db: DatabaseService,
input: {

49
packages/core/src/session/projector.ts
View file

@ -115,7 +115,7 @@ function run(db: DatabaseService, event: SessionEvent.Event) {
const decodeRow = (row: typeof SessionMessageTable.$inferSelect) =>
decodeMessage({ ...row.data, id: row.id, type: row.type })
const updateMessage = (message: SessionMessage.Message) => {
if (event.seq === undefined) return Effect.die("Synchronized Session event is missing aggregate sequence")
if (event.durable === undefined) return Effect.die("Durable Session event is missing aggregate sequence")
const encoded = encodeMessage(message)
const { id, type, ...data } = encoded
return db
@ -192,7 +192,7 @@ function run(db: DatabaseService, event: SessionEvent.Event) {
}
function insertMessage(db: DatabaseService, event: SessionEvent.Event, message: SessionMessage.Message) {
if (event.seq === undefined) return Effect.die("Synchronized Session event is missing aggregate sequence")
if (event.durable === undefined) return Effect.die("Durable Session event is missing aggregate sequence")
const encoded = encodeMessage(message)
const { id, type, ...data } = encoded
return db
@ -201,7 +201,7 @@ function insertMessage(db: DatabaseService, event: SessionEvent.Event, message:
id: SessionMessage.ID.make(id),
session_id: event.data.sessionID,
type,
seq: event.seq,
seq: event.durable.seq,
time_created: DateTime.toEpochMillis(message.time.created),
data,
})
@ -213,7 +213,6 @@ export const layer = Layer.effectDiscard(
Effect.gen(function* () {
const events = yield* EventV2.Service
const { db } = yield* Database.Service
yield* events.beforeCommit((event) => SessionInput.guardReservedID(db, event))
yield* events.project(SessionV1.Event.Created, (event) =>
Effect.gen(function* () {
const stored = yield* db
@ -331,7 +330,7 @@ export const layer = Layer.effectDiscard(
}),
)
yield* events.project(SessionEvent.AgentSwitched, (event) => {
if (event.seq === undefined) return Effect.die("Synchronized Session event is missing aggregate sequence")
if (event.durable === undefined) return Effect.die("Durable Session event is missing aggregate sequence")
return db
.update(SessionTable)
.set({ agent: event.data.agent, time_updated: DateTime.toEpochMillis(event.data.timestamp) })
@ -340,7 +339,7 @@ export const layer = Layer.effectDiscard(
.pipe(
Effect.orDie,
Effect.andThen(run(db, event)),
Effect.andThen(SessionContextEpoch.requestReplacement(db, event.data.sessionID, event.seq)),
Effect.andThen(SessionContextEpoch.requestReplacement(db, event.data.sessionID, event.durable.seq)),
)
})
yield* events.project(SessionEvent.ModelSwitched, (event) =>
@ -352,9 +351,9 @@ export const layer = Layer.effectDiscard(
.run()
.pipe(Effect.orDie)
yield* run(db, event)
if (event.seq === undefined)
return yield* Effect.die("Synchronized Session event is missing aggregate sequence")
yield* SessionContextEpoch.requestReplacement(db, event.data.sessionID, event.seq)
if (event.durable === undefined)
return yield* Effect.die("Durable Session event is missing aggregate sequence")
yield* SessionContextEpoch.requestReplacement(db, event.data.sessionID, event.durable.seq)
}),
)
yield* events.project(SessionEvent.Prompted, (event) =>
@ -368,24 +367,24 @@ export const layer = Layer.effectDiscard(
.pipe(Effect.orDie)
if (existing) return yield* Effect.die(new PromptAlreadyProjected())
yield* run(db, event)
if (event.seq === undefined)
return yield* Effect.die("Synchronized Session event is missing aggregate sequence")
if (event.durable === undefined)
return yield* Effect.die("Durable Session event is missing aggregate sequence")
yield* SessionInput.projectLegacyPrompted(db, {
id: messageID,
sessionID: event.data.sessionID,
prompt: event.data.prompt,
delivery: event.data.delivery,
timeCreated: event.data.timestamp,
promotedSeq: event.seq,
promotedSeq: event.durable.seq,
})
}),
)
yield* events.project(SessionEvent.PromptLifecycle.Admitted, (event) =>
Effect.gen(function* () {
if (event.seq === undefined)
return yield* Effect.die("Synchronized Session event is missing aggregate sequence")
if (event.durable === undefined)
return yield* Effect.die("Durable Session event is missing aggregate sequence")
yield* SessionInput.projectAdmitted(db, {
admittedSeq: event.seq,
admittedSeq: event.durable.seq,
id: event.data.messageID,
sessionID: event.data.sessionID,
prompt: event.data.prompt,
@ -396,8 +395,8 @@ export const layer = Layer.effectDiscard(
)
yield* events.project(SessionEvent.PromptLifecycle.Promoted, (event) =>
Effect.gen(function* () {
if (event.seq === undefined)
return yield* Effect.die("Synchronized Session event is missing aggregate sequence")
if (event.durable === undefined)
return yield* Effect.die("Durable Session event is missing aggregate sequence")
yield* insertMessage(
db,
event,
@ -406,18 +405,14 @@ export const layer = Layer.effectDiscard(
sessionID: event.data.sessionID,
prompt: event.data.prompt,
timeCreated: event.data.timeCreated,
promotedSeq: event.seq,
promotedSeq: event.durable.seq,
}),
)
}),
)
yield* events.project(SessionEvent.InterruptRequested, () => Effect.void)
yield* events.project(SessionEvent.ContextUpdated, (event) => {
if (!event.replay || event.seq === undefined) return run(db, event)
return run(db, event).pipe(
Effect.andThen(SessionContextEpoch.requestReplacement(db, event.data.sessionID, event.seq)),
)
})
// TODO: Reconstruct context epoch replacement state during replay without adding replay state to every EventV2 payload.
yield* events.project(SessionEvent.ContextUpdated, (event) => run(db, event))
yield* events.project(SessionEvent.Synthetic, (event) => run(db, event))
yield* events.project(SessionEvent.Shell.Started, (event) => run(db, event))
yield* events.project(SessionEvent.Shell.Ended, (event) => run(db, event))
@ -436,9 +431,9 @@ export const layer = Layer.effectDiscard(
yield* events.project(SessionEvent.Reasoning.Ended, (event) => run(db, event))
// yield* events.project(SessionEvent.Retried, (event) => run(db, event))
yield* events.project(SessionEvent.Compaction.Ended, (event) => {
if (event.version === 1) return Effect.void
const seq = event.seq
if (seq === undefined) return Effect.die("Synchronized Session event is missing aggregate sequence")
if (event.durable === undefined) return Effect.die("Durable Session event is missing aggregate sequence")
if (event.durable.version === 1) return Effect.void
const seq = event.durable.seq
return Effect.gen(function* () {
yield* run(db, event)
yield* SessionContextEpoch.requestReplacement(db, event.data.sessionID, seq)

2
packages/core/src/v1/session.ts
View file

@ -502,7 +502,7 @@ export type WithParts = {
}
const options = {
sync: {
durable: {
aggregate: "sessionID",
version: 1,
},