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

refactor(core): make v2 session inputs event sourced (#30785)

Browse source
This commit is contained in:
Kit Langton 2026-06-04 19:24:30 -04:00 • committed by GitHub
commit 76ecf2e58c
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
43 changed files with 4671 additions and 757 deletions
Download patch file Download diff file Expand all files Collapse all files

1
packages/core/src/database/migration.gen.ts generated
View file

@ -31,5 +31,6 @@ export const migrations = (
import("./migration/20260603040000_session_message_projection_order"),
import("./migration/20260603141458_session_input_inbox"),
import("./migration/20260603160727_jittery_ezekiel_stane"),
import("./migration/20260604172448_event_sourced_session_input"),
])
).map((module) => module.default) satisfies DatabaseMigration.Migration[]

47
packages/core/src/database/migration/20260604172448_event_sourced_session_input.ts Normal file
View file

@ -0,0 +1,47 @@
import { Effect } from "effect"
import type { DatabaseMigration } from "../migration"
export default {
id: "20260604172448_event_sourced_session_input",
up(tx) {
return Effect.gen(function* () {
yield* tx.run(`DELETE FROM \`session_input\`;`)
yield* tx.run(`DELETE FROM \`session_message\`;`)
yield* tx.run(`DELETE FROM \`event\`;`)
yield* tx.run(`DELETE FROM \`event_sequence\`;`)
yield* tx.run(`UPDATE \`session\` SET \`workspace_id\` = NULL;`)
yield* tx.run(`DELETE FROM \`workspace\`;`)
yield* tx.run(`DROP INDEX IF EXISTS \`event_aggregate_seq_idx\`;`)
yield* tx.run(`CREATE UNIQUE INDEX \`event_aggregate_seq_idx\` ON \`event\` (\`aggregate_id\`,\`seq\`);`)
yield* tx.run(`DROP INDEX IF EXISTS \`session_message_session_seq_idx\`;`)
yield* tx.run(
`CREATE UNIQUE INDEX \`session_message_session_seq_idx\` ON \`session_message\` (\`session_id\`,\`seq\`);`,
)
yield* tx.run(`PRAGMA foreign_keys=OFF;`)
yield* tx.run(`
CREATE TABLE \`__new_session_input\` (
\`id\` text PRIMARY KEY,
\`session_id\` text NOT NULL,
\`prompt\` text NOT NULL,
\`delivery\` text NOT NULL,
\`admitted_seq\` integer NOT NULL,
\`promoted_seq\` integer,
\`time_created\` integer NOT NULL,
CONSTRAINT \`fk_session_input_session_id_session_id_fk\` FOREIGN KEY (\`session_id\`) REFERENCES \`session\`(\`id\`) ON DELETE CASCADE
);
`)
yield* tx.run(`DROP TABLE \`session_input\`;`)
yield* tx.run(`ALTER TABLE \`__new_session_input\` RENAME TO \`session_input\`;`)
yield* tx.run(`PRAGMA foreign_keys=ON;`)
yield* tx.run(
`CREATE INDEX \`session_input_session_pending_delivery_seq_idx\` ON \`session_input\` (\`session_id\`,\`promoted_seq\`,\`delivery\`,\`admitted_seq\`);`,
)
yield* tx.run(
`CREATE UNIQUE INDEX \`session_input_session_admitted_seq_idx\` ON \`session_input\` (\`session_id\`,\`admitted_seq\`);`,
)
yield* tx.run(
`CREATE UNIQUE INDEX \`session_input_session_promoted_seq_idx\` ON \`session_input\` (\`session_id\`,\`promoted_seq\`);`,
)
})
},
} satisfies DatabaseMigration.Migration

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

@ -1,14 +1,15 @@
export * as EventV2 from "./event"
import { Context, Effect, Layer, Option, PubSub, Schema, Stream } from "effect"
import { Cause, Context, Effect, Layer, Option, PubSub, Schema, Stream } from "effect"
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 { Identifier } from "./util/identifier"
import { isDeepStrictEqual } from "node:util"
export const ID = Schema.String.pipe(
export const ID = Schema.String.check(Schema.isStartsWith("evt_")).pipe(
Schema.brand("Event.ID"),
withStatics((schema) => ({
create: () => schema.make("evt_" + Identifier.ascending()),
@ -258,16 +259,47 @@ export const layerWith = (options?: LayerOptions) =>
.get()
.pipe(Effect.orDie)
const latest = row?.seq ?? -1
if (input && input.seq <= latest) return
if (input && row?.ownerID && row.ownerID !== input.ownerID) {
if (input.strictOwner) {
yield* Effect.die(
new InvalidSyncEventError({
type: event.type,
message: `Replay owner mismatch for aggregate ${aggregateID}: expected ${row.ownerID}, got ${input.ownerID ?? "none"}`,
}),
)
const encoded = syncRegistry
.get(versionedType(definition.type, sync.version))!
.encode(event.data) as Record<string, unknown>
if (input?.strictOwner && row?.ownerID && row.ownerID !== input.ownerID) {
yield* Effect.die(
new InvalidSyncEventError({
type: event.type,
message: `Replay owner mismatch for aggregate ${aggregateID}: expected ${row.ownerID}, got ${input.ownerID ?? "none"}`,
}),
)
}
if (input && input.seq <= latest) {
const stored = yield* db
.select()
.from(EventTable)
.where(and(eq(EventTable.aggregate_id, aggregateID), eq(EventTable.seq, input.seq)))
.get()
.pipe(Effect.orDie)
if (
stored?.id === event.id &&
stored.type === versionedType(definition.type, sync.version) &&
isDeepStrictEqual(stored.data, encoded)
) {
if (input.ownerID && row?.ownerID == null) {
yield* db
.update(EventSequenceTable)
.set({ owner_id: input.ownerID })
.where(eq(EventSequenceTable.aggregate_id, aggregateID))
.run()
.pipe(Effect.orDie)
}
return
}
yield* Effect.die(
new InvalidSyncEventError({
type: event.type,
message: `Replay diverged at aggregate ${aggregateID} sequence ${input.seq}`,
}),
)
}
if (input && row?.ownerID && row.ownerID !== input.ownerID) {
return
}
const seq = input?.seq ?? latest + 1
@ -279,15 +311,25 @@ export const layerWith = (options?: LayerOptions) =>
}),
)
}
const stored = yield* db
.select({ aggregateID: EventTable.aggregate_id, seq: EventTable.seq })
.from(EventTable)
.where(eq(EventTable.id, event.id))
.get()
.pipe(Effect.orDie)
if (stored)
yield* Effect.die(
new InvalidSyncEventError({
type: event.type,
message: `Event ${event.id} already exists at aggregate ${stored.aggregateID} sequence ${stored.seq}`,
}),
)
for (const guard of commitGuards) {
yield* guard(event)
}
for (const projector of list) {
yield* projector({ ...event, seq } as Payload)
}
const encoded = syncRegistry
.get(versionedType(definition.type, sync.version))!
.encode(event.data)
yield* db
.insert(EventSequenceTable)
.values([{ aggregate_id: aggregateID, seq, owner_id: input?.ownerID }])
@ -308,7 +350,7 @@ export const layerWith = (options?: LayerOptions) =>
aggregate_id: aggregateID,
seq,
type: versionedType(definition.type, sync.version),
data: encoded as Record<string, unknown>,
data: encoded,
},
])
.run()
@ -337,22 +379,43 @@ export const layerWith = (options?: LayerOptions) =>
return Effect.gen(function* () {
const durable = registry.get(event.type)?.sync !== undefined
if (durable) {
for (const sync of syncHandlers) {
yield* sync(event as Payload)
}
const committed = yield* commitSyncEvent(event as Payload)
if (committed) event = { ...event, seq: committed.seq }
if (committed) {
event = { ...event, seq: committed.seq }
yield* Effect.forEach(syncHandlers, (sync) => observe(event as Payload, "sync", sync), { discard: true })
yield* notify(event as Payload, true)
return event
}
}
for (const listener of listeners) {
yield* listener(event as Payload)
}
const pubsub = typed.get(event.type)
if (pubsub) yield* PubSub.publish(pubsub, event as Payload)
yield* PubSub.publish(all, event as Payload)
yield* notify(event as Payload, false)
return event
})
}
const observe = (event: Payload, kind: "sync" | "listener", observer: (event: Payload) => Effect.Effect<void>) =>
Effect.suspend(() => observer(event)).pipe(
Effect.catchCauseIf(
(cause) => !Cause.hasInterrupts(cause),
(cause) =>
Effect.logError("Event observer failed").pipe(
Effect.annotateLogs({ eventID: event.id, eventType: event.type, kind, cause }),
),
),
)
function notify(event: Payload, isolateListeners: boolean) {
return Effect.gen(function* () {
yield* Effect.forEach(
listeners,
(listener) => (isolateListeners ? observe(event, "listener", listener) : listener(event)),
{ discard: true },
)
const pubsub = typed.get(event.type)
if (pubsub) yield* PubSub.publish(pubsub, event)
yield* PubSub.publish(all, event)
})
}
function publish<D extends Definition>(definition: D, data: Data<D>, options?: PublishOptions) {
return Effect.gen(function* () {
const serviceLocation = Option.getOrUndefined(yield* Effect.serviceOption(Location.Service))
@ -396,13 +459,7 @@ export const layerWith = (options?: LayerOptions) =>
strictOwner: options?.strictOwner,
})
if (committed && options?.publish) {
const published = { ...payload, seq: committed.seq }
for (const listener of listeners) {
yield* listener(published)
}
const pubsub = typed.get(payload.type)
if (pubsub) yield* PubSub.publish(pubsub, published)
yield* PubSub.publish(all, published)
yield* notify({ ...payload, seq: committed.seq }, true)
}
}
})

4
packages/core/src/event/sql.ts
View file

@ -1,4 +1,4 @@
import { sqliteTable, text, integer, index } from "drizzle-orm/sqlite-core"
import { sqliteTable, text, integer, index, uniqueIndex } from "drizzle-orm/sqlite-core"
import type { EventV2 } from "../event"
export const EventSequenceTable = sqliteTable("event_sequence", {
@ -19,7 +19,7 @@ export const EventTable = sqliteTable(
data: text({ mode: "json" }).$type<Record<string, unknown>>().notNull(),
},
(table) => [
index("event_aggregate_seq_idx").on(table.aggregate_id, table.seq),
uniqueIndex("event_aggregate_seq_idx").on(table.aggregate_id, table.seq),
index("event_aggregate_type_seq_idx").on(table.aggregate_id, table.type, table.seq),
],
)

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

@ -1,7 +1,7 @@
export * as SessionV2 from "./session"
export * from "./session/schema"
import { Cause, DateTime, Effect, Layer, Schema, Context, Stream } from "effect"
import { Cause, Effect, Layer, Schema, Context, Stream } from "effect"
import { and, asc, desc, eq, gt, like, lt, or, type SQL } from "drizzle-orm"
import { ProjectV2 } from "./project"
import { WorkspaceV2 } from "./workspace"
@ -10,7 +10,6 @@ import { Location } from "./location"
import { SessionMessage } from "./session/message"
import { Prompt } from "./session/prompt"
import { EventV2 } from "./event"
import { ProviderV2 } from "./provider"
import { Database } from "./database/database"
import { SessionProjector } from "./session/projector"
import { SessionMessageTable, SessionTable } from "./session/sql"
@ -140,7 +139,7 @@ export interface Interface {
prompt: Prompt
delivery?: SessionInput.Delivery
resume?: boolean
}) => Effect.Effect<SessionMessage.User, NotFoundError | PromptConflictError>
}) => Effect.Effect<SessionInput.Admitted, NotFoundError | PromptConflictError>
readonly shell: (input: {
id?: EventV2.ID
sessionID: SessionSchema.ID
@ -198,20 +197,6 @@ export const layer = Layer.effect(
),
)
const findExistingPrompt = Effect.fnUntraced(function* (input: {
sessionID: SessionSchema.ID
messageID: SessionMessage.ID
prompt: Prompt
delivery: SessionInput.Delivery
}) {
const stored = yield* SessionInput.find(db, input.messageID)
if (!stored) return yield* SessionInput.reconcileProjected(db, { id: input.messageID, ...input })
if (!SessionInput.equivalent(stored, input)) {
return yield* new PromptConflictError({ sessionID: input.sessionID, messageID: input.messageID })
}
return stored
})
const result = Service.of({
create: Effect.fn("V2Session.create")(function* (input) {
const sessionID = input.id ?? SessionSchema.ID.create()
@ -367,20 +352,23 @@ export const layer = Layer.effect(
yield* result.get(input.sessionID)
const returnPrompt = Effect.fnUntraced(function* (admitted: SessionInput.Admitted) {
if (input.resume !== false) yield* enqueueWake(input.sessionID)
return SessionInput.toMessage(admitted)
return admitted
}, Effect.uninterruptible)
const messageID = input.id ?? SessionMessage.ID.create()
const delivery = input.delivery ?? "steer"
const expected = { sessionID: input.sessionID, messageID, prompt: input.prompt, delivery }
const existing = yield* findExistingPrompt(expected)
if (existing) return yield* returnPrompt(existing)
const admitted = yield* SessionInput.admit(db, {
const admitted = yield* SessionInput.admit(db, events, {
id: messageID,
sessionID: input.sessionID,
prompt: input.prompt,
delivery,
})
if (!admitted) return yield* new PromptConflictError({ sessionID: input.sessionID, messageID })
}).pipe(
Effect.catchDefect((defect) =>
defect instanceof SessionInput.LifecycleConflict
? new PromptConflictError({ sessionID: input.sessionID, messageID })
: Effect.die(defect),
),
)
if (!SessionInput.equivalent(admitted, expected))
return yield* new PromptConflictError({ sessionID: input.sessionID, messageID })
return yield* returnPrompt(admitted)

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

@ -9,6 +9,7 @@ import { FileAttachment, Prompt } from "./prompt"
import { SessionSchema } from "./schema"
import { Location } from "../location"
import { RelativePath } from "../schema"
import { SessionMessageID } from "./message-id"
export { FileAttachment }
@ -52,6 +53,7 @@ export const AgentSwitched = EventV2.define({
...options,
schema: {
...Base,
messageID: SessionMessageID.ID,
agent: Schema.String,
},
})
@ -62,6 +64,7 @@ export const ModelSwitched = EventV2.define({
...options,
schema: {
...Base,
messageID: SessionMessageID.ID,
model: ModelV2.Ref,
},
})
@ -83,17 +86,45 @@ export const Prompted = EventV2.define({
...options,
schema: {
...Base,
messageID: SessionMessageID.ID,
prompt: Prompt,
delivery: Schema.Literals(["steer", "queue"]),
},
})
export type Prompted = typeof Prompted.Type
export namespace PromptLifecycle {
export const Admitted = EventV2.define({
type: "session.next.prompt.admitted",
...options,
schema: {
...Base,
messageID: SessionMessageID.ID,
prompt: Prompt,
delivery: Schema.Literals(["steer", "queue"]),
},
})
export type Admitted = typeof Admitted.Type
export const Promoted = EventV2.define({
type: "session.next.prompt.promoted",
...options,
schema: {
...Base,
messageID: SessionMessageID.ID,
prompt: Prompt,
timeCreated: V2Schema.DateTimeUtcFromMillis,
},
})
export type Promoted = typeof Promoted.Type
}
export const Synthetic = EventV2.define({
type: "session.next.synthetic",
...options,
schema: {
...Base,
messageID: SessionMessageID.ID,
text: Schema.String,
},
})
@ -105,6 +136,7 @@ export namespace Shell {
...options,
schema: {
...Base,
messageID: SessionMessageID.ID,
callID: Schema.String,
command: Schema.String,
},
@ -129,6 +161,7 @@ export namespace Step {
...options,
schema: {
...Base,
assistantMessageID: SessionMessageID.ID,
agent: Schema.String,
model: ModelV2.Ref,
snapshot: Schema.String.pipe(Schema.optional),
@ -141,7 +174,7 @@ export namespace Step {
...stepSettlementOptions,
schema: {
...Base,
assistantMessageID: EventV2.ID,
assistantMessageID: SessionMessageID.ID,
finish: Schema.String,
cost: Schema.Finite,
tokens: Schema.Struct({
@ -163,7 +196,7 @@ export namespace Step {
...stepSettlementOptions,
schema: {
...Base,
assistantMessageID: EventV2.ID,
assistantMessageID: SessionMessageID.ID,
error: UnknownError,
},
})
@ -176,6 +209,7 @@ export namespace Text {
...options,
schema: {
...Base,
assistantMessageID: SessionMessageID.ID,
textID: Schema.String,
},
})
@ -186,6 +220,7 @@ export namespace Text {
type: "session.next.text.delta",
schema: {
...Base,
assistantMessageID: SessionMessageID.ID,
textID: Schema.String,
delta: Schema.String,
},
@ -197,6 +232,7 @@ export namespace Text {
...options,
schema: {
...Base,
assistantMessageID: SessionMessageID.ID,
textID: Schema.String,
text: Schema.String,
},
@ -210,6 +246,7 @@ export namespace Reasoning {
...options,
schema: {
...Base,
assistantMessageID: SessionMessageID.ID,
reasoningID: Schema.String,
providerMetadata: ProviderMetadata.pipe(Schema.optional),
},
@ -221,6 +258,7 @@ export namespace Reasoning {
type: "session.next.reasoning.delta",
schema: {
...Base,
assistantMessageID: SessionMessageID.ID,
reasoningID: Schema.String,
delta: Schema.String,
},
@ -232,6 +270,7 @@ export namespace Reasoning {
...options,
schema: {
...Base,
assistantMessageID: SessionMessageID.ID,
reasoningID: Schema.String,
text: Schema.String,
providerMetadata: ProviderMetadata.pipe(Schema.optional),
@ -243,7 +282,7 @@ export namespace Reasoning {
export namespace Tool {
const ToolBase = {
...Base,
assistantMessageID: EventV2.ID,
assistantMessageID: SessionMessageID.ID,
callID: Schema.String,
}
@ -370,6 +409,7 @@ export namespace Compaction {
...options,
schema: {
...Base,
messageID: SessionMessageID.ID,
reason: Schema.Union([Schema.Literal("auto"), Schema.Literal("manual")]),
},
})
@ -402,6 +442,8 @@ const DurableDefinitions = [
ModelSwitched,
Moved,
Prompted,
PromptLifecycle.Admitted,
PromptLifecycle.Promoted,
Synthetic,
Shell.Started,
Shell.Ended,

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

@ -1,11 +1,11 @@
export * as SessionInput from "./input"
import { and, asc, eq, inArray, isNull } from "drizzle-orm"
import { and, asc, eq, isNull, lte } from "drizzle-orm"
import { DateTime, Effect, Schema } from "effect"
import type { Database } from "../database/database"
import type { EventV2 } from "../event"
import { EventTable } from "../event/sql"
import { NonNegativeInt, PositiveInt } from "../schema"
import { EventSequenceTable } from "../event/sql"
import { NonNegativeInt } from "../schema"
import { V2Schema } from "../v2-schema"
import { SessionEvent } from "./event"
import { SessionMessage } from "./message"
@ -19,7 +19,7 @@ export const Delivery = Schema.Literals(["steer", "queue"])
export type Delivery = typeof Delivery.Type
export class Admitted extends Schema.Class<Admitted>("SessionInput.Admitted")({
seq: PositiveInt,
admittedSeq: NonNegativeInt,
id: SessionMessage.ID,
sessionID: SessionSchema.ID,
prompt: Prompt,
@ -30,11 +30,10 @@ export class Admitted extends Schema.Class<Admitted>("SessionInput.Admitted")({
const decodePrompt = Schema.decodeUnknownSync(Prompt)
const encodePrompt = Schema.encodeSync(Prompt)
const decodeMessage = Schema.decodeUnknownSync(SessionMessage.Message)
const fromRow = (row: typeof SessionInputTable.$inferSelect): Admitted =>
new Admitted({
seq: row.seq,
admittedSeq: row.admitted_seq,
id: SessionMessage.ID.make(row.id),
sessionID: SessionSchema.ID.make(row.session_id),
prompt: decodePrompt(row.prompt),
@ -48,8 +47,13 @@ export const find = Effect.fn("SessionInput.find")(function* (db: DatabaseServic
return row === undefined ? undefined : fromRow(row)
})
export class LifecycleConflict extends Schema.TaggedErrorClass<LifecycleConflict>()("SessionInput.LifecycleConflict", {
id: SessionMessage.ID,
}) {}
export const admit = Effect.fn("SessionInput.admit")(function* (
db: DatabaseService,
events: EventV2.Interface,
input: {
readonly id: SessionMessage.ID
readonly sessionID: SessionSchema.ID
@ -57,49 +61,124 @@ export const admit = Effect.fn("SessionInput.admit")(function* (
readonly delivery: Delivery
},
) {
return yield* db
.transaction(
() =>
Effect.gen(function* () {
const existing = yield* find(db, input.id)
if (existing !== undefined) return existing
const event = yield* db
.select({ id: EventTable.id })
.from(EventTable)
.where(eq(EventTable.id, input.id))
.get()
.pipe(Effect.orDie)
const message = yield* db
.select({ id: SessionMessageTable.id })
.from(SessionMessageTable)
.where(eq(SessionMessageTable.id, input.id))
.get()
.pipe(Effect.orDie)
if (event !== undefined || message !== undefined) return undefined
const row = yield* db
.insert(SessionInputTable)
.values({
id: input.id,
session_id: input.sessionID,
prompt: encodePrompt(input.prompt),
delivery: input.delivery,
})
.returning()
.get()
.pipe(Effect.orDie)
return fromRow(row)
}),
{ behavior: "immediate" },
const existing = yield* find(db, input.id)
if (existing !== undefined) return existing
const timestamp = yield* DateTime.now
return yield* events
.publish(SessionEvent.PromptLifecycle.Admitted, {
messageID: input.id,
sessionID: input.sessionID,
timestamp,
prompt: input.prompt,
delivery: input.delivery,
})
.pipe(
Effect.flatMap((event) =>
event.seq === undefined
? Effect.die("Prompt admission event is missing aggregate sequence")
: Effect.succeed(
new Admitted({
admittedSeq: event.seq,
id: input.id,
sessionID: input.sessionID,
prompt: input.prompt,
delivery: input.delivery,
timeCreated: timestamp,
}),
),
),
Effect.catchDefect((defect) =>
find(db, input.id).pipe(Effect.flatMap((stored) => (stored ? Effect.succeed(stored) : Effect.die(defect)))),
),
)
})
export const latestSeq = Effect.fn("SessionInput.latestSeq")(function* (
db: DatabaseService,
sessionID: SessionSchema.ID,
) {
const row = yield* db
.select({ seq: EventSequenceTable.seq })
.from(EventSequenceTable)
.where(eq(EventSequenceTable.aggregate_id, sessionID))
.get()
.pipe(Effect.orDie)
return row?.seq ?? -1
})
export const projectAdmitted = Effect.fn("SessionInput.projectAdmitted")(function* (
db: DatabaseService,
input: {
readonly admittedSeq: number
readonly id: SessionMessage.ID
readonly sessionID: SessionSchema.ID
readonly prompt: Prompt
readonly delivery: Delivery
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({
id: input.id,
session_id: input.sessionID,
admitted_seq: input.admittedSeq,
prompt: encodePrompt(input.prompt),
delivery: input.delivery,
time_created: DateTime.toEpochMillis(input.timeCreated),
})
.onConflictDoNothing()
.returning({ id: SessionInputTable.id })
.get()
.pipe(Effect.orDie)
if (!stored) return yield* Effect.die(new LifecycleConflict({ id: input.id }))
})
export const projectPromoted = Effect.fn("SessionInput.projectPromoted")(function* (
db: DatabaseService,
input: {
readonly id: SessionMessage.ID
readonly sessionID: SessionSchema.ID
readonly prompt: Prompt
readonly timeCreated: DateTime.Utc
readonly promotedSeq: number
},
) {
const updated = yield* db
.update(SessionInputTable)
.set({ promoted_seq: input.promotedSeq })
.where(
and(
eq(SessionInputTable.id, input.id),
eq(SessionInputTable.session_id, input.sessionID),
isNull(SessionInputTable.promoted_seq),
),
)
.returning()
.get()
.pipe(Effect.orDie)
if (!updated) return yield* Effect.die(new LifecycleConflict({ id: input.id }))
const stored = fromRow(updated)
if (
!matchesPrompt(stored, input) ||
DateTime.toEpochMillis(stored.timeCreated) !== DateTime.toEpochMillis(input.timeCreated)
)
return yield* Effect.die(new LifecycleConflict({ id: input.id }))
return toMessage(stored)
})
export const hasPending = Effect.fn("SessionInput.hasPending")(function* (
db: DatabaseService,
sessionID: SessionSchema.ID,
deliveries: ReadonlyArray<Delivery> = ["steer", "queue"],
delivery: Delivery,
) {
if (deliveries.length === 0) return false
const row = yield* db
.select({ id: SessionInputTable.id })
.from(SessionInputTable)
@ -107,7 +186,7 @@ export const hasPending = Effect.fn("SessionInput.hasPending")(function* (
and(
eq(SessionInputTable.session_id, sessionID),
isNull(SessionInputTable.promoted_seq),
inArray(SessionInputTable.delivery, deliveries),
eq(SessionInputTable.delivery, delivery),
),
)
.limit(1)
@ -133,14 +212,34 @@ export const guardReservedID = Effect.fn("SessionInput.guardReservedID")(functio
db: DatabaseService,
event: EventV2.Payload,
) {
const admitted = yield* find(db, event.id)
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
if (!Schema.is(SessionEvent.Prompted)(event))
return yield* Effect.die("Durable event conflicts with admitted prompt input")
if (!equivalent(admitted, event.data)) return yield* Effect.die("Prompt projection conflicts with admitted input")
return yield* Effect.die(new LifecycleConflict({ id }))
})
export const project = Effect.fn("SessionInput.project")(function* (
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: {
readonly id: SessionMessage.ID
@ -151,82 +250,49 @@ export const project = Effect.fn("SessionInput.project")(function* (
readonly promotedSeq: number
},
) {
yield* db
const inserted = yield* db
.insert(SessionInputTable)
.values({
id: input.id,
session_id: input.sessionID,
admitted_seq: input.promotedSeq,
prompt: encodePrompt(input.prompt),
delivery: input.delivery,
promoted_seq: input.promotedSeq,
time_created: DateTime.toEpochMillis(input.timeCreated),
})
.onConflictDoNothing()
.run()
.pipe(Effect.orDie)
const admitted = yield* find(db, input.id)
if (admitted === undefined || admitted.delivery !== input.delivery || !matchesPrompt(admitted, input))
return yield* Effect.die("Prompt projection conflicts with admitted input")
yield* db
.update(SessionInputTable)
.set({ promoted_seq: input.promotedSeq })
.where(
and(
eq(SessionInputTable.id, input.id),
eq(SessionInputTable.session_id, input.sessionID),
isNull(SessionInputTable.promoted_seq),
),
)
.run()
.pipe(Effect.orDie)
return yield* find(db, input.id)
})
export const reconcileProjected = Effect.fn("SessionInput.reconcileProjected")(function* (
db: DatabaseService,
expected: {
readonly id: SessionMessage.ID
readonly sessionID: SessionSchema.ID
readonly prompt: Prompt
readonly delivery: Delivery
},
) {
if (expected.delivery !== "steer") return undefined
const row = yield* db
.select()
.from(SessionMessageTable)
.where(eq(SessionMessageTable.id, expected.id))
.returning()
.get()
.pipe(Effect.orDie)
if (row === undefined || row.session_id !== expected.sessionID || row.type !== "user") return undefined
const message = decodeMessage({ ...row.data, id: row.id, type: row.type })
if (message.type !== "user" || !Prompt.equivalence(Prompt.fromUserMessage(message), expected.prompt)) return undefined
return yield* project(db, {
id: expected.id,
sessionID: expected.sessionID,
prompt: expected.prompt,
delivery: expected.delivery,
timeCreated: message.time.created,
promotedSeq: row.seq,
})
if (!inserted) return yield* Effect.die("Prompt projection conflicts with admitted input")
return fromRow(inserted)
})
const publish = Effect.fn("SessionInput.publish")(function* (
db: DatabaseService,
events: EventV2.Interface,
sessionID: SessionSchema.ID,
rows: ReadonlyArray<typeof SessionInputTable.$inferSelect>,
) {
for (const row of rows) {
yield* events.publish(
SessionEvent.Prompted,
{
yield* events
.publish(SessionEvent.PromptLifecycle.Promoted, {
sessionID,
timestamp: DateTime.makeUnsafe(row.time_created),
timestamp: yield* DateTime.now,
messageID: SessionMessage.ID.make(row.id),
prompt: decodePrompt(row.prompt),
delivery: row.delivery,
},
{ id: SessionMessage.ID.make(row.id) },
)
timeCreated: DateTime.makeUnsafe(row.time_created),
})
.pipe(
Effect.catchDefect((defect) =>
defect instanceof LifecycleConflict
? find(db, SessionMessage.ID.make(row.id)).pipe(
Effect.flatMap((stored) => (stored?.promotedSeq === undefined ? Effect.die(defect) : Effect.void)),
)
: Effect.die(defect),
),
)
}
return rows.length
})
@ -235,6 +301,7 @@ export const promoteSteers = Effect.fn("SessionInput.promoteSteers")(function* (
db: DatabaseService,
events: EventV2.Interface,
sessionID: SessionSchema.ID,
cutoff: number,
) {
const rows = yield* db
.select()
@ -244,12 +311,13 @@ export const promoteSteers = Effect.fn("SessionInput.promoteSteers")(function* (
eq(SessionInputTable.session_id, sessionID),
isNull(SessionInputTable.promoted_seq),
eq(SessionInputTable.delivery, "steer"),
lte(SessionInputTable.admitted_seq, cutoff),
),
)
.orderBy(asc(SessionInputTable.seq))
.orderBy(asc(SessionInputTable.admitted_seq))
.all()
.pipe(Effect.orDie)
return yield* publish(events, sessionID, rows)
return yield* publish(db, events, sessionID, rows)
})
export const promoteNextQueued = Effect.fn("SessionInput.promoteNextQueued")(function* (
@ -267,14 +335,14 @@ export const promoteNextQueued = Effect.fn("SessionInput.promoteNextQueued")(fun
eq(SessionInputTable.delivery, "queue"),
),
)
.orderBy(asc(SessionInputTable.seq))
.orderBy(asc(SessionInputTable.admitted_seq))
.limit(1)
.get()
.pipe(Effect.orDie)
return row === undefined ? false : yield* publish(events, sessionID, [row]).pipe(Effect.as(true))
return row === undefined ? false : yield* publish(db, events, sessionID, [row]).pipe(Effect.as(true))
})
export const toMessage = (input: Admitted) =>
const toMessage = (input: Admitted) =>
new SessionMessage.User({
id: input.id,
type: "user",

13
packages/core/src/session/message-id.ts Normal file
View file

@ -0,0 +1,13 @@
export * as SessionMessageID from "./message-id"
import { Schema } from "effect"
import { withStatics } from "../schema"
import { Identifier } from "../util/identifier"
export const ID = Schema.String.check(Schema.isStartsWith("msg_")).pipe(
Schema.brand("Session.Message.ID"),
withStatics((schema) => ({
create: () => schema.make("msg_" + Identifier.ascending()),
})),
)
export type ID = typeof ID.Type

114
packages/core/src/session/message-updater.ts
View file

@ -123,7 +123,7 @@ export function update(adapter: Adapter, event: SessionEvent.Event) {
"session.next.agent.switched": (event) => {
return adapter.appendMessage(
new SessionMessage.AgentSwitched({
id: event.id,
id: event.data.messageID,
type: "agent-switched",
metadata: event.metadata,
agent: event.data.agent,
@ -134,7 +134,7 @@ export function update(adapter: Adapter, event: SessionEvent.Event) {
"session.next.model.switched": (event) => {
return adapter.appendMessage(
new SessionMessage.ModelSwitched({
id: event.id,
id: event.data.messageID,
type: "model-switched",
metadata: event.metadata,
model: event.data.model,
@ -146,7 +146,7 @@ export function update(adapter: Adapter, event: SessionEvent.Event) {
"session.next.prompted": (event) => {
return adapter.appendMessage(
new SessionMessage.User({
id: event.id,
id: event.data.messageID,
type: "user",
metadata: event.metadata,
text: event.data.prompt.text,
@ -157,12 +157,14 @@ export function update(adapter: Adapter, event: SessionEvent.Event) {
}),
)
},
"session.next.prompt.admitted": () => Effect.void,
"session.next.prompt.promoted": () => Effect.void,
"session.next.synthetic": (event) => {
return adapter.appendMessage(
new SessionMessage.Synthetic({
sessionID: event.data.sessionID,
text: event.data.text,
id: event.id,
id: event.data.messageID,
type: "synthetic",
time: { created: event.data.timestamp },
}),
@ -171,7 +173,7 @@ export function update(adapter: Adapter, event: SessionEvent.Event) {
"session.next.shell.started": (event) => {
return adapter.appendMessage(
new SessionMessage.Shell({
id: event.id,
id: event.data.messageID,
type: "shell",
metadata: event.metadata,
callID: event.data.callID,
@ -206,7 +208,7 @@ export function update(adapter: Adapter, event: SessionEvent.Event) {
}
yield* adapter.appendMessage(
new SessionMessage.Assistant({
id: event.id,
id: event.data.assistantMessageID,
type: "assistant",
agent: event.data.agent,
model: event.data.model,
@ -234,43 +236,22 @@ export function update(adapter: Adapter, event: SessionEvent.Event) {
})
},
"session.next.text.started": (event) => {
return Effect.gen(function* () {
const currentAssistant = yield* adapter.getCurrentAssistant()
if (currentAssistant) {
yield* adapter.updateAssistant(
produce(currentAssistant, (draft) => {
draft.content.push(
castDraft(new SessionMessage.AssistantText({ type: "text", id: event.data.textID, text: "" })),
)
}),
)
}
return updateOwnedAssistant(event.data.assistantMessageID, (draft) => {
draft.content.push(
castDraft(new SessionMessage.AssistantText({ type: "text", id: event.data.textID, text: "" })),
)
})
},
"session.next.text.delta": (event) => {
return Effect.gen(function* () {
const currentAssistant = yield* adapter.getCurrentAssistant()
if (currentAssistant) {
yield* adapter.updateAssistant(
produce(currentAssistant, (draft) => {
const match = latestText(draft, event.data.textID)
if (match) match.text += event.data.delta
}),
)
}
return updateOwnedAssistant(event.data.assistantMessageID, (draft) => {
const match = latestText(draft, event.data.textID)
if (match) match.text += event.data.delta
})
},
"session.next.text.ended": (event) => {
return Effect.gen(function* () {
const currentAssistant = yield* adapter.getCurrentAssistant()
if (currentAssistant) {
yield* adapter.updateAssistant(
produce(currentAssistant, (draft) => {
const match = latestText(draft, event.data.textID)
if (match) match.text = event.data.text
}),
)
}
return updateOwnedAssistant(event.data.assistantMessageID, (draft) => {
const match = latestText(draft, event.data.textID)
if (match) match.text = event.data.text
})
},
"session.next.tool.input.started": (event) => {
@ -367,52 +348,31 @@ export function update(adapter: Adapter, event: SessionEvent.Event) {
})
},
"session.next.reasoning.started": (event) => {
return Effect.gen(function* () {
const currentAssistant = yield* adapter.getCurrentAssistant()
if (currentAssistant) {
yield* adapter.updateAssistant(
produce(currentAssistant, (draft) => {
draft.content.push(
castDraft(
new SessionMessage.AssistantReasoning({
type: "reasoning",
id: event.data.reasoningID,
text: "",
providerMetadata: event.data.providerMetadata,
}),
),
)
return updateOwnedAssistant(event.data.assistantMessageID, (draft) => {
draft.content.push(
castDraft(
new SessionMessage.AssistantReasoning({
type: "reasoning",
id: event.data.reasoningID,
text: "",
providerMetadata: event.data.providerMetadata,
}),
)
}
),
)
})
},
"session.next.reasoning.delta": (event) => {
return Effect.gen(function* () {
const currentAssistant = yield* adapter.getCurrentAssistant()
if (currentAssistant) {
yield* adapter.updateAssistant(
produce(currentAssistant, (draft) => {
const match = latestReasoning(draft, event.data.reasoningID)
if (match) match.text += event.data.delta
}),
)
}
return updateOwnedAssistant(event.data.assistantMessageID, (draft) => {
const match = latestReasoning(draft, event.data.reasoningID)
if (match) match.text += event.data.delta
})
},
"session.next.reasoning.ended": (event) => {
return Effect.gen(function* () {
const currentAssistant = yield* adapter.getCurrentAssistant()
if (currentAssistant) {
yield* adapter.updateAssistant(
produce(currentAssistant, (draft) => {
const match = latestReasoning(draft, event.data.reasoningID)
if (match) {
match.text = event.data.text
if (event.data.providerMetadata !== undefined) match.providerMetadata = event.data.providerMetadata
}
}),
)
return updateOwnedAssistant(event.data.assistantMessageID, (draft) => {
const match = latestReasoning(draft, event.data.reasoningID)
if (match) {
match.text = event.data.text
if (event.data.providerMetadata !== undefined) match.providerMetadata = event.data.providerMetadata
}
})
},
@ -420,7 +380,7 @@ export function update(adapter: Adapter, event: SessionEvent.Event) {
"session.next.compaction.started": (event) => {
return adapter.appendMessage(
new SessionMessage.Compaction({
id: event.id,
id: event.data.messageID,
type: "compaction",
metadata: event.metadata,
reason: event.data.reason,

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

@ -2,15 +2,15 @@ export * as SessionMessage from "./message"
import { Schema } from "effect"
import { ProviderMetadata } from "@opencode-ai/llm"
import { EventV2 } from "../event"
import { ModelV2 } from "../model"
import { ToolOutput } from "../tool-output"
import { V2Schema } from "../v2-schema"
import { SessionEvent } from "./event"
import { Prompt } from "./prompt"
import { SessionMessageID } from "./message-id"
export const ID = EventV2.ID
export type ID = Schema.Schema.Type<typeof ID>
export const ID = SessionMessageID.ID
export type ID = typeof ID.Type
const Base = {
id: ID,

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

@ -19,7 +19,7 @@ type DatabaseService = Database.Interface["db"]
const decodeMessage = Schema.decodeUnknownSync(SessionMessage.Message)
const encodeMessage = Schema.encodeSync(SessionMessage.Message)
export class PromptAlreadyProjected extends Error {}
class PromptAlreadyProjected extends Error {}
export class SessionAlreadyProjected extends Error {}
type Usage = {
@ -112,29 +112,23 @@ function run(db: DatabaseService, event: SessionEvent.Event) {
return Effect.gen(function* () {
const decodeRow = (row: typeof SessionMessageTable.$inferSelect) =>
decodeMessage({ ...row.data, id: row.id, type: row.type })
const writeMessage = (message: SessionMessage.Message) => {
const updateMessage = (message: SessionMessage.Message) => {
if (event.seq === undefined) return Effect.die("Synchronized Session event is missing aggregate sequence")
const encoded = encodeMessage(message)
const { id, type, ...data } = encoded
return db
.insert(SessionMessageTable)
.values([
{
id: SessionMessage.ID.make(id),
session_id: event.data.sessionID,
type,
seq: event.seq,
time_created: DateTime.toEpochMillis(message.time.created),
data,
},
])
.onConflictDoUpdate({
target: SessionMessageTable.id,
set: { type, time_created: DateTime.toEpochMillis(message.time.created), data },
})
.update(SessionMessageTable)
.set({ type, time_created: DateTime.toEpochMillis(message.time.created), data })
.where(
and(
eq(SessionMessageTable.id, SessionMessage.ID.make(id)),
eq(SessionMessageTable.session_id, event.data.sessionID),
),
)
.run()
.pipe(Effect.orDie)
}
const appendMessage = (message: SessionMessage.Message) => insertMessage(db, event, message)
const adapter: SessionMessageUpdater.Adapter = {
getCurrentAssistant() {
return Effect.gen(function* () {
@ -204,15 +198,33 @@ function run(db: DatabaseService, event: SessionEvent.Event) {
.find((message): message is SessionMessage.Shell => message.type === "shell" && message.callID === callID)
})
},
updateAssistant: writeMessage,
updateCompaction: writeMessage,
updateShell: writeMessage,
appendMessage: writeMessage,
updateAssistant: updateMessage,
updateCompaction: updateMessage,
updateShell: updateMessage,
appendMessage,
}
yield* SessionMessageUpdater.update(adapter, 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")
const encoded = encodeMessage(message)
const { id, type, ...data } = encoded
return db
.insert(SessionMessageTable)
.values({
id: SessionMessage.ID.make(id),
session_id: event.data.sessionID,
type,
seq: event.seq,
time_created: DateTime.toEpochMillis(message.time.created),
data,
})
.run()
.pipe(Effect.orDie)
}
export const layer = Layer.effectDiscard(
Effect.gen(function* () {
const events = yield* EventV2.Service
@ -349,27 +361,19 @@ export const layer = Layer.effectDiscard(
)
yield* events.project(SessionEvent.Prompted, (event) =>
Effect.gen(function* () {
const messageID = event.data.messageID
const existing = yield* db
.select({ id: SessionMessageTable.id })
.from(SessionMessageTable)
.where(eq(SessionMessageTable.id, event.id))
.where(eq(SessionMessageTable.id, messageID))
.get()
.pipe(Effect.orDie)
if (existing) return yield* Effect.die(new PromptAlreadyProjected())
yield* run(db, event)
const row = yield* db
.select()
.from(SessionMessageTable)
.where(eq(SessionMessageTable.id, event.id))
.get()
.pipe(Effect.orDie)
if (!row) return yield* Effect.die("Prompt projection was not stored")
const message = decodeMessage({ ...row.data, id: row.id, type: row.type })
if (message.type !== "user") return yield* Effect.die("Prompt projection did not produce a user message")
if (event.seq === undefined)
return yield* Effect.die("Synchronized Session event is missing aggregate sequence")
yield* SessionInput.project(db, {
id: SessionMessage.ID.make(event.id),
yield* SessionInput.projectLegacyPrompted(db, {
id: messageID,
sessionID: event.data.sessionID,
prompt: event.data.prompt,
delivery: event.data.delivery,
@ -378,6 +382,37 @@ export const layer = Layer.effectDiscard(
})
}),
)
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")
yield* SessionInput.projectAdmitted(db, {
admittedSeq: event.seq,
id: event.data.messageID,
sessionID: event.data.sessionID,
prompt: event.data.prompt,
delivery: event.data.delivery,
timeCreated: event.data.timestamp,
})
}),
)
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")
yield* insertMessage(
db,
event,
yield* SessionInput.projectPromoted(db, {
id: event.data.messageID,
sessionID: event.data.sessionID,
prompt: event.data.prompt,
timeCreated: event.data.timeCreated,
promotedSeq: event.seq,
}),
)
}),
)
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))

19
packages/core/src/session/runner/llm.ts
View file

@ -132,10 +132,13 @@ export const layer = Layer.effect(
const model = yield* models.resolve(session)
const toolFibers = yield* FiberSet.make<void, never>()
let needsContinuation = false
if (promotion === "steer") yield* SessionInput.promoteSteers(db, events, session.id)
if (promotion === "queue") {
yield* SessionInput.promoteNextQueued(db, events, session.id)
yield* SessionInput.promoteSteers(db, events, session.id)
if (promotion) {
const cutoff = yield* SessionInput.latestSeq(db, session.id)
if (promotion === "steer") yield* SessionInput.promoteSteers(db, events, session.id, cutoff)
if (promotion === "queue") {
yield* SessionInput.promoteNextQueued(db, events, session.id)
yield* SessionInput.promoteSteers(db, events, session.id, cutoff)
}
}
yield* failInterruptedTools(session.id)
const context = yield* getContext(session.id)
@ -233,8 +236,8 @@ export const layer = Layer.effect(
readonly force?: boolean
}) {
const session = yield* getSession(input.sessionID)
const hasSteer = yield* SessionInput.hasPending(db, input.sessionID, ["steer"])
const hasQueue = yield* SessionInput.hasPending(db, input.sessionID, ["queue"])
const hasSteer = yield* SessionInput.hasPending(db, input.sessionID, "steer")
const hasQueue = hasSteer ? false : yield* SessionInput.hasPending(db, input.sessionID, "queue")
if (input.force !== true && !hasSteer && !hasQueue) return
let promotion: "steer" | "queue" | undefined = hasSteer ? "steer" : hasQueue ? "queue" : undefined
let openActivity = input.force === true || hasSteer || hasQueue
@ -243,12 +246,12 @@ export const layer = Layer.effect(
for (let step = 0; step < MAX_STEPS; step++) {
needsContinuation = yield* runTurn(session, promotion)
promotion = "steer"
if (!needsContinuation) needsContinuation = yield* SessionInput.hasPending(db, input.sessionID, ["steer"])
if (!needsContinuation) needsContinuation = yield* SessionInput.hasPending(db, input.sessionID, "steer")
if (!needsContinuation) break
}
if (needsContinuation)
return yield* new StepLimitExceededError({ sessionID: input.sessionID, limit: MAX_STEPS })
openActivity = yield* SessionInput.hasPending(db, input.sessionID, ["queue"])
openActivity = yield* SessionInput.hasPending(db, input.sessionID, "queue")
promotion = openActivity ? "queue" : undefined
}
})

19
packages/core/src/session/runner/publish-llm-event.ts
View file

@ -10,6 +10,7 @@ import { DateTime, Effect } from "effect"
import { EventV2 } from "../../event"
import { ModelV2 } from "../../model"
import { SessionEvent } from "../event"
import { SessionMessage } from "../message"
import { SessionSchema } from "../schema"
type Input = {
@ -60,7 +61,7 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input)
const tools = new Map<
string,
{
readonly assistantMessageID: EventV2.ID
readonly assistantMessageID: SessionMessage.ID
readonly name: string
inputEnded: boolean
called: boolean
@ -70,13 +71,17 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input)
}
>()
const timestamp = DateTime.now
let assistantMessageID: EventV2.ID | undefined
let assistantMessageID: SessionMessage.ID | undefined
let providerFailed = false
const startAssistant = Effect.fnUntraced(function* () {
if (assistantMessageID !== undefined) return assistantMessageID
assistantMessageID = (yield* events.publish(SessionEvent.Step.Started, { ...input, timestamp: yield* timestamp }))
.id
assistantMessageID = SessionMessage.ID.create()
yield* events.publish(SessionEvent.Step.Started, {
...input,
assistantMessageID,
timestamp: yield* timestamp,
})
return assistantMessageID
})
const currentAssistantMessageID = () =>
@ -118,6 +123,7 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input)
Effect.gen(function* () {
yield* events.publish(SessionEvent.Text.Ended, {
sessionID: input.sessionID,
assistantMessageID: yield* currentAssistantMessageID(),
timestamp: yield* timestamp,
textID,
text: value,
@ -128,6 +134,7 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input)
Effect.gen(function* () {
yield* events.publish(SessionEvent.Reasoning.Ended, {
sessionID: input.sessionID,
assistantMessageID: yield* currentAssistantMessageID(),
timestamp: yield* timestamp,
reasoningID,
text: value,
@ -220,6 +227,7 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input)
yield* text.start(event.id)
yield* events.publish(SessionEvent.Text.Started, {
sessionID: input.sessionID,
assistantMessageID: yield* startAssistant(),
timestamp: yield* timestamp,
textID: event.id,
})
@ -228,6 +236,7 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input)
yield* text.append(event.id, event.text)
yield* events.publish(SessionEvent.Text.Delta, {
sessionID: input.sessionID,
assistantMessageID: yield* currentAssistantMessageID(),
timestamp: yield* timestamp,
textID: event.id,
delta: event.text,
@ -240,6 +249,7 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input)
yield* reasoning.start(event.id)
yield* events.publish(SessionEvent.Reasoning.Started, {
sessionID: input.sessionID,
assistantMessageID: yield* startAssistant(),
timestamp: yield* timestamp,
reasoningID: event.id,
providerMetadata: event.providerMetadata,
@ -249,6 +259,7 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input)
yield* reasoning.append(event.id, event.text)
yield* events.publish(SessionEvent.Reasoning.Delta, {
sessionID: input.sessionID,
assistantMessageID: yield* currentAssistantMessageID(),
timestamp: yield* timestamp,
reasoningID: event.id,
delta: event.text,

12
packages/core/src/session/sql.ts
View file

@ -1,4 +1,4 @@
import { sqliteTable, text, integer, index, primaryKey, real } from "drizzle-orm/sqlite-core"
import { sqliteTable, text, integer, index, primaryKey, real, uniqueIndex } from "drizzle-orm/sqlite-core"
import * as DatabasePath from "../database/path"
import { ProjectTable } from "../project/sql"
import type { SessionMessage } from "./message"
@ -127,7 +127,7 @@ export const SessionMessageTable = sqliteTable(
data: text({ mode: "json" }).notNull().$type<SessionMessageData>(),
},
(table) => [
index("session_message_session_seq_idx").on(table.session_id, table.seq),
uniqueIndex("session_message_session_seq_idx").on(table.session_id, table.seq),
index("session_message_session_type_seq_idx").on(table.session_id, table.type, table.seq),
index("session_message_session_time_created_id_idx").on(table.session_id, table.time_created, table.id),
index("session_message_time_created_idx").on(table.time_created),
@ -137,14 +137,14 @@ export const SessionMessageTable = sqliteTable(
export const SessionInputTable = sqliteTable(
"session_input",
{
seq: integer().primaryKey({ autoIncrement: true }),
id: text().$type<SessionMessage.ID>().notNull().unique(),
id: text().$type<SessionMessage.ID>().primaryKey(),
session_id: text()
.$type<SessionSchema.ID>()
.notNull()
.references(() => SessionTable.id, { onDelete: "cascade" }),
prompt: text({ mode: "json" }).notNull().$type<Prompt>(),
delivery: text().$type<SessionInput.Delivery>().notNull(),
admitted_seq: integer().notNull(),
promoted_seq: integer(),
time_created: integer()
.notNull()
@ -155,7 +155,9 @@ export const SessionInputTable = sqliteTable(
table.session_id,
table.promoted_seq,
table.delivery,
table.seq,
table.admitted_seq,
),
uniqueIndex("session_input_session_admitted_seq_idx").on(table.session_id, table.admitted_seq),
uniqueIndex("session_input_session_promoted_seq_idx").on(table.session_id, table.promoted_seq),
],
)