fix: keep fork boundary check simple

packages/core/session/README.md

@@ -9,7 +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 selected prefix to be turn-enclosed, and create a live child session with lineage metadata.
+- `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[]`
 
@@ -68,7 +68,7 @@ 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. Ordinary live-session forks use `ctx.sessions.fork({ source, boundary?, childSessionId? })`, where `boundary` is the inclusive source event seq to fork through.
+- 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)

packages/core/session/src/index.ts

@@ -487,8 +487,7 @@ export class SessionStore extends Service {
   /**
    * 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 be turn-enclosed and end
-   * at `turn/end`; this rejects open turns rather than clipping silently.
+   * current last event. A non-empty selected slice must end at `turn/end`.
    *
    * @param options Source, optional boundary, and optional child id for the fork.
    * @returns The created live child session.
@@ -540,10 +539,14 @@ export class SessionStore extends Service {
         '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',
+      )
+    }
 
-    const seed = events.slice(0, boundary + 1)
-    this._assertForkBoundary(session, seed, boundary)
-    return seed.map(event => structuredClone(event))
+    return events.slice(0, boundary + 1).map(event => structuredClone(event))
   }
 
   private _resolveForkSource(source: SessionForkSource): Session {
@@ -561,50 +564,6 @@ export class SessionStore extends Service {
     return source
   }
 
-  private _assertForkBoundary(session: Session, seed: readonly SessionEvent[], boundary: number): void {
-    let openTurn: SessionEvent<'turn/start'> | undefined
-    for (const event of seed) {
-      switch (event.type) {
-        case 'turn/start': {
-          if (openTurn !== undefined) {
-            throw new SessionForkError(
-              `cannot fork session "${session.id}" at boundary ${boundary}: turn ${event.data.turn} starts before turn ${openTurn.data.turn} ended`,
-              'OPEN_TURN',
-            )
-          }
-          openTurn = event
-          break
-        }
-        case 'turn/end': {
-          if (openTurn === undefined) {
-            throw new SessionForkError(
-              `cannot fork session "${session.id}" at boundary ${boundary}: turn/end at seq ${event.seq} has no matching turn/start`,
-              'OPEN_TURN',
-            )
-          }
-          openTurn = undefined
-          break
-        }
-        default: {
-          if (openTurn === undefined) {
-            throw new SessionForkError(
-              `cannot fork session "${session.id}" at boundary ${boundary}: event ${event.seq} (${event.type}) is outside a turn`,
-              'OPEN_TURN',
-            )
-          }
-          break
-        }
-      }
-    }
-
-    const last = seed.at(-1)
-    if (openTurn !== undefined || last?.type !== 'turn/end') {
-      throw new SessionForkError(
-        `cannot fork session "${session.id}" at boundary ${boundary}: slice ends inside an open turn (last event: ${last?.type ?? 'none'})`,
-        'OPEN_TURN',
-      )
-    }
-  }
 }
 
 export default SessionStore

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

@@ -210,38 +210,10 @@ describe('SessionStore.fork', () => {
       const boundary = build(source)
 
       expect(() => sessions.fork({ source, boundary }))
-        .toThrow(new SessionForkError(`cannot fork session "open-${lastType}" at boundary ${boundary}: slice ends inside an open turn (last event: ${lastType})`, 'OPEN_TURN'))
+        .toThrow(new SessionForkError(`fork boundary ${boundary} in session "open-${lastType}" must be turn/end, got ${lastType}`, 'OPEN_TURN'))
     }
   })
 
-  it('rejects malformed turn enclosure in the selected slice', async () => {
-    const { ctx, sessions } = await setup()
-    const outside = ctx.sessions.create(SessionId('outside'), {
-      seed: [
-        { type: 'step/start', seq: 0, time: 1, data: { turn: 1, step: 1 } },
-      ],
-    })
-    expect(() => sessions.fork({ source: outside, boundary: 0 }))
-      .toThrow(new SessionForkError('cannot fork session "outside" at boundary 0: event 0 (step/start) is outside a turn', 'OPEN_TURN'))
-
-    const nested = ctx.sessions.create(SessionId('nested'), {
-      seed: [
-        { type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } },
-        { type: 'turn/start', seq: 1, time: 2, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } },
-      ],
-    })
-    expect(() => sessions.fork({ source: nested, boundary: 1 }))
-      .toThrow(new SessionForkError('cannot fork session "nested" at boundary 1: turn 2 starts before turn 1 ended', 'OPEN_TURN'))
-
-    const orphanEnd = ctx.sessions.create(SessionId('orphan-end'), {
-      seed: [
-        { type: 'turn/end', seq: 0, time: 1, data: { turn: 1, reason: { kind: 'completed' } } },
-      ],
-    })
-    expect(() => sessions.fork({ source: orphanEnd, boundary: 0 }))
-      .toThrow(new SessionForkError('cannot fork session "orphan-end" at boundary 0: turn/end at seq 0 has no matching turn/start', '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'))

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())
@@ -417,7 +419,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)