fix(session): share live append validation and expose batch publication

docs/subsystems/session.md

@@ -584,6 +584,8 @@ declare class Session {
   isOwnSeq(seq: SessionSeq): boolean;
   /** The next event's sequence number — always the log length (the `seq = log.length` contiguity contract). */
   get seq(): SessionLogOffset;
+  /** Current member and final sequence while synchronously notifying a batch's observers. */
+  get currentBatchPublication(): { readonly event: SessionEvent; readonly endSeq: SessionSeq } | undefined;
   /**
    * Append consecutive events together after validating every candidate and dispatch.
    * Observers read the complete committed log and final message surface.

docs/subsystems/session.zh.md

@@ -586,6 +586,8 @@ declare class Session {
   isOwnSeq(seq: SessionSeq): boolean;
   /** The next event's sequence number — always the log length (the `seq = log.length` contiguity contract). */
   get seq(): SessionLogOffset;
+  /** Current member and final sequence while synchronously notifying a batch's observers. */
+  get currentBatchPublication(): { readonly event: SessionEvent; readonly endSeq: SessionSeq } | undefined;
   /**
    * Append consecutive events together after validating every candidate and dispatch.
    * Observers read the complete committed log and final message surface.

packages/core/session/src/index.ts

@@ -514,6 +514,33 @@ export function pluginRecordOf(event: SessionEvent): PluginRecord | undefined {
   return { type, seq: event.seq, time: event.time, data: event.data }
 }
 
+/** Copy live append inputs before sequence assignment. */
+function snapshotAppend(
+  type: SessionEventType,
+  data: SessionEventMap[SessionEventType] | undefined,
+  opts: SurfaceIntent | undefined,
+  subject: string,
+) {
+  const surfaceOp = opts?.surfaceOp
+  const sourceEventSeqs = opts?.sourceEventSeqs
+  const dataSnapshot = snapshotJsonValue(data)
+  if (dataSnapshot === undefined) throw new Error(`${subject} carries non-JSON-serializable data`)
+  const surface = snapshotJsonValue({
+    ...sourceEventSeqs === undefined ? {} : { sourceEventSeqs },
+    ...surfaceOp === undefined ? {} : { surfaceOp },
+  })
+  if (surface === undefined) throw new Error(`${subject} carries non-JSON-serializable surface metadata`)
+  return { type, data: dataSnapshot, surface }
+}
+
+/** Construct and validate one live event from detached inputs. */
+function createAppendEvent(snapshot: ReturnType<typeof snapshotAppend>, seq: SessionSeq): SessionEvent {
+  const { type, data, surface } = snapshot
+  const event = deepFreeze({ type, seq, time: Date.now(), data, ...surface } as SessionEvent)
+  validateSessionEventData(event, `session event "${type}" at seq ${seq}`)
+  return event
+}
+
 /**
  * An event-sourced session: an append-only log of {@link SessionEvent}s.
  *
@@ -524,6 +551,7 @@ export function pluginRecordOf(event: SessionEvent): PluginRecord | undefined {
  */
 export class Session {
   private log: SessionEvent[] = []
+  private batchPublication: { readonly event: SessionEvent; readonly endSeq: SessionSeq } | undefined
   /** Single incremental owner of surface acceptance and projection state. */
   private readonly surfaceManager: SurfaceManager
 
@@ -761,6 +789,11 @@ export class Session {
     return SessionLogOffset(this.log.length)
   }
 
+  /** Current member and final sequence while synchronously notifying a batch's observers. */
+  get currentBatchPublication(): { readonly event: SessionEvent; readonly endSeq: SessionSeq } | undefined {
+    return this.batchPublication
+  }
+
   /**
    * Append consecutive events together after validating every candidate and dispatch.
    * Observers read the complete committed log and final message surface.
@@ -826,29 +859,10 @@ export class Session {
   ): SessionEvent | readonly SessionEvent[] {
     if (typeof type !== 'string') return this.#appendEvents(type)
     this.publicationEntry()
-    const surfaceOpts = opts[0]
-    const surfaceMetadata = {
-      ...surfaceOpts?.sourceEventSeqs === undefined ? {} : { sourceEventSeqs: surfaceOpts.sourceEventSeqs },
-      ...surfaceOpts?.surfaceOp === undefined ? {} : { surfaceOp: surfaceOpts.surfaceOp },
-    }
-    const dataSnapshot = snapshotJsonValue(data)
-    if (dataSnapshot === undefined) {
-      throw new Error(`session event "${type}" carries non-JSON-serializable data`)
-    }
-    const surfaceMetadataSnapshot = snapshotJsonValue(surfaceMetadata)
-    if (surfaceMetadataSnapshot === undefined) {
-      throw new Error(`session event "${type}" carries non-JSON-serializable surface metadata`)
-    }
+    const snapshot = snapshotAppend(type, data, opts[0], `session event "${type}"`)
     // Payload getters can detach the Session while the snapshots are read.
     const entry = this.publicationEntry()
-    const event = deepFreeze({
-      type,
-      seq: SessionSeq(this.log.length),
-      time: Date.now(),
-      data: dataSnapshot,
-      ...(surfaceMetadataSnapshot as { surfaceOp?: unknown; sourceEventSeqs?: unknown }),
-    } as SessionEvent)
-    validateSessionEventData(event, `session event "${type}" at seq ${event.seq}`)
+    const event = createAppendEvent(snapshot, SessionSeq(this.log.length))
     this.commit(event, entry)
     return event
   }
@@ -866,41 +880,32 @@ export class Session {
       const input = inputs.at(index)
       if (input === undefined) throw new Error(`session batch has an empty member at index ${index}`)
       const [type, data, surfaceOpts] = input
-      const snapshot = snapshotJsonValue({
-        type, data,
-        ...surfaceOpts?.surfaceOp === undefined ? {} : { surfaceOp: surfaceOpts.surfaceOp },
-        ...surfaceOpts?.sourceEventSeqs === undefined ? {} : { sourceEventSeqs: surfaceOpts.sourceEventSeqs },
-      })
-      if (snapshot === undefined) {
-        throw new Error(`session batch member ${index} ("${type}") carries non-JSON-serializable data`)
-      }
-      return snapshot
+      return snapshotAppend(type, data, surfaceOpts, `session batch member ${index} ("${type}")`)
     })
     // Payload getters may append to or detach the Session.
     const entry = this.publicationEntry()
     const startSeq = this.log.length
-    const events = snapshots.map((snapshot, index) => {
-      const event = { ...snapshot, seq: SessionSeq(startSeq + index), time: Date.now() }
-      assertSessionEventEnvelope(event, startSeq + index)
-      return deepFreeze(event)
-    })
+    const events = snapshots.map((snapshot, index) => createAppendEvent(snapshot, SessionSeq(startSeq + index)))
     const acceptSurface = this.surfaceManager.prepareBatch(events)
     if (entry !== undefined) entry.appending = true
     try {
       const publications = entry === undefined ? [] : events.map((event) => {
         const args: unknown[] = [this, event]
         const callbacks = collectSessionCallbacks(entry.emitCtx, [entry.carrier, 'session/event', ...args])
-        return { args, callbacks }
+        return { event, args, callbacks }
       })
       for (const event of events) this.log.push(event)
       this.eventsSnapshot = undefined
       acceptSurface()
       if (entry !== undefined) {
-        for (const { args, callbacks } of publications) {
+        const endSeq = SessionSeq(this.log.length - 1)
+        for (const { event, args, callbacks } of publications) {
+          this.batchPublication = { event, endSeq }
           invokeContainedSessionObservers(entry.emitCtx, 'session/event', entry.id, args, callbacks)
         }
       }
     } finally {
+      this.batchPublication = undefined
       if (entry !== undefined) {
         entry.appending = false
         if (entry.detachRequested) entry.detach()

packages/core/session/src/surface.ts

@@ -367,14 +367,13 @@ function assertSourceEventReferences(
 /** Resolve tool additions against their immutable historical request header. */
 function assertDeveloperHeader(
   event: SessionEvent,
-  events: readonly SessionEvent[],
-  baseSeq: SessionLogOffset,
+  read: (seq: number) => SessionEvent | undefined,
 ): void {
   if (event.type !== 'developer/message') return
   validateSessionEventData(event, `developer/message at seq ${event.seq}`)
   if (event.data.headerSeq === undefined) return
   const headerSeq = event.data.headerSeq
-  const headerEvent = events[headerSeq - baseSeq]
+  const headerEvent = read(headerSeq)
   if (headerSeq >= event.seq || headerEvent?.type !== 'request/header') {
     throw new Error('developer/message headerSeq must reference an earlier request/header')
   }
@@ -456,15 +455,14 @@ function isDeepEqualJson(a: unknown, b: unknown): boolean {
 function assertToolResultRewrite(
   event: SessionEvent,
   shadowedSeqs: readonly SessionSeq[],
-  events: readonly SessionEvent[],
-  baseSeq: SessionLogOffset,
+  read: (seq: number) => SessionEvent | undefined,
 ): void {
   if (event.type !== 'tool/result') return
   if (shadowedSeqs.length !== 1) {
     throw new Error('tool/result surface replacement must rewrite exactly one current node')
   }
   for (const originalSeq of shadowedSeqs) {
-    const original = events[originalSeq - baseSeq]
+    const original = read(originalSeq)
     if (original?.type !== 'tool/result') {
       throw new Error('tool/result surface replacement must target a current tool/result')
     }
@@ -495,11 +493,10 @@ function assertSystemHeadRewrite(
   state: SurfaceFoldState,
   startIdx: number,
   shadowedSeqs: readonly SessionSeq[],
-  events: readonly SessionEvent[],
-  baseSeq: SessionLogOffset,
+  read: (seq: number) => SessionEvent | undefined,
 ): void {
   if (startIdx !== 0) return
-  const head = events[state.nodes[0] as number - baseSeq]
+  const head = read(state.nodes[0] as number)
   if (head?.type !== 'system/message') return
   if (event.type !== 'system/message' || shadowedSeqs.length !== 1) {
     throw new Error('surface replace: node 0 holds the system prompt and may be rewritten only by a system/message over exactly that node')
@@ -514,16 +511,24 @@ function planSurfaceEvent(
   events: readonly SessionEvent[],
   baseSeq: SessionLogOffset,
   projections: readonly SessionMessageProjection[],
+  candidates?: readonly SessionEvent[],
 ): SurfacePlan | undefined {
   if (event.seq !== expectedSeq) {
     throw new Error(`session event seq ${event.seq} is not contiguous; expected ${expectedSeq}`)
   }
   const surfaceOp = validateSurfaceMetadata(event)
-  assertDeveloperHeader(event, events, baseSeq)
+  const read = (seq: number): SessionEvent | undefined => {
+    const index = seq - baseSeq
+    return index < events.length ? events[index] : candidates?.[index - events.length]
+  }
+  assertDeveloperHeader(event, read)
   const projection = projections.find(item => item.type === event.type)
   if (projection !== undefined) {
     return { kind: 'project', projection, messages: projection.project(event, {
-      nodes: state.nodes, events, baseSeq, messages: state.projectedMessages,
+      nodes: state.nodes,
+      get events() { return candidates === undefined ? events : [...events, ...candidates] },
+      baseSeq,
+      messages: state.projectedMessages,
     }) }
   }
   if (MESSAGE_PROJECTION_EVENT_TYPES.has(event.type)) {
@@ -535,8 +540,8 @@ function planSurfaceEvent(
   }
   const range = replacementRange(state, surfaceOp)
   assertSourceEventReferences(event, range.shadowedSeqs)
-  assertToolResultRewrite(event, range.shadowedSeqs, events, baseSeq)
-  assertSystemHeadRewrite(event, state, range.startIdx, range.shadowedSeqs, events, baseSeq)
+  assertToolResultRewrite(event, range.shadowedSeqs, read)
+  assertSystemHeadRewrite(event, state, range.startIdx, range.shadowedSeqs, read)
   return {
     kind: 'replace',
     seq: event.seq,
@@ -672,12 +677,15 @@ export class SurfaceManager implements SessionSurface {
       projectedMessages: new Map(this._state.projectedMessages),
       projections: new Set(this._state.projections),
     }
-    const log = [...this.log, ...events]
-    for (const [index, event] of events.entries()) {
-      applySurfaceEvent(state, event, SessionSeq(this.baseSeq + this.log.length + index), log, this.baseSeq, this.projections)
-    }
+    const plans = events.map((event, index) => {
+      const plan = planSurfaceEvent(
+        state, event, SessionSeq(this.baseSeq + this.log.length + index), this.log, this.baseSeq, this.projections, events,
+      )
+      applySurfacePlan(state, plan)
+      return plan
+    })
     return () => {
-      this._state = state
+      for (const plan of plans) applySurfacePlan(this._state, plan)
       this._lastProcessedSeq = lastProcessedSeq
       this._pendingPlan = undefined
     }

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

@@ -54,6 +54,7 @@ describe('multiple Session appends', () => {
     const observed: number[] = []
     ctx.on('session/event', (current, event) => {
       if (current !== session) return
+      expect(current.currentBatchPublication).toEqual({ event, endSeq: 1 })
       expect(current.seq).toBe(2)
       expect(current.surface.nodes).toEqual([1])
       expect(current.deriveMessages()).toEqual([messageValue])
@@ -65,16 +66,18 @@ describe('multiple Session appends', () => {
       ['user/message', messageValue, { surfaceOp: 'append' }],
     ])
     expect(observed).toEqual([0, 1])
+    expect(session.currentBatchPublication).toBeUndefined()
   })
 
   it('replaces an earlier member in the same append', () => {
     const session = Session.create(SessionId('replace-append'))
+    const nodes = session.surface.nodes
     const replacement = message('replacement')
     session.append([
       ['user/message', message('original'), { surfaceOp: 'append' }],
       ['user/message', replacement, { surfaceOp: { op: 'replace', startSeq: SessionSeq(0), endSeq: SessionSeq(0) }, sourceEventSeqs: [SessionSeq(0)] }],
     ])
-    expect(session.surface.nodes).toEqual([1])
+    expect(nodes).toEqual([1])
     expect(session.deriveMessages()).toEqual([replacement])
   })
 

packages/extensions/tool-cordis/src/api-catalog.ts

@@ -8539,7 +8539,7 @@ export const TYPE_API: readonly TypeApiEntry[] = [
   },
   {
     name: 'Session',
-    declaration: 'export class Session {\n    get surface(): SessionSurface;\n    readonly header: SessionHeader;\n    readonly inheritedEventCount: SessionLogOffset;\n    get id(): SessionId;\n    readonly firstLiveSeq: SessionLogOffset;\n    readonly firstLifecycleSeq: SessionLogOffset;\n    static create(id: SessionId, seed?: readonly SessionEvent[], header?: SessionHeader, inheritedEventCount?: SessionLogOffset, projections?: readonly SessionMessageProjection[]): Session;\n    static fromRestore(id: SessionId, seed: readonly SessionEvent[], header: SessionHeader, inheritedEventCount: SessionLogOffset, eventState: SessionSeedEventState, projections?: readonly SessionMessageProjection[]): Session;\n    eventAt(seq: SessionSeq): SessionEvent | undefined;\n    snapshotEvents(fromSeq: SessionLogOffset = SessionLogOffset(0), toSeqExclusive: SessionLogOffset = this.seq): readonly SessionEvent[];\n    ownEvents(): readonly SessionEvent[];\n    isOwnSeq(seq: SessionSeq): boolean;\n    get seq(): SessionLogOffset;\n    append(inputs: readonly SessionEventInput[]): readonly SessionEvent[];\n    append<T extends SessionEventType>(type: T, data: SessionEventMap[T], ...opts: T extends SurfaceEventType ? [\n        opts: SurfaceIntent<T>\n    ] : [\n    ]): SessionEvent<T>;\n    append(type: SessionEventType | readonly SessionEventInput[], data?: SessionEventMap[SessionEventType], ...opts: [\n        opts?: SurfaceIntent\n    ]): SessionEvent | readonly SessionEvent[];\n    requestHeader(): EpochHeader | undefined;\n    reques /* …truncated — full shape in source */',
+    declaration: 'export class Session {\n    get surface(): SessionSurface;\n    readonly header: SessionHeader;\n    readonly inheritedEventCount: SessionLogOffset;\n    get id(): SessionId;\n    readonly firstLiveSeq: SessionLogOffset;\n    readonly firstLifecycleSeq: SessionLogOffset;\n    static create(id: SessionId, seed?: readonly SessionEvent[], header?: SessionHeader, inheritedEventCount?: SessionLogOffset, projections?: readonly SessionMessageProjection[]): Session;\n    static fromRestore(id: SessionId, seed: readonly SessionEvent[], header: SessionHeader, inheritedEventCount: SessionLogOffset, eventState: SessionSeedEventState, projections?: readonly SessionMessageProjection[]): Session;\n    eventAt(seq: SessionSeq): SessionEvent | undefined;\n    snapshotEvents(fromSeq: SessionLogOffset = SessionLogOffset(0), toSeqExclusive: SessionLogOffset = this.seq): readonly SessionEvent[];\n    ownEvents(): readonly SessionEvent[];\n    isOwnSeq(seq: SessionSeq): boolean;\n    get seq(): SessionLogOffset;\n    get currentBatchPublication(): {\n        readonly event: SessionEvent;\n        readonly endSeq: SessionSeq;\n    } | undefined;\n    append(inputs: readonly SessionEventInput[]): readonly SessionEvent[];\n    append<T extends SessionEventType>(type: T, data: SessionEventMap[T], ...opts: T extends SurfaceEventType ? [\n        opts: SurfaceIntent<T>\n    ] : [\n    ]): SessionEvent<T>;\n    append(type: SessionEventType | readonly SessionEventInput[], data?: SessionEventMap[SessionEventType], ...opts: [\n   /* …truncated — full shape in source */',
   },
   {
     name: 'SessionAccess',