refactor(core): simplify session fork event

This commit is contained in:
Dax Raad 2026-06-29 16:16:13 -04:00
commit ff4cab03c1
5 changed files with 40 additions and 52 deletions

View file

@ -799,12 +799,7 @@ export type SessionsHistoryOutput = {
readonly timestamp: number readonly timestamp: number
readonly sessionID: string readonly sessionID: string
readonly parentID: string readonly parentID: string
readonly slug: string
readonly title: string
readonly agent?: string
readonly model?: { readonly id: string; readonly providerID: string; readonly variant?: string }
readonly messageID?: string readonly messageID?: string
readonly copiedSeq: number
} }
} }
| { | {
@ -1283,12 +1278,7 @@ export type SessionsEventsOutput =
readonly timestamp: number readonly timestamp: number
readonly sessionID: string readonly sessionID: string
readonly parentID: string readonly parentID: string
readonly slug: string
readonly title: string
readonly agent?: string
readonly model?: { readonly id: string; readonly providerID: string; readonly variant?: string }
readonly messageID?: string readonly messageID?: string
readonly copiedSeq: number
} }
} }
| { | {

View file

@ -298,35 +298,12 @@ export const layer = Layer.effect(
: undefined : undefined
if (input.messageID && !boundary) if (input.messageID && !boundary)
return yield* new MessageNotFoundError({ sessionID: input.sessionID, messageID: input.messageID }) return yield* new MessageNotFoundError({ sessionID: input.sessionID, messageID: input.messageID })
const copied = yield* db
.select({ seq: SessionMessageTable.seq })
.from(SessionMessageTable)
.where(
and(
eq(SessionMessageTable.session_id, input.sessionID),
boundary === undefined ? undefined : lt(SessionMessageTable.seq, boundary.seq),
),
)
.orderBy(desc(SessionMessageTable.seq))
.limit(1)
.get()
.pipe(Effect.orDie)
const sessionID = SessionSchema.ID.create() const sessionID = SessionSchema.ID.create()
yield* events.publish(SessionEvent.Forked, { yield* events.publish(SessionEvent.Forked, {
sessionID, sessionID,
parentID: parent.id, parentID: parent.id,
slug: Slug.create(),
title: forkTitle(parent.title),
agent: parent.agent,
model: parent.model,
messageID: input.messageID, messageID: input.messageID,
copiedSeq: copied?.seq ?? 0,
timestamp: yield* DateTime.now, timestamp: yield* DateTime.now,
}, {
commit: (seq) =>
copied && copied.seq > seq
? EventV2.reserveSequence(db, sessionID, copied.seq)
: Effect.void,
}) })
return yield* result.get(sessionID).pipe(Effect.orDie) return yield* result.get(sessionID).pipe(Effect.orDie)
}), }),
@ -563,12 +540,6 @@ export const defaultLayer = layer.pipe(
Layer.orDie, Layer.orDie,
) )
const forkTitle = (value: string) => {
const match = value.match(/^(.+) \(fork #(\d+)\)$/)
if (match) return `${match[1]} (fork #${Number.parseInt(match[2], 10) + 1})`
return `${value} (fork #1)`
}
const resolvePrompt = (input: PromptInput.Prompt) => const resolvePrompt = (input: PromptInput.Prompt) =>
Prompt.make({ Prompt.make({
text: input.text, text: input.text,

View file

@ -15,6 +15,7 @@ import { WorkspaceV2 } from "../workspace"
import { SessionContextEpoch } from "./context-epoch" import { SessionContextEpoch } from "./context-epoch"
import { MessageTable, PartTable, SessionInputTable, SessionMessageTable, SessionTable } from "./sql" import { MessageTable, PartTable, SessionInputTable, SessionMessageTable, SessionTable } from "./sql"
import type { DeepMutable } from "../schema" import type { DeepMutable } from "../schema"
import { Slug } from "../util/slug"
type DatabaseService = Database.Interface["db"] type DatabaseService = Database.Interface["db"]
type MessageEvent = Exclude<SessionEvent.Event, typeof SessionEvent.Forked.Type> type MessageEvent = Exclude<SessionEvent.Event, typeof SessionEvent.Forked.Type>
@ -41,6 +42,12 @@ const emptyUsage = (): Usage => ({
tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } }, tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } },
}) })
const forkTitle = (value: string) => {
const match = value.match(/^(.+) \(fork #(\d+)\)$/)
if (match) return `${match[1]} (fork #${Number.parseInt(match[2], 10) + 1})`
return `${value} (fork #1)`
}
function usage(part: (typeof SessionV1.Event.PartUpdated.Type)["data"]["part"] | unknown): Usage | undefined { function usage(part: (typeof SessionV1.Event.PartUpdated.Type)["data"]["part"] | unknown): Usage | undefined {
if (typeof part !== "object" || part === null) return undefined if (typeof part !== "object" || part === null) return undefined
const value = part as Record<string, unknown> const value = part as Record<string, unknown>
@ -144,6 +151,31 @@ const projectFork = Effect.fn("SessionProjector.projectFork")(function* (
.get() .get()
.pipe(Effect.orDie) .pipe(Effect.orDie)
if (!parent) return yield* Effect.die(`Fork parent session not found: ${event.data.parentID}`) if (!parent) return yield* Effect.die(`Fork parent session not found: ${event.data.parentID}`)
const boundary = event.data.messageID
? yield* db
.select({ seq: SessionMessageTable.seq })
.from(SessionMessageTable)
.where(
and(eq(SessionMessageTable.session_id, event.data.parentID), eq(SessionMessageTable.id, event.data.messageID)),
)
.get()
.pipe(Effect.orDie)
: undefined
if (event.data.messageID && !boundary) return yield* Effect.die(`Fork boundary message not found: ${event.data.messageID}`)
const copied = yield* db
.select({ seq: SessionMessageTable.seq })
.from(SessionMessageTable)
.where(
and(
eq(SessionMessageTable.session_id, event.data.parentID),
boundary === undefined ? undefined : lt(SessionMessageTable.seq, boundary.seq),
),
)
.orderBy(desc(SessionMessageTable.seq))
.limit(1)
.get()
.pipe(Effect.orDie)
const copiedSeq = copied?.seq ?? 0
const stored = yield* db const stored = yield* db
.insert(SessionTable) .insert(SessionTable)
@ -152,12 +184,12 @@ const projectFork = Effect.fn("SessionProjector.projectFork")(function* (
parent_id: event.data.parentID, parent_id: event.data.parentID,
project_id: parent.project_id, project_id: parent.project_id,
workspace_id: parent.workspace_id, workspace_id: parent.workspace_id,
slug: event.data.slug, slug: Slug.create(),
directory: parent.directory, directory: parent.directory,
path: parent.path, path: parent.path,
title: event.data.title, title: forkTitle(parent.title),
agent: event.data.agent, agent: parent.agent,
model: event.data.model, model: parent.model,
version: parent.version, version: parent.version,
cost: 0, cost: 0,
tokens_input: 0, tokens_input: 0,
@ -184,7 +216,7 @@ const projectFork = Effect.fn("SessionProjector.projectFork")(function* (
and( and(
eq(SessionMessageTable.session_id, event.data.parentID), eq(SessionMessageTable.session_id, event.data.parentID),
gt(SessionMessageTable.seq, cursor), gt(SessionMessageTable.seq, cursor),
event.data.messageID === undefined ? undefined : lt(SessionMessageTable.seq, event.data.copiedSeq + 1), copiedSeq === 0 ? undefined : lt(SessionMessageTable.seq, copiedSeq + 1),
), ),
) )
.orderBy(asc(SessionMessageTable.seq)) .orderBy(asc(SessionMessageTable.seq))
@ -273,7 +305,7 @@ const projectFork = Effect.fn("SessionProjector.projectFork")(function* (
.where(eq(SessionTable.id, event.data.sessionID)) .where(eq(SessionTable.id, event.data.sessionID))
.run() .run()
.pipe(Effect.orDie) .pipe(Effect.orDie)
if (event.data.copiedSeq > 0) yield* EventV2.reserveSequence(db, event.data.sessionID, event.data.copiedSeq) if (copiedSeq > 0) yield* EventV2.reserveSequence(db, event.data.sessionID, copiedSeq)
}) })
function run(db: DatabaseService, event: MessageEvent) { function run(db: DatabaseService, event: MessageEvent) {

View file

@ -166,7 +166,7 @@ describe("SessionV2.create", () => {
expect(history.events[0]).toMatchObject({ expect(history.events[0]).toMatchObject({
type: "session.next.forked", type: "session.next.forked",
durable: { seq: 0 }, durable: { seq: 0 },
data: { sessionID: forked.id, parentID: parent.id, copiedSeq: 3 }, data: { sessionID: forked.id, parentID: parent.id },
}) })
expect(yield* SessionInput.find(db, forkContext[0]!.id)).toMatchObject({ expect(yield* SessionInput.find(db, forkContext[0]!.id)).toMatchObject({
sessionID: forked.id, sessionID: forked.id,
@ -216,7 +216,7 @@ describe("SessionV2.create", () => {
const history = yield* session.history({ sessionID: forked.id, limit: 10 }) const history = yield* session.history({ sessionID: forked.id, limit: 10 })
expect(context).toMatchObject([{ text: "First" }]) expect(context).toMatchObject([{ text: "First" }])
expect(context[0]?.id).not.toBe(first.id) expect(context[0]?.id).not.toBe(first.id)
expect(history.events[0]).toMatchObject({ data: { copiedSeq: 2, messageID: second.id } }) expect(history.events[0]).toMatchObject({ data: { messageID: second.id } })
}), }),
) )

View file

@ -100,12 +100,7 @@ export const Forked = Event.define({
schema: { schema: {
...Base, ...Base,
parentID: SessionID, parentID: SessionID,
slug: Schema.String,
title: Schema.String,
agent: Schema.String.pipe(optional),
model: Model.Ref.pipe(optional),
messageID: SessionMessage.ID.pipe(optional), messageID: SessionMessage.ID.pipe(optional),
copiedSeq: NonNegativeInt,
}, },
}) })
export type Forked = typeof Forked.Type export type Forked = typeof Forked.Type