From 143fabbc34877aa4570d13ea524273cd2bdf3e0f Mon Sep 17 00:00:00 2001 From: liukx0205 Date: Fri, 6 Nov 2026 15:11:01 +0800 Subject: [PATCH] fix(session): share live append validation and expose batch publication --- docs/subsystems/session.md | 2 + docs/subsystems/session.zh.md | 2 + packages/core/session/src/index.ts | 79 ++++++++++--------- packages/core/session/src/surface.ts | 44 ++++++----- packages/core/session/tests/append.spec.ts | 5 +- .../extensions/tool-cordis/src/api-catalog.ts | 2 +- 6 files changed, 77 insertions(+), 57 deletions(-) diff --git a/docs/subsystems/session.md b/docs/subsystems/session.md index d33640e9ba..3cc0812474 100644 --- a/docs/subsystems/session.md +++ b/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. diff --git a/docs/subsystems/session.zh.md b/docs/subsystems/session.zh.md index 1c9a931fd9..d178a41690 100644 --- a/docs/subsystems/session.zh.md +++ b/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. diff --git a/packages/core/session/src/index.ts b/packages/core/session/src/index.ts index c573d40f60..082d3159e4 100644 --- a/packages/core/session/src/index.ts +++ b/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, 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() diff --git a/packages/core/session/src/surface.ts b/packages/core/session/src/surface.ts index a3d89e40bf..b94935d9ad 100644 --- a/packages/core/session/src/surface.ts +++ b/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 } diff --git a/packages/core/session/tests/append.spec.ts b/packages/core/session/tests/append.spec.ts index 9d047b1660..bddd56434f 100644 --- a/packages/core/session/tests/append.spec.ts +++ b/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]) }) diff --git a/packages/extensions/tool-cordis/src/api-catalog.ts b/packages/extensions/tool-cordis/src/api-catalog.ts index 03d0c00d8f..ea129e4641 100644 --- a/packages/extensions/tool-cordis/src/api-catalog.ts +++ b/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(type: T, data: SessionEventMap[T], ...opts: T extends SurfaceEventType ? [\n opts: SurfaceIntent\n ] : [\n ]): SessionEvent;\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(type: T, data: SessionEventMap[T], ...opts: T extends SurfaceEventType ? [\n opts: SurfaceIntent\n ] : [\n ]): SessionEvent;\n append(type: SessionEventType | readonly SessionEventInput[], data?: SessionEventMap[SessionEventType], ...opts: [\n /* …truncated — full shape in source */', }, { name: 'SessionAccess',