fix: fold session fork into session store

packages/core/session/README.md

@@ -9,6 +9,8 @@ Creates and holds event-sourced `Session` instances. Persistence is intentionall
 ### Public API
 
 - `ctx.sessions.create(id?: SessionId, options?: { seed?: SessionEvent[]; meta?: { cwd?: string; parentSession?: SessionId; createdAt?: number; seedLength?: number } }): Session` — Create a session. `options.seed` replays/forks an existing event log; `options.meta` attaches creation metadata (validated absolute `cwd`, `parentSession` lineage, seed boundary) as the immutable `SessionHeader`. The store fills `version`/`id` and defaults `createdAt` to now; a caller reconstructing a persisted session passes the original `createdAt` and persisted `seedLength` to preserve them. Disposed with the calling fiber.
+- `ctx.sessions.snapshot(source: Session | SessionId): SessionForkSeed` — Resolve a live session object or id, reject non-boundary logs, and return a deep-cloned seed plus `parentSession` / `seedLength` metadata. Use this when the caller will pass the seed/meta into another creation path instead of creating a detached session immediately.
+- `ctx.sessions.fork({ source, sessionId? }): Session` — Convenience wrapper around `snapshot(source)` + `create(sessionId, { seed, meta })`; creates a live child session with lineage metadata.
 - `ctx.sessions.get(id: SessionId): Session | undefined`
 - `ctx.sessions.list(): Session[]`
 
@@ -67,9 +69,9 @@ Every `SessionEvent` carries two optional top-level fields (structural metadata)
 ### Extension points
 
 - Persistence plugins: subscribe to `session/event` (write-behind) and drain on `session/flush` (awaited) and fiber dispose. A durable backend reads the log and reloads it into a live session; the metadata seam (`SessionHeader`, `session.header`) is what such a backend stores beside the log.
-- Replay/fork: `ctx.sessions.create(id, { seed })` seeds a new session with an existing event log. The surface rebuilds deterministically from `surfaceOp` markers in the seeded events. The seed is validated to the SAME invariants `append` enforces — including that every surface-eligible event (`SurfaceEventType`) carries a `surfaceOp` marker — so a marker-less message event is rejected at construction rather than silently vanishing from `deriveMessages()` (the surface is the sole derivation path) on resume.
+- Replay/fork: `ctx.sessions.create(id, { seed })` seeds a new session with an existing event log. The surface rebuilds deterministically from `surfaceOp` markers in the seeded events. The seed is validated to the SAME invariants `append` enforces — including that every surface-eligible event (`SurfaceEventType`) carries a `surfaceOp` marker — so a marker-less message event is rejected at construction rather than silently vanishing from `deriveMessages()` (the surface is the sole derivation path) on resume. Ordinary live-session forks use `ctx.sessions.snapshot()` to validate an empty or `turn/end` boundary and build reusable seed metadata, or `ctx.sessions.fork()` to create the child session immediately.
 - Compaction: the `dsh-compact-basic` plugin appends a `user/message` with `surfaceOp: { op: 'replace', start, end }` to shadow old surface nodes behind a summary checkpoint.
 
 ### What is NOT here (TODO)
 
-- **Session branching/tree** (pi-style entry tree) — deferred unless needed beyond seed-based forking.
+- **Session branching/tree** (pi-style entry tree) — deferred unless needed beyond turn-boundary `snapshot()` / `fork()`.

packages/core/session/src/index.ts

@@ -318,6 +318,48 @@ export class Session {
   }
 }
 
+/** A fork source: either the live session object or its live store id. */
+export type SessionForkSource = Session | SessionId
+
+/** Metadata and seed events that can create a forked child session or agent. */
+export interface SessionForkSeed {
+  /** The resolved live source session. */
+  source: Session
+  /** Deep-cloned seed events copied from the source session at a turn boundary. */
+  seed: SessionEvent[]
+  /** Session creation metadata for the forked child. */
+  meta: {
+    /** The source session id. */
+    parentSession: SessionId
+    /** How many leading child events were inherited rather than produced. */
+    seedLength: number
+    /** The source session workspace, inherited by the child when present. */
+    cwd?: string
+  }
+}
+
+/** Inputs for the convenience session-creation path. */
+export interface ForkSessionOptions {
+  /** Live source session object or id. */
+  source: SessionForkSource
+  /** Optional child session id; omitted delegates to SessionStore's id policy. */
+  sessionId?: SessionId
+}
+
+export type SessionForkErrorCode =
+  | 'SESSION_NOT_FOUND'
+  | 'SESSION_NOT_LIVE'
+  | 'SESSION_ALREADY_EXISTS'
+  | 'OPEN_TURN'
+
+/** Typed error for session fork rejections. */
+export class SessionForkError extends Error {
+  constructor(message: string, public readonly code: SessionForkErrorCode) {
+    super(message)
+    this.name = 'SessionForkError'
+  }
+}
+
 /**
  * In-memory session store (`ctx.sessions`).
  *
@@ -452,6 +494,73 @@ export class SessionStore extends Service {
   list(): Session[] {
     return [...this.store.values()]
   }
+
+  /**
+   * Resolve and validate a live source session, then return a reusable deep-
+   * cloned fork seed. A non-empty source must end exactly at `turn/end`; this
+   * rejects open turns rather than clipping to an older boundary.
+   *
+   * @param source Live session object or live store id to snapshot.
+   * @returns Deep-cloned seed events plus child session metadata.
+   */
+  snapshot(source: SessionForkSource): SessionForkSeed {
+    const session = this._resolveForkSource(source)
+    this._assertForkBoundary(session)
+    const seed = session.events.map(event => structuredClone(event))
+    return {
+      source: session,
+      seed,
+      meta: {
+        ...session.header.cwd !== undefined ? { cwd: session.header.cwd } : {},
+        parentSession: session.id,
+        seedLength: seed.length,
+      },
+    }
+  }
+
+  /**
+   * Convenience path: create a live child session from a fork snapshot. Callers
+   * that create agents can use {@link snapshot} and pass its seed/meta through
+   * `ctx.agents.create` instead.
+   *
+   * @param options Source and optional child session id for the fork.
+   * @returns The created live child session.
+   */
+  fork(options: ForkSessionOptions): Session {
+    if (options.sessionId !== undefined && this.get(options.sessionId) !== undefined) {
+      throw new SessionForkError(`session "${options.sessionId}" already exists`, 'SESSION_ALREADY_EXISTS')
+    }
+    const snapshot = this.snapshot(options.source)
+    return this.create(options.sessionId, {
+      seed: snapshot.seed,
+      meta: snapshot.meta,
+    })
+  }
+
+  private _resolveForkSource(source: SessionForkSource): Session {
+    if (typeof source === 'string') {
+      const session = this.get(source)
+      if (session === undefined) throw new SessionForkError(`session "${source}" not found`, 'SESSION_NOT_FOUND')
+      return session
+    }
+
+    const live = this.get(source.id)
+    if (live === undefined) {
+      throw new SessionForkError(`session "${source.id}" not found`, 'SESSION_NOT_FOUND')
+    }
+    if (live !== source) throw new SessionForkError(`session "${source.id}" is not the live store instance`, 'SESSION_NOT_LIVE')
+    return source
+  }
+
+  private _assertForkBoundary(session: Session): void {
+    const last = session.events.at(-1)
+    if (last !== undefined && last.type !== 'turn/end') {
+      throw new SessionForkError(
+        `cannot fork session "${session.id}" inside an open turn (last event: ${last.type})`,
+        'OPEN_TURN',
+      )
+    }
+  }
 }
 
 export default SessionStore

packages/core/session/tests/fork.spec.ts

@@ -0,0 +1,186 @@
+import { describe, expect, it } from 'vitest'
+import { Context } from 'cordis'
+import { CallId } from '@deepseek-ai/dsh-llm'
+import SessionStore, { Session, SessionForkError, SessionId } from '@deepseek-ai/dsh-session'
+import type { SessionEvent, TurnEndReason } from '@deepseek-ai/dsh-session'
+
+async function setup(): Promise<{ ctx: Context; sessions: SessionStore }> {
+  const ctx = new Context()
+  await ctx.plugin(SessionStore)
+  return { ctx, sessions: ctx.sessions }
+}
+
+function appendClosedTurn(session: Session, reason: TurnEndReason = { kind: 'completed' }): void {
+  session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
+  session.append('user/message', {
+    content: [{ type: 'text', text: 'hello' }],
+    source: { kind: 'user' },
+  }, { surfaceOp: 'append' })
+  session.append('turn/end', { turn: 1, reason })
+}
+
+function firstUserMessage(events: readonly SessionEvent[]): SessionEvent<'user/message'> {
+  const event = events.find((e): e is SessionEvent<'user/message'> => e.type === 'user/message')
+  if (event === undefined) throw new Error('missing user/message')
+  return event
+}
+
+describe('SessionStore fork helpers', () => {
+  it('snapshots an empty live session as an empty seed with lineage metadata', async () => {
+    const { ctx, sessions } = await setup()
+    const source = ctx.sessions.create(SessionId('empty-parent'), { meta: { cwd: '/workspace' } })
+
+    const snapshot = sessions.snapshot(source)
+
+    expect(snapshot.source).toBe(source)
+    expect(snapshot.seed).toEqual([])
+    expect(snapshot.meta).toEqual({
+      cwd: '/workspace',
+      parentSession: SessionId('empty-parent'),
+      seedLength: 0,
+    })
+  })
+
+  it('snapshots a completed boundary by live session id and deep-clones seed events', async () => {
+    const { ctx, sessions } = await setup()
+    const source = ctx.sessions.create(SessionId('parent'), { meta: { cwd: '/workspace' } })
+    appendClosedTurn(source)
+
+    const snapshot = sessions.snapshot(SessionId('parent'))
+
+    expect(snapshot.source).toBe(source)
+    expect(snapshot.seed).toEqual(source.events)
+    expect(snapshot.seed).not.toBe(source.events)
+    expect(snapshot.seed[1]).not.toBe(source.events[1])
+    firstUserMessage(snapshot.seed).data.content[0] = { type: 'text', text: 'mutated' }
+    expect(firstUserMessage(source.events).data.content).toEqual([{ type: 'text', text: 'hello' }])
+    expect(snapshot.meta).toEqual({
+      cwd: '/workspace',
+      parentSession: SessionId('parent'),
+      seedLength: source.events.length,
+    })
+  })
+
+  it('accepts every turn/end reason as a fork boundary', async () => {
+    const { ctx, sessions } = await setup()
+    const reasons: TurnEndReason[] = [
+      { kind: 'completed' },
+      { kind: 'aborted', reason: 'cancelled by user' },
+      { kind: 'error', step: 1, message: 'model failed', code: 'MODEL' },
+      { kind: 'disposed' },
+      { kind: 'max-tokens' },
+      { kind: 'interrupted' },
+    ]
+
+    for (const reason of reasons) {
+      const source = ctx.sessions.create(SessionId(`parent-${reason.kind}`))
+      appendClosedTurn(source, reason)
+
+      const snapshot = sessions.snapshot(source)
+
+      expect(snapshot.seed.at(-1)?.type).toBe('turn/end')
+      expect(snapshot.meta.seedLength).toBe(source.events.length)
+    }
+  })
+
+  it('rejects an unknown live session id', async () => {
+    const { sessions } = await setup()
+
+    expect(() => sessions.snapshot(SessionId('missing')))
+      .toThrow(new SessionForkError('session "missing" not found', 'SESSION_NOT_FOUND'))
+  })
+
+  it('rejects a detached Session object that is not live in ctx.sessions', async () => {
+    const { sessions } = await setup()
+    const detached = new Session(SessionId('detached'))
+
+    expect(() => sessions.snapshot(detached))
+      .toThrow(new SessionForkError('session "detached" not found', 'SESSION_NOT_FOUND'))
+  })
+
+  it('rejects a stale Session object whose id is live on a different instance', async () => {
+    const { ctx, sessions } = await setup()
+    ctx.sessions.create(SessionId('same-id'))
+    const stale = new Session(SessionId('same-id'))
+
+    expect(() => sessions.snapshot(stale))
+      .toThrow(new SessionForkError('session "same-id" is not the live store instance', 'SESSION_NOT_LIVE'))
+  })
+
+  it('rejects non-empty logs whose last event is not turn/end', async () => {
+    const { ctx, sessions } = await setup()
+    const cases: [string, (session: Session) => void][] = [
+      ['turn/start', (session) => {
+        session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
+      }],
+      ['step/start', (session) => {
+        session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
+        session.append('step/start', { turn: 1, step: 1 })
+      }],
+      ['user/message', (session) => {
+        session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
+        session.append('user/message', { content: [{ type: 'text', text: 'open' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
+      }],
+      ['assistant/message', (session) => {
+        session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
+        session.append('step/start', { turn: 1, step: 1 })
+        session.append('assistant/message', { turn: 1, step: 1, content: [{ type: 'text', text: 'partial' }] }, { surfaceOp: 'append' })
+      }],
+      ['tool/call', (session) => {
+        const callId = CallId('call-open')
+        session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
+        session.append('step/start', { turn: 1, step: 1 })
+        session.append('assistant/message', {
+          turn: 1,
+          step: 1,
+          content: [{ type: 'tool-call', id: callId, name: 'bash', arguments: '{}' }],
+        }, { surfaceOp: 'append' })
+        session.append('tool/call', { turn: 1, step: 1, callId, name: 'bash', arguments: '{}' })
+      }],
+    ]
+
+    for (const [lastType, build] of cases) {
+      const source = ctx.sessions.create(SessionId(`open-${lastType}`))
+      build(source)
+
+      expect(() => sessions.snapshot(source))
+        .toThrow(new SessionForkError(`cannot fork session "open-${lastType}" inside an open turn (last event: ${lastType})`, 'OPEN_TURN'))
+    }
+  })
+
+  it('creates a forked child session with the seed and lineage metadata', async () => {
+    const { ctx, sessions } = await setup()
+    const source = ctx.sessions.create(SessionId('parent'), { meta: { cwd: '/workspace' } })
+    appendClosedTurn(source)
+
+    const child = sessions.fork({ source, sessionId: SessionId('child') })
+
+    expect(child.id).toBe(SessionId('child'))
+    expect(child.events).toEqual(source.events)
+    expect(child.header.parentSession).toBe(source.id)
+    expect(child.header.seedLength).toBe(source.events.length)
+    expect(child.header.cwd).toBe('/workspace')
+    firstUserMessage(child.events).data.content[0] = { type: 'text', text: 'child mutation' }
+    expect(firstUserMessage(source.events).data.content).toEqual([{ type: 'text', text: 'hello' }])
+  })
+
+  it('rejects a child session id that is already live with a typed fork error', async () => {
+    const { ctx, sessions } = await setup()
+    const source = ctx.sessions.create(SessionId('parent'))
+    appendClosedTurn(source)
+    ctx.sessions.create(SessionId('child'))
+
+    expect(() => sessions.fork({ source, sessionId: SessionId('child') }))
+      .toThrow(new SessionForkError('session "child" already exists', 'SESSION_ALREADY_EXISTS'))
+  })
+
+  it('rejects a duplicate child session id before validating the source boundary', async () => {
+    const { ctx, sessions } = await setup()
+    const source = ctx.sessions.create(SessionId('open-parent'))
+    source.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
+    ctx.sessions.create(SessionId('child'))
+
+    expect(() => sessions.fork({ source, sessionId: SessionId('child') }))
+      .toThrow(new SessionForkError('session "child" already exists', 'SESSION_ALREADY_EXISTS'))
+  })
+})