diff --git a/packages/core/src/session/history.ts b/packages/core/src/session/history.ts index 637ca0ebaed..14b36d4e4a7 100644 --- a/packages/core/src/session/history.ts +++ b/packages/core/src/session/history.ts @@ -29,26 +29,6 @@ export const latestCompaction = Effect.fnUntraced(function* (db: DatabaseService .pipe(Effect.orDie) }) -const messageRows = Effect.fnUntraced(function* ( - db: DatabaseService, - sessionID: SessionSchema.ID, - compaction: { readonly seq: number } | undefined, -) { - const rows = yield* db - .select() - .from(SessionMessageTable) - .where( - and( - eq(SessionMessageTable.session_id, sessionID), - compaction ? gte(SessionMessageTable.seq, compaction.seq) : undefined, - ), - ) - .orderBy(asc(SessionMessageTable.seq)) - .all() - .pipe(Effect.orDie) - return rows -}) - const decodeMessageRow = (row: typeof SessionMessageTable.$inferSelect) => decode({ ...row.data, id: row.id, type: row.type }).pipe( Effect.mapError( @@ -61,7 +41,19 @@ const decodeMessageRow = (row: typeof SessionMessageTable.$inferSelect) => ) const messageEntries = Effect.fnUntraced(function* (db: DatabaseService, sessionID: SessionSchema.ID) { - const rows = yield* messageRows(db, sessionID, yield* latestCompaction(db, sessionID)) + const compaction = yield* latestCompaction(db, sessionID) + const rows = yield* db + .select() + .from(SessionMessageTable) + .where( + and( + eq(SessionMessageTable.session_id, sessionID), + compaction ? gte(SessionMessageTable.seq, compaction.seq) : undefined, + ), + ) + .orderBy(asc(SessionMessageTable.seq)) + .all() + .pipe(Effect.orDie) return yield* Effect.forEach(rows, (row) => decodeMessageRow(row).pipe(Effect.map((message) => ({ seq: row.seq, message }))), ) diff --git a/packages/core/src/session/message-updater.ts b/packages/core/src/session/message-updater.ts index ba3a7e0eb4c..767f0ac45fe 100644 --- a/packages/core/src/session/message-updater.ts +++ b/packages/core/src/session/message-updater.ts @@ -61,8 +61,8 @@ export function update(adapter: Adapter, event: SessionEvent.DurableEvent) { Match.discriminatorsExhaustive("type")({ "session.created": () => Effect.void, "session.usage.recorded": () => Effect.void, - "session.agent.selected": (event) => { - return Effect.gen(function* () { + "session.agent.selected": (event) => + Effect.gen(function* () { const previous = event.data.previous ?? (yield* adapter.getAgent()) yield* adapter.appendMessage( SessionMessage.AgentSelected.make({ @@ -74,10 +74,9 @@ export function update(adapter: Adapter, event: SessionEvent.DurableEvent) { time: { created }, }), ) - }) - }, - "session.model.selected": (event) => { - return Effect.gen(function* () { + }), + "session.model.selected": (event) => + Effect.gen(function* () { const previous = event.data.previous ?? (yield* adapter.getModel()) yield* adapter.appendMessage( SessionMessage.ModelSelected.make({ @@ -89,10 +88,9 @@ export function update(adapter: Adapter, event: SessionEvent.DurableEvent) { time: { created }, }), ) - }) - }, - "session.moved": (event) => { - return Effect.gen(function* () { + }), + "session.moved": (event) => + Effect.gen(function* () { yield* adapter.appendMessage( SessionMessage.LocationSwitched.make({ id: SessionMessage.ID.fromEvent(event.id), @@ -105,8 +103,7 @@ export function update(adapter: Adapter, event: SessionEvent.DurableEvent) { time: { created }, }), ) - }) - }, + }), "session.renamed": () => Effect.void, "session.deleted": () => Effect.void, "session.forked": () => Effect.void, @@ -169,8 +166,8 @@ export function update(adapter: Adapter, event: SessionEvent.DurableEvent) { }), ) }, - "session.shell.ended": (event) => { - return Effect.gen(function* () { + "session.shell.ended": (event) => + Effect.gen(function* () { const currentShell = yield* adapter.getShell(event.data.shell.id) if (currentShell) { yield* adapter.updateShell( @@ -182,10 +179,9 @@ export function update(adapter: Adapter, event: SessionEvent.DurableEvent) { }), ) } - }) - }, - "session.step.started": (event) => { - return Effect.gen(function* () { + }), + "session.step.started": (event) => + Effect.gen(function* () { const existing = yield* adapter.getAssistant(event.data.assistantMessageID) if (existing) { yield* adapter.updateAssistant( @@ -224,8 +220,7 @@ export function update(adapter: Adapter, event: SessionEvent.DurableEvent) { snapshot: event.data.snapshot ? { start: event.data.snapshot } : undefined, }), ) - }) - }, + }), "session.step.ended": (event) => { return updateOwnedAssistant(event.data.assistantMessageID, (draft) => { draft.time.completed = created @@ -399,8 +394,8 @@ export function update(adapter: Adapter, event: SessionEvent.DurableEvent) { time: { created }, }), ), - "session.compaction.ended": (event) => { - return Effect.gen(function* () { + "session.compaction.ended": (event) => + Effect.gen(function* () { const current = yield* adapter.getCompaction() if (current?.status === "running") { yield* adapter.updateCompaction({ @@ -424,8 +419,7 @@ export function update(adapter: Adapter, event: SessionEvent.DurableEvent) { time: { created }, }), ) - }) - }, + }), "session.compaction.failed": (event) => Effect.gen(function* () { const current = yield* adapter.getCompaction() diff --git a/packages/core/src/session/store.ts b/packages/core/src/session/store.ts index d1e0b150aae..d62e2dd0730 100644 --- a/packages/core/src/session/store.ts +++ b/packages/core/src/session/store.ts @@ -58,9 +58,7 @@ const layer = Layer.effect( const row = yield* db.select().from(SessionTable).where(eq(SessionTable.id, sessionID)).get().pipe(Effect.orDie) return row ? fromRow(row) : undefined }), - context: Effect.fn("SessionStore.context")(function* (sessionID) { - return yield* SessionHistory.load(db, sessionID) - }), + context: Effect.fn("SessionStore.context")((sessionID) => SessionHistory.load(db, sessionID)), message: Effect.fn("SessionStore.message")(function* (messageID) { const row = yield* db .select()