From b2d46ecd7ede61f4b593a300f9953cadb9e109b3 Mon Sep 17 00:00:00 2001 From: Dax Raad Date: Mon, 29 Jun 2026 16:13:51 -0400 Subject: [PATCH] feat(core): add durable session fork event --- .../client/src/generated-effect/client.ts | 158 ++++++++-------- packages/client/src/generated/client.ts | 14 ++ packages/client/src/generated/types.ts | 93 +++++++++- packages/core/src/event.ts | 20 +- packages/core/src/session.ts | 66 ++++++- packages/core/src/session/message-updater.ts | 1 + packages/core/src/session/projector.ts | 172 +++++++++++++++++- packages/core/test/session-create.test.ts | 91 ++++++++- packages/protocol/src/groups/session.ts | 17 ++ packages/schema/src/session-event.ts | 18 ++ packages/server/src/handlers/session.ts | 26 +++ 11 files changed, 588 insertions(+), 88 deletions(-) diff --git a/packages/client/src/generated-effect/client.ts b/packages/client/src/generated-effect/client.ts index ed041ed50f..30047f4a04 100644 --- a/packages/client/src/generated-effect/client.ts +++ b/packages/client/src/generated-effect/client.ts @@ -86,45 +86,56 @@ const Endpoint3_3 = (raw: RawClient["server.session"]) => (input: Endpoint3_3Inp Effect.map((value) => value.data), ) -type Endpoint3_4Request = Parameters[0] +type Endpoint3_4Request = Parameters[0] type Endpoint3_4Input = { readonly sessionID: Endpoint3_4Request["params"]["sessionID"] - readonly agent: Endpoint3_4Request["payload"]["agent"] + readonly messageID?: Endpoint3_4Request["payload"]["messageID"] } const Endpoint3_4 = (raw: RawClient["server.session"]) => (input: Endpoint3_4Input) => + raw["session.fork"]({ params: { sessionID: input["sessionID"] }, payload: { messageID: input["messageID"] } }).pipe( + Effect.mapError(mapClientError), + Effect.map((value) => value.data), + ) + +type Endpoint3_5Request = Parameters[0] +type Endpoint3_5Input = { + readonly sessionID: Endpoint3_5Request["params"]["sessionID"] + readonly agent: Endpoint3_5Request["payload"]["agent"] +} +const Endpoint3_5 = (raw: RawClient["server.session"]) => (input: Endpoint3_5Input) => raw["session.switchAgent"]({ params: { sessionID: input["sessionID"] }, payload: { agent: input["agent"] } }).pipe( Effect.mapError(mapClientError), ) -type Endpoint3_5Request = Parameters[0] -type Endpoint3_5Input = { - readonly sessionID: Endpoint3_5Request["params"]["sessionID"] - readonly model: Endpoint3_5Request["payload"]["model"] +type Endpoint3_6Request = Parameters[0] +type Endpoint3_6Input = { + readonly sessionID: Endpoint3_6Request["params"]["sessionID"] + readonly model: Endpoint3_6Request["payload"]["model"] } -const Endpoint3_5 = (raw: RawClient["server.session"]) => (input: Endpoint3_5Input) => +const Endpoint3_6 = (raw: RawClient["server.session"]) => (input: Endpoint3_6Input) => raw["session.switchModel"]({ params: { sessionID: input["sessionID"] }, payload: { model: input["model"] } }).pipe( Effect.mapError(mapClientError), ) -type Endpoint3_6Request = Parameters[0] -type Endpoint3_6Input = { - readonly sessionID: Endpoint3_6Request["params"]["sessionID"] - readonly title: Endpoint3_6Request["payload"]["title"] +type Endpoint3_7Request = Parameters[0] +type Endpoint3_7Input = { + readonly sessionID: Endpoint3_7Request["params"]["sessionID"] + readonly title: Endpoint3_7Request["payload"]["title"] } -const Endpoint3_6 = (raw: RawClient["server.session"]) => (input: Endpoint3_6Input) => +const Endpoint3_7 = (raw: RawClient["server.session"]) => (input: Endpoint3_7Input) => raw["session.rename"]({ params: { sessionID: input["sessionID"] }, payload: { title: input["title"] } }).pipe( Effect.mapError(mapClientError), ) -type Endpoint3_7Request = Parameters[0] -type Endpoint3_7Input = { - readonly sessionID: Endpoint3_7Request["params"]["sessionID"] - readonly id?: Endpoint3_7Request["payload"]["id"] - readonly prompt: Endpoint3_7Request["payload"]["prompt"] - readonly delivery?: Endpoint3_7Request["payload"]["delivery"] - readonly resume?: Endpoint3_7Request["payload"]["resume"] +type Endpoint3_8Request = Parameters[0] +type Endpoint3_8Input = { + readonly sessionID: Endpoint3_8Request["params"]["sessionID"] + readonly id?: Endpoint3_8Request["payload"]["id"] + readonly prompt: Endpoint3_8Request["payload"]["prompt"] + readonly delivery?: Endpoint3_8Request["payload"]["delivery"] + readonly resume?: Endpoint3_8Request["payload"]["resume"] } -const Endpoint3_7 = (raw: RawClient["server.session"]) => (input: Endpoint3_7Input) => +const Endpoint3_8 = (raw: RawClient["server.session"]) => (input: Endpoint3_8Input) => raw["session.prompt"]({ params: { sessionID: input["sessionID"] }, payload: { id: input["id"], prompt: input["prompt"], delivery: input["delivery"], resume: input["resume"] }, @@ -133,23 +144,23 @@ const Endpoint3_7 = (raw: RawClient["server.session"]) => (input: Endpoint3_7Inp Effect.map((value) => value.data), ) -type Endpoint3_8Request = Parameters[0] -type Endpoint3_8Input = { readonly sessionID: Endpoint3_8Request["params"]["sessionID"] } -const Endpoint3_8 = (raw: RawClient["server.session"]) => (input: Endpoint3_8Input) => - raw["session.compact"]({ params: { sessionID: input["sessionID"] } }).pipe(Effect.mapError(mapClientError)) - -type Endpoint3_9Request = Parameters[0] +type Endpoint3_9Request = Parameters[0] type Endpoint3_9Input = { readonly sessionID: Endpoint3_9Request["params"]["sessionID"] } const Endpoint3_9 = (raw: RawClient["server.session"]) => (input: Endpoint3_9Input) => + raw["session.compact"]({ params: { sessionID: input["sessionID"] } }).pipe(Effect.mapError(mapClientError)) + +type Endpoint3_10Request = Parameters[0] +type Endpoint3_10Input = { readonly sessionID: Endpoint3_10Request["params"]["sessionID"] } +const Endpoint3_10 = (raw: RawClient["server.session"]) => (input: Endpoint3_10Input) => raw["session.wait"]({ params: { sessionID: input["sessionID"] } }).pipe(Effect.mapError(mapClientError)) -type Endpoint3_10Request = Parameters[0] -type Endpoint3_10Input = { - readonly sessionID: Endpoint3_10Request["params"]["sessionID"] - readonly messageID: Endpoint3_10Request["payload"]["messageID"] - readonly files?: Endpoint3_10Request["payload"]["files"] +type Endpoint3_11Request = Parameters[0] +type Endpoint3_11Input = { + readonly sessionID: Endpoint3_11Request["params"]["sessionID"] + readonly messageID: Endpoint3_11Request["payload"]["messageID"] + readonly files?: Endpoint3_11Request["payload"]["files"] } -const Endpoint3_10 = (raw: RawClient["server.session"]) => (input: Endpoint3_10Input) => +const Endpoint3_11 = (raw: RawClient["server.session"]) => (input: Endpoint3_11Input) => raw["session.revert.stage"]({ params: { sessionID: input["sessionID"] }, payload: { messageID: input["messageID"], files: input["files"] }, @@ -158,42 +169,42 @@ const Endpoint3_10 = (raw: RawClient["server.session"]) => (input: Endpoint3_10I Effect.map((value) => value.data), ) -type Endpoint3_11Request = Parameters[0] -type Endpoint3_11Input = { readonly sessionID: Endpoint3_11Request["params"]["sessionID"] } -const Endpoint3_11 = (raw: RawClient["server.session"]) => (input: Endpoint3_11Input) => - raw["session.revert.clear"]({ params: { sessionID: input["sessionID"] } }).pipe(Effect.mapError(mapClientError)) - -type Endpoint3_12Request = Parameters[0] +type Endpoint3_12Request = Parameters[0] type Endpoint3_12Input = { readonly sessionID: Endpoint3_12Request["params"]["sessionID"] } const Endpoint3_12 = (raw: RawClient["server.session"]) => (input: Endpoint3_12Input) => - raw["session.revert.commit"]({ params: { sessionID: input["sessionID"] } }).pipe(Effect.mapError(mapClientError)) + raw["session.revert.clear"]({ params: { sessionID: input["sessionID"] } }).pipe(Effect.mapError(mapClientError)) -type Endpoint3_13Request = Parameters[0] +type Endpoint3_13Request = Parameters[0] type Endpoint3_13Input = { readonly sessionID: Endpoint3_13Request["params"]["sessionID"] } const Endpoint3_13 = (raw: RawClient["server.session"]) => (input: Endpoint3_13Input) => + raw["session.revert.commit"]({ params: { sessionID: input["sessionID"] } }).pipe(Effect.mapError(mapClientError)) + +type Endpoint3_14Request = Parameters[0] +type Endpoint3_14Input = { readonly sessionID: Endpoint3_14Request["params"]["sessionID"] } +const Endpoint3_14 = (raw: RawClient["server.session"]) => (input: Endpoint3_14Input) => raw["session.context"]({ params: { sessionID: input["sessionID"] } }).pipe( Effect.mapError(mapClientError), Effect.map((value) => value.data), ) -type Endpoint3_14Request = Parameters[0] -type Endpoint3_14Input = { - readonly sessionID: Endpoint3_14Request["params"]["sessionID"] - readonly limit?: Endpoint3_14Request["query"]["limit"] - readonly after?: Endpoint3_14Request["query"]["after"] +type Endpoint3_15Request = Parameters[0] +type Endpoint3_15Input = { + readonly sessionID: Endpoint3_15Request["params"]["sessionID"] + readonly limit?: Endpoint3_15Request["query"]["limit"] + readonly after?: Endpoint3_15Request["query"]["after"] } -const Endpoint3_14 = (raw: RawClient["server.session"]) => (input: Endpoint3_14Input) => +const Endpoint3_15 = (raw: RawClient["server.session"]) => (input: Endpoint3_15Input) => raw["session.history"]({ params: { sessionID: input["sessionID"] }, query: { limit: input["limit"], after: input["after"] }, }).pipe(Effect.mapError(mapClientError)) -type Endpoint3_15Request = Parameters[0] -type Endpoint3_15Input = { - readonly sessionID: Endpoint3_15Request["params"]["sessionID"] - readonly after?: Endpoint3_15Request["query"]["after"] +type Endpoint3_16Request = Parameters[0] +type Endpoint3_16Input = { + readonly sessionID: Endpoint3_16Request["params"]["sessionID"] + readonly after?: Endpoint3_16Request["query"]["after"] } -const Endpoint3_15 = (raw: RawClient["server.session"]) => (input: Endpoint3_15Input) => +const Endpoint3_16 = (raw: RawClient["server.session"]) => (input: Endpoint3_16Input) => Stream.unwrap( raw["session.events"]({ params: { sessionID: input["sessionID"] }, query: { after: input["after"] } }).pipe( Effect.mapError(mapClientError), @@ -201,17 +212,17 @@ const Endpoint3_15 = (raw: RawClient["server.session"]) => (input: Endpoint3_15I ), ) -type Endpoint3_16Request = Parameters[0] -type Endpoint3_16Input = { readonly sessionID: Endpoint3_16Request["params"]["sessionID"] } -const Endpoint3_16 = (raw: RawClient["server.session"]) => (input: Endpoint3_16Input) => +type Endpoint3_17Request = Parameters[0] +type Endpoint3_17Input = { readonly sessionID: Endpoint3_17Request["params"]["sessionID"] } +const Endpoint3_17 = (raw: RawClient["server.session"]) => (input: Endpoint3_17Input) => raw["session.interrupt"]({ params: { sessionID: input["sessionID"] } }).pipe(Effect.mapError(mapClientError)) -type Endpoint3_17Request = Parameters[0] -type Endpoint3_17Input = { - readonly sessionID: Endpoint3_17Request["params"]["sessionID"] - readonly messageID: Endpoint3_17Request["params"]["messageID"] +type Endpoint3_18Request = Parameters[0] +type Endpoint3_18Input = { + readonly sessionID: Endpoint3_18Request["params"]["sessionID"] + readonly messageID: Endpoint3_18Request["params"]["messageID"] } -const Endpoint3_17 = (raw: RawClient["server.session"]) => (input: Endpoint3_17Input) => +const Endpoint3_18 = (raw: RawClient["server.session"]) => (input: Endpoint3_18Input) => raw["session.message"]({ params: { sessionID: input["sessionID"], messageID: input["messageID"] } }).pipe( Effect.mapError(mapClientError), Effect.map((value) => value.data), @@ -222,20 +233,21 @@ const adaptGroup3 = (raw: RawClient["server.session"]) => ({ create: Endpoint3_1(raw), active: Endpoint3_2(raw), get: Endpoint3_3(raw), - switchAgent: Endpoint3_4(raw), - switchModel: Endpoint3_5(raw), - rename: Endpoint3_6(raw), - prompt: Endpoint3_7(raw), - compact: Endpoint3_8(raw), - wait: Endpoint3_9(raw), - stage: Endpoint3_10(raw), - clear: Endpoint3_11(raw), - commit: Endpoint3_12(raw), - context: Endpoint3_13(raw), - history: Endpoint3_14(raw), - events: Endpoint3_15(raw), - interrupt: Endpoint3_16(raw), - message: Endpoint3_17(raw), + fork: Endpoint3_4(raw), + switchAgent: Endpoint3_5(raw), + switchModel: Endpoint3_6(raw), + rename: Endpoint3_7(raw), + prompt: Endpoint3_8(raw), + compact: Endpoint3_9(raw), + wait: Endpoint3_10(raw), + stage: Endpoint3_11(raw), + clear: Endpoint3_12(raw), + commit: Endpoint3_13(raw), + context: Endpoint3_14(raw), + history: Endpoint3_15(raw), + events: Endpoint3_16(raw), + interrupt: Endpoint3_17(raw), + message: Endpoint3_18(raw), }) type Endpoint4_0Request = Parameters[0] diff --git a/packages/client/src/generated/client.ts b/packages/client/src/generated/client.ts index e0ceb3bf2a..f7825f55fa 100644 --- a/packages/client/src/generated/client.ts +++ b/packages/client/src/generated/client.ts @@ -11,6 +11,8 @@ import type { SessionsActiveOutput, SessionsGetInput, SessionsGetOutput, + SessionsForkInput, + SessionsForkOutput, SessionsSwitchAgentInput, SessionsSwitchAgentOutput, SessionsSwitchModelInput, @@ -361,6 +363,18 @@ export function make(options: ClientOptions) { }, requestOptions, ).then((value) => value.data), + fork: (input: SessionsForkInput, requestOptions?: RequestOptions) => + request<{ readonly data: SessionsForkOutput }>( + { + method: "POST", + path: `/api/session/${encodeURIComponent(input.sessionID)}/fork`, + body: { messageID: input["messageID"] }, + successStatus: 200, + declaredStatuses: [404, 400, 401], + empty: false, + }, + requestOptions, + ).then((value) => value.data), switchAgent: (input: SessionsSwitchAgentInput, requestOptions?: RequestOptions) => request( { diff --git a/packages/client/src/generated/types.ts b/packages/client/src/generated/types.ts index 50aa222117..3c2bd6e5c4 100644 --- a/packages/client/src/generated/types.ts +++ b/packages/client/src/generated/types.ts @@ -33,6 +33,15 @@ export type SessionNotFoundError = { export const isSessionNotFoundError = (value: unknown): value is SessionNotFoundError => typeof value === "object" && value !== null && "_tag" in value && value["_tag"] === "SessionNotFoundError" +export type MessageNotFoundError = { + readonly _tag: "MessageNotFoundError" + readonly sessionID: string + readonly messageID: string + readonly message: string +} +export const isMessageNotFoundError = (value: unknown): value is MessageNotFoundError => + typeof value === "object" && value !== null && "_tag" in value && value["_tag"] === "MessageNotFoundError" + export type ConflictError = { readonly _tag: "ConflictError" readonly message: string @@ -65,15 +74,6 @@ export type UnknownError = { export const isUnknownError = (value: unknown): value is UnknownError => typeof value === "object" && value !== null && "_tag" in value && value["_tag"] === "UnknownError" -export type MessageNotFoundError = { - readonly _tag: "MessageNotFoundError" - readonly sessionID: string - readonly messageID: string - readonly message: string -} -export const isMessageNotFoundError = (value: unknown): value is MessageNotFoundError => - typeof value === "object" && value !== null && "_tag" in value && value["_tag"] === "MessageNotFoundError" - export type ProviderNotFoundError = { readonly _tag: "ProviderNotFoundError" readonly providerID: string @@ -377,6 +377,45 @@ export type SessionsGetOutput = { } }["data"] +export type SessionsForkInput = { + readonly sessionID: { readonly sessionID: string }["sessionID"] + readonly messageID?: { readonly messageID?: string | undefined }["messageID"] +} + +export type SessionsForkOutput = { + readonly data: { + readonly id: string + readonly parentID?: string + readonly projectID: string + readonly agent?: string + readonly model?: { readonly id: string; readonly providerID: string; readonly variant?: string } + readonly cost: number + readonly tokens: { + readonly input: number + readonly output: number + readonly reasoning: number + readonly cache: { readonly read: number; readonly write: number } + } + readonly time: { readonly created: number; readonly updated: number; readonly archived?: number } + readonly title: string + readonly location: { readonly directory: string; readonly workspaceID?: string } + readonly subpath?: string + readonly revert?: { + readonly messageID: string + readonly partID?: string + readonly snapshot?: string + readonly diff?: string + readonly files?: ReadonlyArray<{ + readonly path: string + readonly status: "added" | "modified" | "deleted" + readonly additions: number + readonly deletions: number + readonly patch: string + }> + } + } +}["data"] + export type SessionsSwitchAgentInput = { readonly sessionID: { readonly sessionID: string }["sessionID"] readonly agent: { readonly agent: string }["agent"] @@ -750,6 +789,24 @@ export type SessionsHistoryOutput = { readonly location?: { readonly directory: string; readonly workspaceID?: string } readonly data: { readonly timestamp: number; readonly sessionID: string; readonly title: string } } + | { + readonly id: string + readonly metadata?: { readonly [x: string]: JsonValue } + readonly type: "session.next.forked" + readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } + readonly location?: { readonly directory: string; readonly workspaceID?: string } + readonly data: { + readonly timestamp: number + readonly sessionID: 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 copiedSeq: number + } + } | { readonly id: string readonly metadata?: { readonly [x: string]: JsonValue } @@ -1216,6 +1273,24 @@ export type SessionsEventsOutput = readonly location?: { readonly directory: string; readonly workspaceID?: string } readonly data: { readonly timestamp: number; readonly sessionID: string; readonly title: string } } + | { + readonly id: string + readonly metadata?: { readonly [x: string]: unknown } + readonly type: "session.next.forked" + readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } + readonly location?: { readonly directory: string; readonly workspaceID?: string } + readonly data: { + readonly timestamp: number + readonly sessionID: 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 copiedSeq: number + } + } | { readonly id: string readonly metadata?: { readonly [x: string]: unknown } diff --git a/packages/core/src/event.ts b/packages/core/src/event.ts index c9195be79f..877b6aa8bf 100644 --- a/packages/core/src/event.ts +++ b/packages/core/src/event.ts @@ -3,7 +3,7 @@ export * as EventV2 from "./event" import { Cause, Context, Effect, Layer, Option, PubSub, Queue, Schema, Stream } from "effect" import { Event } from "@opencode-ai/schema/event" import type { Data, Definition, Payload } from "@opencode-ai/schema/event" -import { and, asc, eq, gt, inArray } from "drizzle-orm" +import { and, asc, eq, gt, inArray, sql } from "drizzle-orm" import { Database } from "./database/database" import { EventSequenceTable, EventTable } from "./event/sql" import { Location } from "./location" @@ -31,6 +31,22 @@ export const latestSequence = Effect.fn("EventV2.latestSequence")(function* ( return row?.seq ?? -1 }) +export const reserveSequence = Effect.fn("EventV2.reserveSequence")(function* ( + db: Database.Interface["db"], + aggregateID: string, + seq: number, +) { + yield* db + .insert(EventSequenceTable) + .values([{ aggregate_id: aggregateID, seq }]) + .onConflictDoUpdate({ + target: EventSequenceTable.aggregate_id, + set: { seq: sql`max(${EventSequenceTable.seq}, ${seq})` }, + }) + .run() + .pipe(Effect.orDie) +}) + export type SerializedEvent = { readonly id: ID readonly type: string @@ -327,7 +343,7 @@ export const layerWith = (options?: LayerOptions) => .onConflictDoUpdate({ target: EventSequenceTable.aggregate_id, set: { - seq, + seq: sql`max(${EventSequenceTable.seq}, ${seq})`, ...(input?.ownerID && row?.ownerID == null ? { owner_id: input.ownerID } : {}), }, }) diff --git a/packages/core/src/session.ts b/packages/core/src/session.ts index 1323c8aa3a..79da32aca7 100644 --- a/packages/core/src/session.ts +++ b/packages/core/src/session.ts @@ -90,6 +90,11 @@ type CompactInput = { sessionID: SessionSchema.ID } +type ForkInput = { + sessionID: SessionSchema.ID + messageID?: SessionMessage.ID +} + export class NotFoundError extends Schema.TaggedErrorClass()("Session.NotFoundError", { sessionID: SessionSchema.ID, }) {} @@ -113,11 +118,18 @@ export class BusyError extends Schema.TaggedErrorClass()("Session.Bus export const MessageNotFoundError = SessionRevert.MessageNotFoundError export type MessageNotFoundError = SessionRevert.MessageNotFoundError -export type Error = NotFoundError | MessageDecodeError | OperationUnavailableError | PromptConflictError | BusyError +export type Error = + | NotFoundError + | MessageDecodeError + | OperationUnavailableError + | PromptConflictError + | BusyError + | MessageNotFoundError export interface Interface { readonly list: (input?: ListInput) => Effect.Effect readonly create: (input: CreateInput) => Effect.Effect + readonly fork: (input: ForkInput) => Effect.Effect readonly get: (sessionID: SessionSchema.ID) => Effect.Effect readonly messages: (input: { sessionID: SessionSchema.ID @@ -272,6 +284,52 @@ export const layer = Layer.effect( // TODO: Restore recorded sessions onto replacement synchronized workspaces in a future API slice. return yield* result.get(sessionID).pipe(Effect.orDie) }), + fork: Effect.fn("V2Session.fork")(function* (input) { + const parent = yield* result.get(input.sessionID) + const boundary = input.messageID + ? yield* db + .select({ seq: SessionMessageTable.seq }) + .from(SessionMessageTable) + .where( + and(eq(SessionMessageTable.session_id, input.sessionID), eq(SessionMessageTable.id, input.messageID)), + ) + .get() + .pipe(Effect.orDie) + : undefined + if (input.messageID && !boundary) + 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() + yield* events.publish(SessionEvent.Forked, { + sessionID, + parentID: parent.id, + slug: Slug.create(), + title: forkTitle(parent.title), + agent: parent.agent, + model: parent.model, + messageID: input.messageID, + copiedSeq: copied?.seq ?? 0, + 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) + }), get: Effect.fn("V2Session.get")(function* (sessionID) { const session = yield* store.get(sessionID) if (!session) return yield* new NotFoundError({ sessionID }) @@ -505,6 +563,12 @@ export const defaultLayer = layer.pipe( 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) => Prompt.make({ text: input.text, diff --git a/packages/core/src/session/message-updater.ts b/packages/core/src/session/message-updater.ts index 3269aa1d72..cfb424620d 100644 --- a/packages/core/src/session/message-updater.ts +++ b/packages/core/src/session/message-updater.ts @@ -124,6 +124,7 @@ export function update(adapter: Adapter, event: SessionEvent.Event) { }, "session.next.moved": () => Effect.void, "session.next.renamed": () => Effect.void, + "session.next.forked": () => Effect.void, "session.next.prompted": (event) => { return adapter.appendMessage( SessionMessage.User.make({ diff --git a/packages/core/src/session/projector.ts b/packages/core/src/session/projector.ts index b202f94211..47acd7d937 100644 --- a/packages/core/src/session/projector.ts +++ b/packages/core/src/session/projector.ts @@ -1,6 +1,6 @@ export * as SessionProjector from "./projector" -import { and, desc, eq, gt, or, sql } from "drizzle-orm" +import { and, asc, desc, eq, gt, inArray, lt, or, sql } from "drizzle-orm" import { DateTime, Effect, Layer, Schema } from "effect" import { Database } from "../database/database" import { EventV2 } from "../event" @@ -17,6 +17,7 @@ import { MessageTable, PartTable, SessionInputTable, SessionMessageTable, Sessio import type { DeepMutable } from "../schema" type DatabaseService = Database.Interface["db"] +type MessageEvent = Exclude const decodeMessage = Schema.decodeUnknownSync(SessionMessage.Message) const encodeMessage = Schema.encodeSync(SessionMessage.Message) @@ -33,6 +34,13 @@ type Usage = { } } +const ForkBatchSize = 500 + +const emptyUsage = (): Usage => ({ + cost: 0, + tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } }, +}) + function usage(part: (typeof SessionV1.Event.PartUpdated.Type)["data"]["part"] | unknown): Usage | undefined { if (typeof part !== "object" || part === null) return undefined const value = part as Record @@ -41,6 +49,22 @@ function usage(part: (typeof SessionV1.Event.PartUpdated.Type)["data"]["part"] | return { cost: value.cost as Usage["cost"], tokens: value.tokens as Usage["tokens"] } } +function addUsage(target: Usage, value: Usage) { + target.cost += value.cost + target.tokens.input += value.tokens.input + target.tokens.output += value.tokens.output + target.tokens.reasoning += value.tokens.reasoning + target.tokens.cache.read += value.tokens.cache.read + target.tokens.cache.write += value.tokens.cache.write +} + +function messageUsage(row: typeof SessionMessageTable.$inferSelect): Usage | undefined { + if (row.type !== "assistant") return undefined + const message = decodeMessage({ ...row.data, id: row.id, type: row.type }) + if (message.type !== "assistant" || message.cost === undefined || message.tokens === undefined) return undefined + return { cost: message.cost, tokens: message.tokens } +} + function sessionRow(info: SessionV1.SessionInfo): typeof SessionTable.$inferInsert { return { id: info.id, @@ -109,7 +133,150 @@ function applyUsage( .pipe(Effect.orDie) } -function run(db: DatabaseService, event: SessionEvent.Event) { +const projectFork = Effect.fn("SessionProjector.projectFork")(function* ( + db: DatabaseService, + event: typeof SessionEvent.Forked.Type, +) { + const parent = yield* db + .select() + .from(SessionTable) + .where(eq(SessionTable.id, event.data.parentID)) + .get() + .pipe(Effect.orDie) + if (!parent) return yield* Effect.die(`Fork parent session not found: ${event.data.parentID}`) + + const stored = yield* db + .insert(SessionTable) + .values({ + id: event.data.sessionID, + parent_id: event.data.parentID, + project_id: parent.project_id, + workspace_id: parent.workspace_id, + slug: event.data.slug, + directory: parent.directory, + path: parent.path, + title: event.data.title, + agent: event.data.agent, + model: event.data.model, + version: parent.version, + cost: 0, + tokens_input: 0, + tokens_output: 0, + tokens_reasoning: 0, + tokens_cache_read: 0, + tokens_cache_write: 0, + time_created: DateTime.toEpochMillis(event.data.timestamp), + time_updated: DateTime.toEpochMillis(event.data.timestamp), + }) + .onConflictDoNothing() + .returning({ sessionID: SessionTable.id }) + .get() + .pipe(Effect.orDie) + if (!stored) return yield* Effect.die(new SessionAlreadyProjected()) + + const usage = emptyUsage() + let cursor = -1 + while (true) { + const rows = yield* db + .select() + .from(SessionMessageTable) + .where( + and( + eq(SessionMessageTable.session_id, event.data.parentID), + gt(SessionMessageTable.seq, cursor), + event.data.messageID === undefined ? undefined : lt(SessionMessageTable.seq, event.data.copiedSeq + 1), + ), + ) + .orderBy(asc(SessionMessageTable.seq)) + .limit(ForkBatchSize) + .all() + .pipe(Effect.orDie) + if (rows.length === 0) break + + const idMap = new Map(rows.map((row) => [row.id, SessionMessage.ID.create()])) + yield* db + .insert(SessionMessageTable) + .values( + rows.map((row) => { + const id = idMap.get(row.id) + if (!id) throw new Error(`Fork message ID mapping missing: ${row.id}`) + return { + id, + session_id: event.data.sessionID, + type: row.type, + seq: row.seq, + time_created: row.time_created, + time_updated: row.time_updated, + data: row.type === "synthetic" ? { ...row.data, sessionID: event.data.sessionID } : row.data, + } + }), + ) + .run() + .pipe(Effect.orDie) + + const inputRows = yield* db + .select() + .from(SessionInputTable) + .where( + and( + eq(SessionInputTable.session_id, event.data.parentID), + inArray( + SessionInputTable.id, + rows.map((row) => row.id), + ), + ), + ) + .all() + .pipe(Effect.orDie) + if (inputRows.length > 0) { + yield* db + .insert(SessionInputTable) + .values( + inputRows.flatMap((row) => { + const id = idMap.get(row.id) + return id + ? [ + { + id, + session_id: event.data.sessionID, + prompt: row.prompt, + delivery: row.delivery, + admitted_seq: row.admitted_seq, + promoted_seq: row.promoted_seq, + time_created: row.time_created, + }, + ] + : [] + }), + ) + .run() + .pipe(Effect.orDie) + } + + for (const row of rows) { + const value = messageUsage(row) + if (value) addUsage(usage, value) + } + cursor = rows.at(-1)!.seq + } + + yield* db + .update(SessionTable) + .set({ + cost: usage.cost, + tokens_input: usage.tokens.input, + tokens_output: usage.tokens.output, + tokens_reasoning: usage.tokens.reasoning, + tokens_cache_read: usage.tokens.cache.read, + tokens_cache_write: usage.tokens.cache.write, + }) + .where(eq(SessionTable.id, event.data.sessionID)) + .run() + .pipe(Effect.orDie) + if (event.data.copiedSeq > 0) yield* EventV2.reserveSequence(db, event.data.sessionID, event.data.copiedSeq) +}) + +function run(db: DatabaseService, event: MessageEvent) { return Effect.gen(function* () { const decodeRow = (row: typeof SessionMessageTable.$inferSelect) => decodeMessage({ ...row.data, id: row.id, type: row.type }) @@ -355,6 +522,7 @@ export const layer = Layer.effectDiscard( .run() .pipe(Effect.orDie), ) + yield* events.project(SessionEvent.Forked, (event) => projectFork(db, event)) yield* events.project(SessionEvent.Prompted, (event) => Effect.gen(function* () { if (event.durable === undefined) return yield* Effect.die("Durable Session event is missing aggregate sequence") diff --git a/packages/core/test/session-create.test.ts b/packages/core/test/session-create.test.ts index f573d968d6..df7e29db1e 100644 --- a/packages/core/test/session-create.test.ts +++ b/packages/core/test/session-create.test.ts @@ -1,6 +1,6 @@ import { describe, expect } from "bun:test" import path from "path" -import { Effect, Layer, Stream } from "effect" +import { DateTime, Effect, Layer, Stream } from "effect" import { AgentV2 } from "@opencode-ai/core/agent" import { asc, eq } from "drizzle-orm" import { Database } from "@opencode-ai/core/database/database" @@ -20,6 +20,7 @@ import { SessionProjector } from "@opencode-ai/core/session/projector" import { SessionExecution } from "@opencode-ai/core/session/execution" import { SessionInput } from "@opencode-ai/core/session/input" import { SessionEvent } from "@opencode-ai/core/session/event" +import { SessionMessage } from "@opencode-ai/core/session/message" import { SessionTable } from "@opencode-ai/core/session/sql" import { SessionStore } from "@opencode-ai/core/session/store" import { WorkspaceV2 } from "@opencode-ai/core/workspace" @@ -131,6 +132,94 @@ describe("SessionV2.create", () => { }), ) + it.effect("forks a session by replaying a durable fork event into copied projected rows", () => + Effect.gen(function* () { + const session = yield* SessionV2.Service + const events = yield* EventV2.Service + const { db } = yield* Database.Service + const parent = yield* session.create({ location, title: "Parent" }) + const admitted = yield* session.prompt({ + sessionID: parent.id, + prompt: Prompt.make({ text: "First" }), + resume: false, + }) + yield* SessionInput.promoteSteers(db, events, parent.id, Number.MAX_SAFE_INTEGER) + yield* events.publish(SessionEvent.Synthetic, { + sessionID: parent.id, + messageID: SessionMessage.ID.create(), + timestamp: yield* DateTime.now, + text: "parent note", + }) + + const forked = yield* session.fork({ sessionID: parent.id }) + const parentContext = yield* session.context(parent.id) + const forkContext = yield* session.context(forked.id) + const history = yield* session.history({ sessionID: forked.id, limit: 10 }) + + expect(forked).toMatchObject({ parentID: parent.id, title: "Parent (fork #1)" }) + expect(forkContext).toMatchObject([ + { type: "user", text: "First" }, + { type: "synthetic", text: "parent note", sessionID: forked.id }, + ]) + expect(forkContext.map((message) => message.id)).not.toEqual(parentContext.map((message) => message.id)) + expect(history.events).toHaveLength(1) + expect(history.events[0]).toMatchObject({ + type: "session.next.forked", + durable: { seq: 0 }, + data: { sessionID: forked.id, parentID: parent.id, copiedSeq: 3 }, + }) + expect(yield* SessionInput.find(db, forkContext[0]!.id)).toMatchObject({ + sessionID: forked.id, + prompt: { text: "First" }, + promotedSeq: 2, + }) + + yield* session.prompt({ sessionID: parent.id, prompt: Prompt.make({ text: "Parent changed" }), resume: false }) + yield* SessionInput.promoteSteers(db, events, parent.id, Number.MAX_SAFE_INTEGER) + yield* session.prompt({ sessionID: forked.id, prompt: Prompt.make({ text: "Child continues" }), resume: false }) + yield* SessionInput.promoteSteers(db, events, forked.id, Number.MAX_SAFE_INTEGER) + + expect((yield* session.context(parent.id)).map((message) => message.type)).toEqual(["user", "synthetic", "user"]) + expect((yield* session.context(forked.id)).map((message) => message.type)).toEqual(["user", "synthetic", "user"]) + expect((yield* session.context(forked.id)).at(-1)).toMatchObject({ text: "Child continues" }) + expect((yield* session.history({ sessionID: forked.id, limit: 10 })).events.map((event) => event.durable?.seq)).toEqual([ + 0, + 4, + 5, + ]) + expect(yield* SessionInput.find(db, admitted.id)).toMatchObject({ sessionID: parent.id }) + }), + ) + + it.effect("forks before the selected boundary message", () => + Effect.gen(function* () { + const session = yield* SessionV2.Service + const events = yield* EventV2.Service + const { db } = yield* Database.Service + const parent = yield* session.create({ location }) + const first = yield* session.prompt({ + sessionID: parent.id, + prompt: Prompt.make({ text: "First" }), + resume: false, + }) + yield* SessionInput.promoteSteers(db, events, parent.id, Number.MAX_SAFE_INTEGER) + const second = yield* session.prompt({ + sessionID: parent.id, + prompt: Prompt.make({ text: "Second" }), + resume: false, + }) + yield* SessionInput.promoteSteers(db, events, parent.id, Number.MAX_SAFE_INTEGER) + + const forked = yield* session.fork({ sessionID: parent.id, messageID: second.id }) + + const context = yield* session.context(forked.id) + const history = yield* session.history({ sessionID: forked.id, limit: 10 }) + expect(context).toMatchObject([{ text: "First" }]) + expect(context[0]?.id).not.toBe(first.id) + expect(history.events[0]).toMatchObject({ data: { copiedSeq: 2, messageID: second.id } }) + }), + ) + it.effect("returns the existing Session when one ID is reused with different create arguments", () => Effect.gen(function* () { const session = yield* SessionV2.Service diff --git a/packages/protocol/src/groups/session.ts b/packages/protocol/src/groups/session.ts index c33f2076fe..5706e14dbf 100644 --- a/packages/protocol/src/groups/session.ts +++ b/packages/protocol/src/groups/session.ts @@ -170,6 +170,23 @@ export const makeSessionGroup = (sessionLo }), ), ) + .add( + HttpApiEndpoint.post("session.fork", "/api/session/:sessionID/fork", { + params: { sessionID: Session.ID }, + payload: Schema.Struct({ messageID: SessionMessage.ID.pipe(Schema.optional) }), + success: Schema.Struct({ data: Session.Info }), + error: [SessionNotFoundError, MessageNotFoundError], + }) + .middleware(sessionLocationMiddleware) + .annotateMerge( + OpenApi.annotations({ + identifier: "v2.session.fork", + summary: "Fork session", + description: + "Create a child session by copying projected history from the parent. When messageID is supplied, copy messages before that boundary.", + }), + ), + ) .add( HttpApiEndpoint.post("session.switchAgent", "/api/session/:sessionID/agent", { params: { sessionID: Session.ID }, diff --git a/packages/schema/src/session-event.ts b/packages/schema/src/session-event.ts index f27457a29d..0322af14c2 100644 --- a/packages/schema/src/session-event.ts +++ b/packages/schema/src/session-event.ts @@ -94,6 +94,22 @@ export const Renamed = Event.define({ }) export type Renamed = typeof Renamed.Type +export const Forked = Event.define({ + type: "session.next.forked", + ...options, + schema: { + ...Base, + parentID: SessionID, + slug: Schema.String, + title: Schema.String, + agent: Schema.String.pipe(optional), + model: Model.Ref.pipe(optional), + messageID: SessionMessage.ID.pipe(optional), + copiedSeq: NonNegativeInt, + }, +}) +export type Forked = typeof Forked.Type + export const Prompted = Event.define({ type: "session.next.prompted", ...options, @@ -460,6 +476,7 @@ export const DurableDefinitions = Event.inventory( ModelSwitched, Moved, Renamed, + Forked, Prompted, PromptAdmitted, ContextUpdated, @@ -492,6 +509,7 @@ export const Definitions = Event.inventory( ModelSwitched, Moved, Renamed, + Forked, Prompted, PromptAdmitted, ContextUpdated, diff --git a/packages/server/src/handlers/session.ts b/packages/server/src/handlers/session.ts index cf092cc5e1..9be10c57ba 100644 --- a/packages/server/src/handlers/session.ts +++ b/packages/server/src/handlers/session.ts @@ -107,6 +107,32 @@ export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handl } }), ) + .handle( + "session.fork", + Effect.fn(function* (ctx) { + return { + data: yield* session.fork({ sessionID: ctx.params.sessionID, messageID: ctx.payload.messageID }).pipe( + Effect.catchTag( + "Session.NotFoundError", + (error) => + new SessionNotFoundError({ + sessionID: error.sessionID, + message: `Session not found: ${error.sessionID}`, + }), + ), + Effect.catchTag( + "Session.MessageNotFoundError", + (error) => + new MessageNotFoundError({ + sessionID: error.sessionID, + messageID: error.messageID, + message: `Message not found: ${error.messageID}`, + }), + ), + ), + } + }), + ) .handle( "session.switchAgent", Effect.fn(function* (ctx) {