feat(session): define explicit fork boundaries

packages/core/src/session/error.ts

@@ -10,6 +10,14 @@ export class NotFoundError extends Schema.TaggedErrorClass<NotFoundError>()("Ses
   sessionID: SessionSchema.ID,
 }) {}
 
+export class ForkEmptyError extends Schema.TaggedErrorClass<ForkEmptyError>()("Session.ForkEmptyError", {
+  sessionID: SessionSchema.ID,
+}) {
+  override get message() {
+    return `Cannot fork empty session: ${this.sessionID}`
+  }
+}
+
 export class MessageDecodeError extends Schema.TaggedErrorClass<MessageDecodeError>()("Session.MessageDecodeError", {
   sessionID: SessionSchema.ID,
   messageID: SessionMessage.ID,

packages/core/src/session/info.ts

@@ -8,7 +8,6 @@ import { AbsolutePath, RelativePath } from "../schema"
 import { Workspace } from "../workspace"
 import { SessionSchema } from "./schema"
 import { SessionTable } from "./sql"
-import { SessionMessage } from "./message"
 import { PersistedRevert } from "@opencode-ai/schema/session-revert"
 import { Money } from "@opencode-ai/schema/money"
 
@@ -20,12 +19,13 @@ export function fromRow(row: typeof SessionTable.$inferSelect): SessionSchema.In
     projectID: Project.ID.make(row.project_id),
     title: row.title,
     parentID: row.parent_id ? SessionSchema.ID.make(row.parent_id) : undefined,
-    fork: row.fork_session_id
-      ? {
-          sessionID: SessionSchema.ID.make(row.fork_session_id),
-          messageID: row.fork_message_id ? SessionMessage.ID.make(row.fork_message_id) : undefined,
-        }
-      : undefined,
+    fork:
+      row.fork_session_id && row.fork_boundary
+        ? {
+            sessionID: SessionSchema.ID.make(row.fork_session_id),
+            boundary: row.fork_boundary,
+          }
+        : undefined,
     agent: row.agent ? Agent.ID.make(row.agent) : undefined,
     model: row.model
       ? {

packages/core/src/session/instruction-state.ts

@@ -10,11 +10,12 @@ import { SessionEvent } from "./event"
 import { SessionMessage } from "./message"
 import { Event } from "@opencode-ai/schema/event"
 import { SessionSchema } from "./schema"
-import { InstructionBlobTable, InstructionStateTable, SessionTable } from "./sql"
+import { InstructionBlobTable, InstructionStateTable } from "./sql"
 
 type DatabaseService = Database.Interface["db"]
 
 const decodeInstructionsUpdated = Schema.decodeUnknownSync(SessionEvent.InstructionsUpdated.data)
+const decodeForked = Schema.decodeUnknownSync(SessionEvent.Forked.data)
 
 export interface Observation extends Instructions.Admission {
   readonly sessionID: SessionSchema.ID
@@ -94,6 +95,26 @@ export const apply = Effect.fn("InstructionState.apply")(function* (
     .pipe(Effect.orDie)
 })
 
+export const initialize = Effect.fn("InstructionState.initialize")(function* (
+  db: DatabaseService,
+  sessionID: SessionSchema.ID,
+  seq: number,
+  values: Instructions.Values,
+) {
+  yield* db
+    .insert(InstructionStateTable)
+    .values({
+      session_id: sessionID,
+      epoch_start: seq,
+      through_seq: seq,
+      initial_values: values,
+      current_values: values,
+    })
+    .onConflictDoNothing()
+    .run()
+    .pipe(Effect.orDie)
+})
+
 export const advanceEpoch = Effect.fn("InstructionState.advanceEpoch")(function* (
   db: DatabaseService,
   sessionID: SessionSchema.ID,
@@ -255,6 +276,14 @@ const stateFromEvents = Effect.fnUntraced(function* (db: DatabaseService, sessio
   return folded ? foldedState(sessionID, folded) : undefined
 })
 
+export const valuesAt = Effect.fn("InstructionState.valuesAt")(function* (
+  db: DatabaseService,
+  sessionID: SessionSchema.ID,
+  through: number,
+) {
+  return fold(yield* instructionEvents(db, sessionID, through))?.current
+})
+
 const latestRelevantSequence = Effect.fnUntraced(function* (db: DatabaseService, sessionID: SessionSchema.ID) {
   return yield* db
     .select({ seq: EventTable.seq })
@@ -324,15 +353,23 @@ const revertedEventType = Bus.versionedType(
   SessionEvent.RevertEvent.Committed.type,
   SessionEvent.RevertEvent.Committed.durable.version,
 )
-const relevantEventTypes = [instructionEventType, compactionEventType, movedEventType, revertedEventType]
+const forkedEventType = Bus.versionedType(SessionEvent.Forked.type, SessionEvent.Forked.durable.version)
+const relevantEventTypes = [
+  forkedEventType,
+  instructionEventType,
+  compactionEventType,
+  movedEventType,
+  revertedEventType,
+]
 
 type InstructionEventRow = typeof EventTable.$inferSelect
 
 const instructionEvents = Effect.fnUntraced(function* (
   db: DatabaseService,
   sessionID: SessionSchema.ID,
+  through?: number,
 ): Effect.fn.Return<ReadonlyArray<InstructionEventRow>> {
-  return yield* eventRows(db, sessionID, relevantEventTypes)
+  return yield* eventRows(db, sessionID, relevantEventTypes, undefined, through)
 })
 
 const instructionUpdatesAfter = Effect.fnUntraced(function* (
@@ -348,48 +385,22 @@ const eventRows = Effect.fnUntraced(function* (
   sessionID: SessionSchema.ID,
   types: ReadonlyArray<string>,
   after?: number,
-): Effect.fn.Return<ReadonlyArray<InstructionEventRow>> {
-  const segments = (yield* lineage(db, sessionID)).filter(
-    (segment) => after === undefined || segment.through === undefined || segment.through > after,
-  )
-  return (yield* Effect.forEach(segments, (segment) =>
-    db
-      .select()
-      .from(EventTable)
-      .where(
-        and(
-          eq(EventTable.aggregate_id, segment.sessionID),
-          inArray(EventTable.type, types),
-          segment.through === undefined ? undefined : lte(EventTable.seq, segment.through),
-          after === undefined ? undefined : gt(EventTable.seq, after),
-        ),
-      )
-      .orderBy(asc(EventTable.seq))
-      .all()
-      .pipe(Effect.orDie),
-  )).flat()
-})
-
-const lineage = Effect.fnUntraced(function* (
-  db: DatabaseService,
-  sessionID: SessionSchema.ID,
   through?: number,
-): Effect.fn.Return<ReadonlyArray<{ readonly sessionID: SessionSchema.ID; readonly through?: number }>> {
-  const session = yield* db
-    .select({ parentID: SessionTable.fork_session_id, forkSeq: SessionTable.fork_seq })
-    .from(SessionTable)
-    .where(eq(SessionTable.id, sessionID))
-    .get()
+): Effect.fn.Return<ReadonlyArray<InstructionEventRow>> {
+  return yield* db
+    .select()
+    .from(EventTable)
+    .where(
+      and(
+        eq(EventTable.aggregate_id, sessionID),
+        inArray(EventTable.type, types),
+        after === undefined ? undefined : gt(EventTable.seq, after),
+        through === undefined ? undefined : lte(EventTable.seq, through),
+      ),
+    )
+    .orderBy(asc(EventTable.seq))
+    .all()
     .pipe(Effect.orDie)
-  const inherited =
-    session?.parentID && session.forkSeq !== null
-      ? yield* lineage(
-          db,
-          session.parentID,
-          through === undefined ? session.forkSeq : Math.min(session.forkSeq, through),
-        )
-      : []
-  return [...inherited, { sessionID, ...(through === undefined ? {} : { through }) }]
 })
 
 function fold(rows: ReadonlyArray<InstructionEventRow>) {
@@ -402,6 +413,12 @@ function fold(rows: ReadonlyArray<InstructionEventRow>) {
       }
     | undefined
   >((state, row) => {
+    if (row.type === forkedEventType) {
+      const instructions = decodeForked(row.data).instructions
+      return instructions
+        ? { epochStart: row.seq, throughSeq: row.seq, initial: instructions, current: instructions }
+        : undefined
+    }
     if (row.type === movedEventType || row.type === revertedEventType) return undefined
     if (row.type === compactionEventType)
       return state

packages/core/src/session/projector.ts

@@ -1,6 +1,6 @@
 export * as SessionProjector from "./projector"
 
-import { and, asc, desc, eq, gt, gte, inArray, lt, sql } from "drizzle-orm"
+import { and, asc, desc, eq, gt, gte, inArray, lt, lte, sql } from "drizzle-orm"
 import { DateTime, Effect, Layer, Schema, Stream } from "effect"
 import { Database } from "../database/database"
 import { Bus } from "../bus"
@@ -174,25 +174,28 @@ const projectFork = Effect.fn("SessionProjector.projectFork")(function* (
     .get()
     .pipe(Effect.orDie)
   if (!parent) return yield* Effect.die(new Error(`Fork parent session not found: ${event.data.parentID}`))
-  const boundary = event.data.from
-    ? yield* db
-        .select({ seq: SessionMessageTable.seq })
-        .from(SessionMessageTable)
-        .where(
-          and(eq(SessionMessageTable.session_id, event.data.parentID), eq(SessionMessageTable.id, event.data.from)),
-        )
-        .get()
-        .pipe(Effect.orDie)
-    : undefined
-  if (event.data.from && !boundary)
-    return yield* Effect.die(new Error(`Fork boundary message not found: ${event.data.from}`))
+  const boundary = yield* db
+    .select({ seq: SessionMessageTable.seq })
+    .from(SessionMessageTable)
+    .where(
+      and(
+        eq(SessionMessageTable.session_id, event.data.parentID),
+        eq(SessionMessageTable.id, event.data.boundary.messageID),
+      ),
+    )
+    .get()
+    .pipe(Effect.orDie)
+  if (!boundary)
+    return yield* Effect.die(new Error(`Fork boundary message not found: ${event.data.boundary.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),
+        event.data.boundary.type === "before"
+          ? lt(SessionMessageTable.seq, boundary.seq)
+          : lte(SessionMessageTable.seq, boundary.seq),
       ),
     )
     .orderBy(desc(SessionMessageTable.seq))
@@ -207,8 +210,7 @@ const projectFork = Effect.fn("SessionProjector.projectFork")(function* (
       id: event.data.sessionID,
       parent_id: null,
       fork_session_id: event.data.parentID,
-      fork_message_id: event.data.from,
-      fork_seq: event.data.parentSeq,
+      fork_boundary: event.data.boundary,
       project_id: parent.project_id,
       workspace_id: parent.workspace_id,
       slug: Slug.create(),
@@ -314,8 +316,9 @@ const projectFork = Effect.fn("SessionProjector.projectFork")(function* (
 
     cursor = rows.at(-1)!.seq
   }
-  yield* Bus.reserveSequence(db, event.data.sessionID, event.data.parentSeq)
-  yield* InstructionState.rebuild(db, event.data.sessionID)
+  if (copiedSeq !== undefined) yield* Bus.reserveSequence(db, event.data.sessionID, copiedSeq)
+  if (event.data.instructions)
+    yield* InstructionState.initialize(db, event.data.sessionID, event.durable.seq, event.data.instructions)
 })
 
 function run(db: DatabaseService, event: MessageEvent) {

packages/core/src/session/sql.ts

@@ -32,8 +32,7 @@ export const SessionTable = sqliteTable(
     workspace_id: text().$type<Workspace.ID>(),
     parent_id: text().$type<SessionSchema.ID>(),
     fork_session_id: text().$type<SessionSchema.ID>(),
-    fork_message_id: text().$type<SessionMessage.ID>(),
-    fork_seq: integer(),
+    fork_boundary: text({ mode: "json" }).$type<Session.ForkBoundary>(),
     slug: text().notNull(),
     directory: directoryColumn().notNull(),
     path: pathColumn(),