Merge remote-tracking branch 'origin/codex/ask-user-question' into codex/ask-user-question

packages/core/session/README.md

@@ -9,6 +9,7 @@ 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.fork(source, boundary?, childSessionId?): Session` — Resolve a live session object or id, select a seed through the inclusive `boundary` event seq (default: current last event), require that boundary to be `turn/end`, and create a live child session with lineage metadata.
 - `ctx.sessions.get(id: SessionId): Session | undefined`
 - `ctx.sessions.list(): Session[]`
 
@@ -72,9 +73,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 always-on invariants `append` enforces — contiguous seqs, JSON-serializable data, and required `surfaceOp` markers on surface-eligible events — so marker-less message events are rejected at construction rather than silently vanishing from `deriveMessages()`. Broader turn-enclosure checks stay in `dsh-invariants` and persistence repair. Ordinary live-session forks use `ctx.sessions.fork(source, boundary?, childSessionId?)`, where `boundary` is the inclusive source event seq to fork through.
 - 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 boundary-based `fork()`.

packages/core/session/src/index.ts

@@ -362,6 +362,24 @@ export class Session {
   }
 }
 
+/** A fork source: either the live session object or its live store id. */
+export type SessionForkSource = Session | SessionId
+
+export type SessionForkErrorCode =
+  | 'SESSION_NOT_FOUND'
+  | 'SESSION_NOT_LIVE'
+  | 'SESSION_ALREADY_EXISTS'
+  | 'INVALID_BOUNDARY'
+  | '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`).
  *
@@ -496,6 +514,92 @@ export class SessionStore extends Service {
   list(): Session[] {
     return [...this.store.values()]
   }
+
+  /**
+   * Create a live child session from a turn-enclosed prefix of a live source.
+   * `boundary` is an inclusive source event seq; omitted means the source's
+   * current last event. A non-empty selected slice must end at `turn/end`.
+   *
+   * @param source - Live source session object or id.
+   * @param boundary - Inclusive source event seq to fork through; omitted means
+   *   the source's current last event, and omitted on an empty source forks an
+   *   empty child.
+   * @param childSessionId - Optional child session id; omitted delegates to
+   *   `SessionStore`'s id policy.
+   * @returns The created live child session.
+   */
+  fork(source: SessionForkSource, boundary?: number, childSessionId?: SessionId): Session {
+    if (childSessionId !== undefined && this.get(childSessionId) !== undefined) {
+      throw new SessionForkError(`session "${childSessionId}" already exists`, 'SESSION_ALREADY_EXISTS')
+    }
+    const liveSource = this._resolveForkSource(source)
+    const seed = this._forkSeed(liveSource, boundary)
+    return this.create(childSessionId, {
+      seed,
+      meta: {
+        ...liveSource.header.cwd !== undefined ? { cwd: liveSource.header.cwd } : {},
+        parentSession: liveSource.id,
+        seedLength: seed.length,
+      },
+    })
+  }
+
+  private _forkSeed(session: Session, requestedBoundary: number | undefined): SessionEvent[] {
+    const events = session.events
+    const lastEvent = events.at(-1)
+    let boundary: number
+    if (requestedBoundary !== undefined) {
+      boundary = requestedBoundary
+    } else {
+      if (lastEvent === undefined) return []
+      boundary = lastEvent.seq
+    }
+    if (!Number.isSafeInteger(boundary) || boundary < 0) {
+      throw new SessionForkError(
+        `fork boundary for session "${session.id}" must be a non-negative safe integer, got ${String(boundary)}`,
+        'INVALID_BOUNDARY',
+      )
+    }
+    if (boundary >= events.length) {
+      const lastSeq = events.at(-1)?.seq
+      throw new SessionForkError(
+        `fork boundary ${boundary} does not exist in session "${session.id}" (last seq: ${lastSeq ?? 'none'})`,
+        'INVALID_BOUNDARY',
+      )
+    }
+
+    const boundaryEvent = events[boundary]
+    if (boundaryEvent === undefined || boundaryEvent.seq !== boundary) {
+      throw new SessionForkError(
+        `fork boundary ${boundary} does not match a contiguous event seq in session "${session.id}"`,
+        'INVALID_BOUNDARY',
+      )
+    }
+    if (boundaryEvent.type !== 'turn/end') {
+      throw new SessionForkError(
+        `fork boundary ${boundary} in session "${session.id}" must be turn/end, got ${boundaryEvent.type}`,
+        'OPEN_TURN',
+      )
+    }
+
+    return events.slice(0, boundary + 1).map(event => structuredClone(event))
+  }
+
+  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
+  }
+
 }
 
 export default SessionStore

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

@@ -0,0 +1,240 @@
+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,
+  turn: number,
+  text = `hello ${turn}`,
+  reason: TurnEndReason = { kind: 'completed' },
+): void {
+  session.append('turn/start', { turn, trigger: { kind: 'message', source: { kind: 'user' } } })
+  session.append('user/message', {
+    content: [{ type: 'text', text }],
+    source: { kind: 'user' },
+  }, { surfaceOp: 'append' })
+  session.append('turn/end', { turn, reason })
+}
+
+function appendOpenTurn(session: Session, turn: number): void {
+  session.append('turn/start', { turn, trigger: { kind: 'message', source: { kind: 'user' } } })
+  session.append('user/message', {
+    content: [{ type: 'text', text: `open ${turn}` }],
+    source: { kind: 'user' },
+  }, { surfaceOp: 'append' })
+}
+
+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
+}
+
+function lastSeq(session: Session): number {
+  const event = session.events.at(-1)
+  if (event === undefined) throw new Error('missing last event')
+  return event.seq
+}
+
+describe('SessionStore.fork', () => {
+  it('forks an empty live session as an empty child with lineage metadata', async () => {
+    const { ctx, sessions } = await setup()
+    const source = ctx.sessions.create(SessionId('empty-parent'), { meta: { cwd: '/workspace' } })
+
+    const child = sessions.fork(source, undefined, SessionId('empty-child'))
+
+    expect(child.events).toEqual([])
+    expect(child.header).toMatchObject({
+      id: SessionId('empty-child'),
+      cwd: '/workspace',
+      parentSession: SessionId('empty-parent'),
+      seedLength: 0,
+    })
+  })
+
+  it('forks the latest completed boundary by default and deep-clones seed events', async () => {
+    const { ctx, sessions } = await setup()
+    const source = ctx.sessions.create(SessionId('parent'), { meta: { cwd: '/workspace' } })
+    appendClosedTurn(source, 1, 'hello')
+
+    const child = sessions.fork(SessionId('parent'), undefined, SessionId('child'))
+
+    expect(child.events).toEqual(source.events)
+    expect(child.events).not.toBe(source.events)
+    expect(child.events[1]).not.toBe(source.events[1])
+    firstUserMessage(child.events).data.content[0] = { type: 'text', text: 'child mutation' }
+    expect(firstUserMessage(source.events).data.content).toEqual([{ type: 'text', text: 'hello' }])
+    expect(child.header).toMatchObject({
+      id: SessionId('child'),
+      cwd: '/workspace',
+      parentSession: SessionId('parent'),
+      seedLength: source.events.length,
+    })
+  })
+
+  it('forks from an earlier turn boundary even when the source currently has an open tail', async () => {
+    const { ctx, sessions } = await setup()
+    const source = ctx.sessions.create(SessionId('parent'), { meta: { cwd: '/workspace' } })
+    appendClosedTurn(source, 1, 'first')
+    const firstBoundary = lastSeq(source)
+    appendClosedTurn(source, 2, 'second')
+    appendOpenTurn(source, 3)
+
+    const child = sessions.fork(source, firstBoundary, SessionId('child-from-first'))
+
+    expect(child.events).toEqual(source.events.slice(0, firstBoundary + 1))
+    expect(child.header.seedLength).toBe(firstBoundary + 1)
+    expect(child.deriveMessages()).toEqual([{ role: 'user', content: [{ type: 'text', text: 'first' }] }])
+  })
+
+  it('accepts every turn/end reason as an explicit 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, 1, reason.kind, reason)
+
+      const child = sessions.fork(source, lastSeq(source), SessionId(`child-${reason.kind}`))
+
+      expect(child.events.at(-1)?.type).toBe('turn/end')
+      expect(child.header.seedLength).toBe(source.events.length)
+    }
+  })
+
+  it('rejects invalid boundaries before creating a child', async () => {
+    const { ctx, sessions } = await setup()
+    const empty = ctx.sessions.create(SessionId('empty'))
+    expect(() => sessions.fork(empty, 0, SessionId('empty-child')))
+      .toThrow(new SessionForkError('fork boundary 0 does not exist in session "empty" (last seq: none)', 'INVALID_BOUNDARY'))
+    expect(ctx.sessions.get(SessionId('empty-child'))).toBeUndefined()
+
+    const source = ctx.sessions.create(SessionId('parent'))
+    appendClosedTurn(source, 1)
+    expect(() => sessions.fork(source, -1, SessionId('negative')))
+      .toThrow(/non-negative safe integer/)
+    expect(() => sessions.fork(source, 0.5, SessionId('fraction')))
+      .toThrow(/non-negative safe integer/)
+    expect(() => sessions.fork(source, Number.MAX_SAFE_INTEGER + 1, SessionId('unsafe')))
+      .toThrow(/non-negative safe integer/)
+    expect(() => sessions.fork(source, source.seq, SessionId('past-end')))
+      .toThrow(new SessionForkError(`fork boundary ${source.seq} does not exist in session "parent" (last seq: ${source.seq - 1})`, 'INVALID_BOUNDARY'))
+  })
+
+  it('rejects a corrupted live source whose array index no longer matches event seq', async () => {
+    const { ctx, sessions } = await setup()
+    const source = ctx.sessions.create(SessionId('corrupt-parent'))
+    appendClosedTurn(source, 1)
+    const mutableLog = (source as unknown as { log: SessionEvent[] }).log
+    mutableLog[2] = { ...mutableLog[2]!, seq: 99 }
+
+    expect(() => sessions.fork(source, 2, SessionId('corrupt-child')))
+      .toThrow(new SessionForkError('fork boundary 2 does not match a contiguous event seq in session "corrupt-parent"', 'INVALID_BOUNDARY'))
+    expect(ctx.sessions.get(SessionId('corrupt-child'))).toBeUndefined()
+  })
+
+  it('rejects an unknown live session id', async () => {
+    const { sessions } = await setup()
+
+    expect(() => sessions.fork(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.fork(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.fork(stale))
+      .toThrow(new SessionForkError('session "same-id" is not the live store instance', 'SESSION_NOT_LIVE'))
+  })
+
+  it('rejects selected slices whose boundary is inside an open turn', async () => {
+    const { ctx, sessions } = await setup()
+    const cases: [string, (session: Session) => number][] = [
+      ['turn/start', (session) => {
+        session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
+        return lastSeq(session)
+      }],
+      ['step/start', (session) => {
+        session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
+        session.append('step/start', { turn: 1, step: 1 })
+        return lastSeq(session)
+      }],
+      ['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' })
+        return lastSeq(session)
+      }],
+      ['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' })
+        return lastSeq(session)
+      }],
+      ['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: '{}' })
+        return lastSeq(session)
+      }],
+    ]
+
+    for (const [lastType, build] of cases) {
+      const source = ctx.sessions.create(SessionId(`open-${lastType}`))
+      const boundary = build(source)
+
+      expect(() => sessions.fork(source, boundary))
+        .toThrow(new SessionForkError(`fork boundary ${boundary} in session "open-${lastType}" must be turn/end, got ${lastType}`, 'OPEN_TURN'))
+    }
+  })
+
+  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, 1)
+    ctx.sessions.create(SessionId('child'))
+
+    expect(() => sessions.fork(source, undefined, SessionId('child')))
+      .toThrow(new SessionForkError('session "child" already exists', 'SESSION_ALREADY_EXISTS'))
+  })
+
+  it('rejects a duplicate child session id before validating the 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, undefined, SessionId('child')))
+      .toThrow(new SessionForkError('session "child" already exists', 'SESSION_ALREADY_EXISTS'))
+  })
+})

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

@@ -61,8 +61,10 @@ describe('Session', () => {
 
   it('replays identically from a seeded event log', () => {
     const original = new Session(SessionId('s3'))
+    original.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
     original.append('user/message', { content: [{ type: 'text', text: 'q' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
     original.append('assistant/message', { turn: 1, step: 1, content: [{ type: 'text', text: 'a' }] }, { surfaceOp: 'append' })
+    original.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
 
     const replayed = new Session(SessionId('s3-replay'), [...original.events])
     expect(replayed.deriveMessages()).toEqual(original.deriveMessages())
@@ -423,7 +425,9 @@ describe('todo/write event', () => {
 
   it('round-trips through a seeded replay identically (durable, no surfaceOp needed)', () => {
     const original = new Session(SessionId('t4'))
+    original.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
     original.append('todo/write', { todos: [{ content: 'only', status: 'completed' }] })
+    original.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
     // Seeding a non-surface event with no surfaceOp must not throw.
     const replayed = new Session(SessionId('t4-replay'), [...original.events])
     expect(replayed.events.findLast(e => e.type === 'todo/write')!.data.todos)

packages/session-persistence/session-persistence-jsonl/tests/jsonl.spec.ts

@@ -23,6 +23,15 @@ afterEach(async () => {
   for (const d of dirs.splice(0)) await rm(d, { recursive: true, force: true })
 })
 
+function appendClosedTurn(session: Session): 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: { kind: 'completed' } })
+}
+
 // Run the shared backend contract against the real JSONL backend.
 runPersistenceContract('jsonl', async () => {
   const dir = await mkdtemp(join(tmpdir(), 'dsh-jsonl-'))
@@ -131,6 +140,23 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
     expect(loaded.events).toEqual(log) // chunks preserved, contiguous seqs
   })
 
+  it('persists a forked child seed through the existing session write path', async () => {
+    const source = ctx.sessions.create(SessionId('persist-parent'), { meta: { cwd: '/workspace' } })
+    appendClosedTurn(source)
+
+    const child = ctx.sessions.fork(source, undefined, SessionId('persist-child'))
+    await ctx.parallel('session/flush', child)
+    const loaded = await ctx.sessionPersistence.load(child.id)
+
+    expect(loaded.events).toEqual(source.events)
+    expect(loaded.meta).toMatchObject({
+      id: SessionId('persist-child'),
+      cwd: '/workspace',
+      parentSession: SessionId('persist-parent'),
+      seedLength: source.events.length,
+    })
+  })
+
   it('crash recovery: load preserves the interrupted turn and closes it with a synthetic turn/end {interrupted}', async () => {
     const m = meta('crash', '/proj')
     await ctx.sessionPersistence.create(m)