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

Author: thdxr

packages/core/src/event.ts

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

packages/core/src/public/session.ts

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

packages/core/src/session.ts

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

packages/core/src/session/event.ts

@@ -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: {

packages/core/src/session/input.ts

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

packages/core/src/session/projector.ts

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