From b5969f9e8d03a0abe080ec739f2fa2805d9f2223 Mon Sep 17 00:00:00 2001 From: liukx0205 Date: Thu, 5 Nov 2026 13:52:57 +0800 Subject: [PATCH] feat(session): accept multiple events through append --- docs/persistence-catalog.i18n.yaml | 4 +- docs/persistence-catalog.md | 2 +- docs/persistence-catalog.zh.md | 2 +- docs/subsystems/session.i18n.yaml | 7 +- docs/subsystems/session.md | 26 ++++ docs/subsystems/session.zh.md | 26 ++++ .../headless/tests/terminal-failures.spec.ts | 16 +- packages/core/session/README.i18n.yaml | 4 +- packages/core/session/README.md | 2 + packages/core/session/README.zh.md | 2 + packages/core/session/src/index.ts | 67 +++++++- packages/core/session/src/surface.ts | 38 +++++ packages/core/session/src/types.ts | 6 + packages/core/session/tests/append.spec.ts | 144 ++++++++++++++++++ .../core/session/tests/append.type-test.ts | 24 +++ .../session/tests/canonical-envelopes.spec.ts | 7 +- .../session/tests/message-projections.spec.ts | 18 +++ .../extensions/tool-cordis/src/api-catalog.ts | 6 +- scripts/type-equiv.manifest.json | 5 + 19 files changed, 380 insertions(+), 26 deletions(-) create mode 100644 packages/core/session/tests/append.spec.ts create mode 100644 packages/core/session/tests/append.type-test.ts diff --git a/docs/persistence-catalog.i18n.yaml b/docs/persistence-catalog.i18n.yaml index a5fb682b48..33dd355517 100644 --- a/docs/persistence-catalog.i18n.yaml +++ b/docs/persistence-catalog.i18n.yaml @@ -12,8 +12,8 @@ en: 35ffb6107a5fbae2 zh: 06ce2128075e157b /session-persistence-event-catalog/event-envelope: - en: 56414ffd00cdc32a - zh: 6dc11eb013e95e41 + en: 28dce8bfdddd2e1f + zh: 6a6440a91255edd1 /session-persistence-event-catalog/events: en: f4776ba3d0fd53ce zh: e86769f50a69055a diff --git a/docs/persistence-catalog.md b/docs/persistence-catalog.md index 6b80fbbc99..77912093d0 100644 --- a/docs/persistence-catalog.md +++ b/docs/persistence-catalog.md @@ -161,7 +161,7 @@ export type SessionEvent = { }[T] ``` -Sources: [`packages/core/session/src/types.ts:439`](../packages/core/session/src/types.ts) · [`packages/core/session/src/types.ts:447`](../packages/core/session/src/types.ts) · [`packages/core/session/src/types.ts:470`](../packages/core/session/src/types.ts) · [`packages/core/session/src/types.ts:501`](../packages/core/session/src/types.ts) +Sources: [`packages/core/session/src/types.ts:439`](../packages/core/session/src/types.ts) · [`packages/core/session/src/types.ts:447`](../packages/core/session/src/types.ts) · [`packages/core/session/src/types.ts:470`](../packages/core/session/src/types.ts) · [`packages/core/session/src/types.ts:507`](../packages/core/session/src/types.ts) ## Events diff --git a/docs/persistence-catalog.zh.md b/docs/persistence-catalog.zh.md index c73b594c28..5baac35c0c 100644 --- a/docs/persistence-catalog.zh.md +++ b/docs/persistence-catalog.zh.md @@ -163,7 +163,7 @@ export type SessionEvent = { }[T] ``` -来源：[`packages/core/session/src/types.ts:439`](../packages/core/session/src/types.ts) · [`packages/core/session/src/types.ts:447`](../packages/core/session/src/types.ts) · [`packages/core/session/src/types.ts:470`](../packages/core/session/src/types.ts) · [`packages/core/session/src/types.ts:501`](../packages/core/session/src/types.ts) +来源：[`packages/core/session/src/types.ts:439`](../packages/core/session/src/types.ts) · [`packages/core/session/src/types.ts:447`](../packages/core/session/src/types.ts) · [`packages/core/session/src/types.ts:470`](../packages/core/session/src/types.ts) · [`packages/core/session/src/types.ts:507`](../packages/core/session/src/types.ts) ## 事件 diff --git a/docs/subsystems/session.i18n.yaml b/docs/subsystems/session.i18n.yaml index b4cd4bba34..f2e38078da 100644 --- a/docs/subsystems/session.i18n.yaml +++ b/docs/subsystems/session.i18n.yaml @@ -27,8 +27,11 @@ en: 8fcc967516a13692 zh: d3978d3b8d02c566 /sessions/surface-types/surfaceintent-the-parameter-to-session-append: - en: c65cc976acad9005 - zh: 3efd10779881cba5 + en: ebdf0684eccf4f86 + zh: e5f545c060bfb824 +/sessions/surface-types/sessioneventinput-one-proposed-event: + en: 58f1d60b60d7fe99 + zh: a162d6d0cb15c6a1 /sessions/surface-types/plugin-owned-message-projections: en: f1f73607d650620f zh: b3ef72f19fb5ee3d diff --git a/docs/subsystems/session.md b/docs/subsystems/session.md index 7a0ad908d9..cb2a840f2a 100644 --- a/docs/subsystems/session.md +++ b/docs/subsystems/session.md @@ -361,6 +361,18 @@ Required for `SurfaceEventType` events — every message-producing event must de `assistant/message` cannot carry `sourceEventSeqs`; its `stream` owns exact provider evidence. Other surface events omit the field when they cite no earlier event and use a complete non-empty list when they do. +### `SessionEventInput` — one proposed event + +```ts type-equiv +/** One event proposed for append, without assigned sequence or time. */ +type SessionEventInput = { + [K in SessionEventType]: { type: K; data: SessionEventMap[K] } + & (K extends SurfaceEventType ? SurfaceIntent : { surfaceOp?: never; sourceEventSeqs?: never }) +}[SessionEventType] +``` + +Each array member keeps its event type paired with its payload and requires surface intent only for message-producing events. + ### Plugin-owned message projections @@ -573,6 +585,15 @@ 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; + /** + * Append consecutive events together after validating every candidate and dispatch. + * Observers read the complete committed log and final message surface. An empty + * array appends nothing. Persistence listeners retain their own durability rules. + * @param inputs - event payloads and surface intents in commit order. + * @returns immutable committed events in the same order. + * @throws before changing the log if any candidate or synchronous dispatch validation fails. + */ + append(inputs: readonly SessionEventInput[]): readonly SessionEvent[]; /** * Append one typed event to the log and synchronously notify observers via * the store-owned, module-private publication hooks. The hot path never blocks @@ -620,6 +641,11 @@ declare class Session { data: SessionEventMap[T], ...opts: T extends SurfaceEventType ? [opts: SurfaceIntent] : [] ): SessionEvent; + append( + type: T | readonly SessionEventInput[], + data?: SessionEventMap[T], + ...opts: [opts?: SurfaceIntent] + ): SessionEvent | readonly SessionEvent[]; /** * The {@link EpochHeader} in force after the log's last header event — the * header the NEXT request will be compared against — or undefined before diff --git a/docs/subsystems/session.zh.md b/docs/subsystems/session.zh.md index ec9fab642f..c5d7cbaf68 100644 --- a/docs/subsystems/session.zh.md +++ b/docs/subsystems/session.zh.md @@ -363,6 +363,18 @@ type SurfaceIntent = { `assistant/message` 不能携带 `sourceEventSeqs`；它的 `stream` 拥有精确 provider 证据。其他 surface event 不引用较早 event 时省略该字段，需要引用时使用完整非空 list。 +### `SessionEventInput`：一个候选事件 + +```ts type-equiv +/** One event proposed for append, without assigned sequence or time. */ +type SessionEventInput = { + [K in SessionEventType]: { type: K; data: SessionEventMap[K] } + & (K extends SurfaceEventType ? SurfaceIntent : { surfaceOp?: never; sourceEventSeqs?: never }) +}[SessionEventType] +``` + +数组中的每个成员将事件类型与对应载荷配对，且仅对产生消息的事件要求消息视图意图。 + ### 插件拥有的消息投影 @@ -575,6 +587,15 @@ 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; + /** + * Append consecutive events together after validating every candidate and dispatch. + * Observers read the complete committed log and final message surface. An empty + * array appends nothing. Persistence listeners retain their own durability rules. + * @param inputs - event payloads and surface intents in commit order. + * @returns immutable committed events in the same order. + * @throws before changing the log if any candidate or synchronous dispatch validation fails. + */ + append(inputs: readonly SessionEventInput[]): readonly SessionEvent[]; /** * Append one typed event to the log and synchronously notify observers via * the store-owned, module-private publication hooks. The hot path never blocks @@ -622,6 +643,11 @@ declare class Session { data: SessionEventMap[T], ...opts: T extends SurfaceEventType ? [opts: SurfaceIntent] : [] ): SessionEvent; + append( + type: T | readonly SessionEventInput[], + data?: SessionEventMap[T], + ...opts: [opts?: SurfaceIntent] + ): SessionEvent | readonly SessionEvent[]; /** * The {@link EpochHeader} in force after the log's last header event — the * header the NEXT request will be compared against — or undefined before diff --git a/packages/bundle/headless/tests/terminal-failures.spec.ts b/packages/bundle/headless/tests/terminal-failures.spec.ts index c92f095334..959eebb401 100644 --- a/packages/bundle/headless/tests/terminal-failures.spec.ts +++ b/packages/bundle/headless/tests/terminal-failures.spec.ts @@ -7,7 +7,7 @@ import AgentDefaultModelConfig from '@deepseek-ai/dsh-agent-default-model' import AgentLoop from '@deepseek-ai/dsh-agent-loop' import { mountAgentLoopTestDependencies } from '@deepseek-ai/dsh-agent-loop-testkit' import { createUserMessage } from '@deepseek-ai/dsh-llm' -import { SessionId } from '@deepseek-ai/dsh-session' +import { SessionId, type SessionEvent } from '@deepseek-ai/dsh-session' import { MockAdapter, textResponse } from '../../../core/agent-loop/tests/mock-adapter.ts' import { apply } from '../src/index.ts' import { internals } from '../src/runner-internals.ts' @@ -62,11 +62,10 @@ describe('headless terminal failures', () => { test.ctx.on('agent/error', ({ error }) => { failures.push(error) }) const waitForTreeWork = vi.fn().mockImplementationOnce((parent: Agent) => { const append = parent.session.append.bind(parent.session) - const failAppend: typeof parent.session.append = (type, data, ...intent) => { + vi.spyOn(parent.session, 'append').mockImplementation((type, data, ...intent) => { if (type === failedEvent) throw new Error(`summary ${failedEvent} unavailable`) - return append(type, data, ...intent) - } - vi.spyOn(parent.session, 'append').mockImplementation(failAppend) + return Reflect.apply(append, undefined, [type, data, ...intent]) as SessionEvent + }) followup(parent) return Promise.resolve(true) }).mockResolvedValue(false) @@ -110,14 +109,13 @@ describe('headless terminal failures', () => { const waitForTreeWork = vi.fn().mockImplementationOnce((parent: Agent) => { const append = parent.session.append.bind(parent.session) let failed = false - const failAppend: typeof parent.session.append = (type, data, ...intent) => { + vi.spyOn(parent.session, 'append').mockImplementation((type, data, ...intent) => { if (!failed && type === failedEvent) { failed = true throw new Error(`summary ${failedEvent} unavailable`) } - return append(type, data, ...intent) - } - vi.spyOn(parent.session, 'append').mockImplementation(failAppend) + return Reflect.apply(append, undefined, [type, data, ...intent]) as SessionEvent + }) followup(parent) return Promise.resolve(true) }).mockImplementationOnce((parent: Agent) => { diff --git a/packages/core/session/README.i18n.yaml b/packages/core/session/README.i18n.yaml index 376a117359..2b551b5213 100644 --- a/packages/core/session/README.i18n.yaml +++ b/packages/core/session/README.i18n.yaml @@ -30,8 +30,8 @@ en: 002b8240caa0c57c zh: e0954dd62913a944 /deepseek-ai-dsh-session/use-this-package/fork-a-session: - en: 1c40d17a786059d3 - zh: 81320ce118977c65 + en: 93def7349171bf4f + zh: 57f7d30f3c632a75 /deepseek-ai-dsh-session/use-this-package/flush-durable-state: en: 70a7673145c851ad zh: 0dd1667d9518993d diff --git a/packages/core/session/README.md b/packages/core/session/README.md index 0f368e9ead..301176d152 100644 --- a/packages/core/session/README.md +++ b/packages/core/session/README.md @@ -83,6 +83,8 @@ The [experimental persistence catalog](../../../docs/experimental-persistence-ca The logical `SessionHeader.isSeeded` field reports whether fork history exists without exposing a positional integer. `Session.inheritedEventCount` retains the exact checked `SessionLogOffset`; `ownEvents()` returns events at and after that cut, and `isOwnSeq(seq)` accepts only an existing child-owned position. A low-level seeded constructor must supply an explicit `seed` and `inheritedEventCount` because the constructor seed can contain child-owned setup events after the inherited prefix. +`session.append(type, data, ...opts)` appends one event and returns that event. `session.append(inputs)` accepts a readonly array of `SessionEventInput` and returns the committed events in input order; an empty array appends nothing. Array admission validates every candidate and synchronous dispatch before changing the log or message surface, so observers read the complete committed group. Persistence listeners retain their own durability rules; array admission alone does not guarantee crash-atomic storage. + ### Flush durable state `ctx.sessions.flush(session)` dispatches the awaited durability checkpoint: every persistence listener flushes and the call settles after all of them. A producer that needs an immediate durability barrier awaits it instead of assuming the write-behind drained. diff --git a/packages/core/session/README.zh.md b/packages/core/session/README.zh.md index 9b4a192d81..84c6988137 100644 --- a/packages/core/session/README.zh.md +++ b/packages/core/session/README.zh.md @@ -83,6 +83,8 @@ session.deriveMessages() // the derived model history 逻辑 `SessionHeader.isSeeded` 字段报告是否存在 fork 历史，而不公开位置整数。`Session.inheritedEventCount` 保留经过校验的精确 `SessionLogOffset`；`ownEvents()` 返回从该切点开始的事件，`isOwnSeq(seq)` 只接受已存在且由子会话拥有的位置。底层带 seed 构造必须显式提供 `seed` 与 `inheritedEventCount`，因为构造 seed 可以在继承前缀之后包含子会话自有的设置事件。 +`session.append(type, data, ...opts)` 追加一个事件并返回该事件。`session.append(inputs)` 接受 `SessionEventInput` 的只读数组，并按输入顺序返回已提交事件；空数组不追加任何内容。数组准入会在改变日志或消息视图前验证所有候选和同步分发，因此观察者读取完整的已提交组。持久化监听器保留各自的持久性规则；仅数组准入不保证进程崩溃时整组存储的原子性。 + ### 刷新持久状态 `ctx.sessions.flush(session)` 分发需等待完成的持久性检查点：每个持久化监听器都会刷新，调用在所有监听器结算后完成。需要立即持久性屏障的生产方应等待它，而不是假定延后写入已经排空。 diff --git a/packages/core/session/src/index.ts b/packages/core/session/src/index.ts index e9baee1b82..7a3c6e6040 100644 --- a/packages/core/session/src/index.ts +++ b/packages/core/session/src/index.ts @@ -15,7 +15,7 @@ import type { Scoped } from '@deepseek-ai/dsh-scope' import type { Message } from '@deepseek-ai/dsh-llm' import { SESSION_FORMAT_VERSION, SessionLogOffset, SessionSeq } from './types.ts' import type { TypertLookup } from '@deepseek-ai/dsh-typert-protocol' -import type { CreateSessionOptions, EpochHeader, PluginRecord, PluginRecordMap, PluginRecordType, PrepareSessionOptions, RequestContext, SessionEvent, SessionEventMap, SessionEventType, SessionHeader, SessionId, SessionSeedEventState, SurfaceIntent, SurfaceEventType } from './types.ts' +import type { CreateSessionOptions, EpochHeader, PluginRecord, PluginRecordMap, PluginRecordType, PrepareSessionOptions, RequestContext, SessionEvent, SessionEventInput, SessionEventMap, SessionEventType, SessionHeader, SessionId, SessionSeedEventState, SurfaceIntent, SurfaceEventType } from './types.ts' import { SurfaceManager, validateSessionEventData, validateSurfaceMetadata } from './surface.ts' import type { SessionSurface, SessionMessageProjection } from './surface.ts' import { foldRequestHeader } from './request-header.ts' @@ -761,6 +761,16 @@ export class Session { return SessionLogOffset(this.log.length) } + /** + * Append consecutive events together after validating every candidate and dispatch. + * Observers read the complete committed log and final message surface. An empty + * array appends nothing. Persistence listeners retain their own durability rules. + * @param inputs - event payloads and surface intents in commit order. + * @returns immutable committed events in the same order. + * @throws before changing the log if any candidate or synchronous dispatch validation fails. + */ + append(inputs: readonly SessionEventInput[]): readonly SessionEvent[] + /** * Append one typed event to the log and synchronously notify observers via * the store-owned, module-private publication hooks. The hot path never blocks @@ -807,9 +817,15 @@ export class Session { type: T, data: SessionEventMap[T], ...opts: T extends SurfaceEventType ? [opts: SurfaceIntent] : [] - ): SessionEvent { + ): SessionEvent + append( + type: T | readonly SessionEventInput[], + data?: SessionEventMap[T], + ...opts: [opts?: SurfaceIntent] + ): SessionEvent | readonly SessionEvent[] { + if (typeof type !== 'string') return this.#appendEvents(type) this.publicationEntry() - const surfaceOpts: SurfaceIntent | undefined = opts[0] + const surfaceOpts = opts[0] const surfaceMetadata = { ...surfaceOpts?.sourceEventSeqs === undefined ? {} : { sourceEventSeqs: surfaceOpts.sourceEventSeqs }, ...surfaceOpts?.surfaceOp === undefined ? {} : { surfaceOp: surfaceOpts.surfaceOp }, @@ -836,6 +852,51 @@ export class Session { return event } + /** + * Validate and publish consecutive events together. Observers see the complete log. + * @param inputs - event payloads and surface intents in commit order. + * @returns immutable committed events in the same order. + * @throws before changing the log if any candidate or synchronous dispatch validation fails. + */ + #appendEvents(inputs: readonly SessionEventInput[]): readonly SessionEvent[] { + this.publicationEntry() + if (inputs.length === 0) return [] + const startSeq = this.log.length + const events = inputs.map((input, index) => { + const snapshot = snapshotJsonValue(input) + if (snapshot === undefined) throw new Error('session batch carries non-JSON-serializable data') + const event = { + ...snapshot, seq: SessionSeq(startSeq + index), time: Date.now(), + } + assertSessionEventEnvelope(event, startSeq + index) + return deepFreeze(event) + }) + const entry = this.publicationEntry() + 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 } + }) + for (const event of events) this.log.push(event) + this.eventsSnapshot = undefined + acceptSurface() + if (entry !== undefined) { + for (const { args, callbacks } of publications) { + invokeContainedSessionObservers(entry.emitCtx, 'session/event', entry.id, args, callbacks) + } + } + } finally { + if (entry !== undefined) { + entry.appending = false + if (entry.detachRequested) entry.detach() + } + } + return events + } + static { commitPluginRecord = (session, type, data) => session.#appendRecord(type, data) } diff --git a/packages/core/session/src/surface.ts b/packages/core/session/src/surface.ts index 0bb84c030e..a3d89e40bf 100644 --- a/packages/core/session/src/surface.ts +++ b/packages/core/session/src/surface.ts @@ -645,6 +645,44 @@ export class SurfaceManager implements SessionSurface { } } + /** + * Validate consecutive candidates against an isolated surface fold. + * @param events - candidates immediately following the committed log. + * @returns a callback adopting the validated fold after all candidates enter the log. + */ + prepareBatch(events: readonly SessionEvent[]): () => void { + this._assertProjections() + if (this._lastProcessedSeq < this.baseSeq + this.log.length - 1) this._processDelta() + const lastProcessedSeq = SessionSeq(this.baseSeq + this.log.length + events.length - 1) + if (events.every(event => event.type !== 'developer/message' + && (event.surfaceOp === undefined || event.surfaceOp === 'append') + && !this.projections.some(projection => projection.type === event.type))) { + const plans = events.map((event, index) => planSurfaceEvent( + this._state, event, SessionSeq(this.baseSeq + this.log.length + index), this.log, this.baseSeq, this.projections, + )) + return () => { + for (const plan of plans) applySurfacePlan(this._state, plan) + this._lastProcessedSeq = lastProcessedSeq + this._pendingPlan = undefined + } + } + const state: SurfaceFoldState = { + ...this._state, + nodes: [...this._state.nodes], + 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) + } + return () => { + this._state = state + this._lastProcessedSeq = lastProcessedSeq + this._pendingPlan = undefined + } + } + /** Monotonic count of folded positional replacements. */ get replaceGeneration(): number { this._assertProjections() diff --git a/packages/core/session/src/types.ts b/packages/core/session/src/types.ts index 7787dd9bd0..28594de0ef 100644 --- a/packages/core/session/src/types.ts +++ b/packages/core/session/src/types.ts @@ -485,6 +485,12 @@ export type SurfaceIntent = { sourceEventSeqs?: SessionSeq[] }) +/** One event proposed for append, without assigned sequence or time. */ +export type SessionEventInput = { + [K in SessionEventType]: { type: K; data: SessionEventMap[K] } + & (K extends SurfaceEventType ? SurfaceIntent : { surfaceOp?: never; sourceEventSeqs?: never }) +}[SessionEventType] + /** * One immutable entry in the session log. * diff --git a/packages/core/session/tests/append.spec.ts b/packages/core/session/tests/append.spec.ts new file mode 100644 index 0000000000..f6354e47d3 --- /dev/null +++ b/packages/core/session/tests/append.spec.ts @@ -0,0 +1,144 @@ +import { describe, expect, it, onTestFinished } from 'vitest' +import { Context } from '@deepseek-ai/cordis' +import { createUserMessage } from '@deepseek-ai/dsh-llm' +import SessionStore, { Session, SessionId, SessionSeq } from '@deepseek-ai/dsh-session' + +function message(text: string) { + return createUserMessage({ content: [{ type: 'text', text }], source: { kind: 'user' } }) +} + +describe('multiple Session appends', () => { + it('leaves the log unchanged for empty input and accepts a single array member', () => { + const session = Session.create(SessionId('empty-append')) + expect(session.append([])).toEqual([]) + expect(session.seq).toBe(0) + const events = session.append([{ type: 'turn/start', data: { turn: 1 } }]) + expect(events).toHaveLength(1) + expect(events[0]).toMatchObject({ type: 'turn/start', seq: 0, data: { turn: 1 } }) + }) + + it('publishes each member after the complete log and surface are committed', async () => { + const ctx = new Context() + onTestFinished(async () => { await ctx.fiber.dispose() }) + await ctx.plugin(SessionStore) + const session = ctx.sessions.create(SessionId('observed-append')) + const observed: number[] = [] + ctx.on('session/event', (current, event) => { + if (current !== session) return + expect(current.seq).toBe(2) + expect(current.surface.nodes).toEqual([1]) + expect(current.deriveMessages()).toEqual([messageValue]) + observed.push(event.seq) + }) + const messageValue = message('visible') + session.append([ + { type: 'turn/start', data: { turn: 1 } }, + { type: 'user/message', data: messageValue, surfaceOp: 'append' }, + ]) + expect(observed).toEqual([0, 1]) + }) + + it('replaces an earlier member in the same append', () => { + const session = Session.create(SessionId('replace-append')) + const replacement = message('replacement') + session.append([ + { type: 'user/message', data: message('original'), surfaceOp: 'append' }, + { + type: 'user/message', data: replacement, + surfaceOp: { op: 'replace', startSeq: SessionSeq(0), endSeq: SessionSeq(0) }, + sourceEventSeqs: [SessionSeq(0)], + }, + ]) + expect(session.surface.nodes).toEqual([1]) + expect(session.deriveMessages()).toEqual([replacement]) + }) + + it('rejects non-JSON data before publishing any member', () => { + const session = Session.create(SessionId('non-json-append')) + expect(() => session.append([ + { type: 'turn/start', data: { turn: 1 } }, + { type: 'turn/start', data: { turn: Number.NaN } }, + ])).toThrow(/non-JSON/) + expect(session.seq).toBe(0) + }) + + it('detaches only after every member has been observed', async () => { + const ctx = new Context() + onTestFinished(async () => { await ctx.fiber.dispose() }) + await ctx.plugin(SessionStore) + const session = ctx.sessions.prepare(SessionId('detached-append')) + const detach = ctx.sessions.enter(session) + ctx.sessions.announce(session) + const observed: number[] = [] + ctx.on('session/event', (current, event) => { + expect(ctx.sessions.get(current.id)).toBe(current) + detach() + observed.push(event.seq) + }) + session.append([ + { type: 'turn/start', data: { turn: 1 } }, + { type: 'step/start', data: { turn: 1, step: 1 } }, + ]) + expect(observed).toEqual([0, 1]) + expect(ctx.sessions.get(session.id)).toBeUndefined() + }) + + it('folds preceding single appends before admitting an array', () => { + const session = Session.create(SessionId('mixed-append')) + session.append('user/message', message('before'), { surfaceOp: 'append' }) + session.append([{ type: 'user/message', data: message('after'), surfaceOp: 'append' }]) + expect(session.surface.nodes).toEqual([0, 1]) + expect(session.surface.replaceGeneration).toBe(0) + }) + + it('publishes a complete group with its final surface', () => { + const session = Session.create(SessionId('batch')) + const events = session.append([ + { type: 'turn/start', data: { turn: 1 } }, + { type: 'user/message', data: message('hello'), surfaceOp: 'append' }, + ]) + expect(events.map(event => event.seq)).toEqual([0, 1]) + expect(session.surface.nodes).toEqual([1]) + expect(session.deriveMessages()).toEqual([events[1]?.data]) + }) + + it('retains the committed log and surface when a later candidate is invalid', () => { + const session = Session.create(SessionId('rejected-batch')) + session.append('user/message', message('before'), { surfaceOp: 'append' }) + const before = session.deriveMessages() + expect(() => session.append([ + { type: 'user/message', data: message('candidate'), surfaceOp: 'append' }, + { type: 'user/message', data: message('invalid'), surfaceOp: { op: 'replace', startSeq: SessionSeq(40), endSeq: SessionSeq(41) } }, + ])).toThrow() + expect(session.seq).toBe(1) + expect(session.deriveMessages()).toEqual(before) + expect(session.surface.nodes).toEqual([0]) + }) + + it('rejects the entire group when dispatch validation vetoes its last event', async () => { + const ctx = new Context() + onTestFinished(async () => { await ctx.fiber.dispose() }) + await ctx.plugin(SessionStore) + const session = ctx.sessions.create(SessionId('veto-batch')) + const observed: number[] = [] + ctx.on('session/event', (current, event) => { + if (current === session) observed.push(event.seq) + }) + const stop = ctx.on('internal/dispatch', (_mode, name, args) => { + if (name !== 'session/event') return + const [current, event] = args as [Session, { seq: number }] + if (current === session && event.seq === 1) throw new Error('veto final event') + }) + expect(() => session.append([ + { type: 'turn/start', data: { turn: 1 } }, + { type: 'user/message', data: message('candidate'), surfaceOp: 'append' }, + ])).toThrow('veto final event') + expect(session.seq).toBe(0) + expect(session.surface.nodes).toEqual([]) + expect(observed).toEqual([]) + stop() + session.append('turn/start', { turn: 1 }) + expect(observed).toEqual([0]) + }) + +}) diff --git a/packages/core/session/tests/append.type-test.ts b/packages/core/session/tests/append.type-test.ts new file mode 100644 index 0000000000..728757b7e6 --- /dev/null +++ b/packages/core/session/tests/append.type-test.ts @@ -0,0 +1,24 @@ +/** Compile-time checks for single and multiple event admission. */ +import { expectTypeOf } from 'vitest' +import type { Session, SessionEvent, SessionEventInput } from '@deepseek-ai/dsh-session' +import type { UserMessage } from '@deepseek-ai/dsh-llm' + +/** + * Check result cardinality and event-specific payload and surface requirements. + * @param session - target Session; this function is typechecked without execution. + * @param message - user input with durable identity. + */ +export function checkAppendTypes(session: Session, message: UserMessage): void { + expectTypeOf(session.append('turn/start', { turn: 1 })).toEqualTypeOf>() + const inputs: readonly SessionEventInput[] = [ + { type: 'turn/start', data: { turn: 1 } }, + { type: 'user/message', data: message, surfaceOp: 'append' }, + ] + expectTypeOf(session.append(inputs)).toEqualTypeOf() + // @ts-expect-error -- message-producing members require a surface operation. + session.append([{ type: 'user/message', data: message }]) + // @ts-expect-error -- log-only members cannot carry a surface operation. + session.append([{ type: 'turn/start', data: { turn: 1 }, surfaceOp: 'append' }]) + // @ts-expect-error -- each payload must match its event type. + session.append([{ type: 'turn/start', data: message }]) +} diff --git a/packages/core/session/tests/canonical-envelopes.spec.ts b/packages/core/session/tests/canonical-envelopes.spec.ts index 1c52750edb..f570919109 100644 --- a/packages/core/session/tests/canonical-envelopes.spec.ts +++ b/packages/core/session/tests/canonical-envelopes.spec.ts @@ -37,13 +37,10 @@ function toolData(isError: boolean) { } function appendEvent(session: Session, event: SessionEvent) { - const append = session.append.bind(session) as ( - type: SessionEvent['type'], data: unknown, metadata?: unknown, - ) => SessionEvent - return append(event.type, event.data, { + return Reflect.apply(session.append.bind(session), session, [event.type, event.data, { ...event.surfaceOp === undefined ? {} : { surfaceOp: event.surfaceOp }, ...event.sourceEventSeqs === undefined ? {} : { sourceEventSeqs: event.sourceEventSeqs }, - }) + }]) as SessionEvent } const entryPaths: Record unknown> = { diff --git a/packages/core/session/tests/message-projections.spec.ts b/packages/core/session/tests/message-projections.spec.ts index 6eccae4f3e..91c4da769b 100644 --- a/packages/core/session/tests/message-projections.spec.ts +++ b/packages/core/session/tests/message-projections.spec.ts @@ -59,6 +59,24 @@ describe('plugin-owned message projections', () => { expect(source.data.content).toEqual([{ type: 'text', text: 'original' }]) }) + it.each(['projected', 'reject'])('admits message projections through array append (%s)', (text) => { + const session = Session.create(SessionId('array-projection'), undefined, undefined, undefined, [projection]) + const original = createUserMessage({ content: [{ type: 'text', text: 'original' }], source: { kind: 'user' } }) + const append = () => session.append([ + { type: 'user/message', data: original, surfaceOp: 'append' }, + { type: 'test/project', data: { seq: SessionSeq(0), text } }, + ]) + if (text === 'reject') { + expect(append).toThrow('rejected decision') + expect(session.seq).toBe(0) + expect(session.deriveMessages()).toEqual([]) + } else { + append() + expect(session.deriveMessages()[0]?.content).toEqual([{ type: 'text', text }]) + expect(session.surface.contentGeneration).toBe(1) + } + }) + it('registers by fiber and refuses cached or pending decisions after disposal', async () => { const ctx = new Context() contexts.push(ctx) diff --git a/packages/extensions/tool-cordis/src/api-catalog.ts b/packages/extensions/tool-cordis/src/api-catalog.ts index de37043f87..2d5a046664 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(type: T, data: SessionEventMap[T], ...opts: T extends SurfaceEventType ? [\n opts: SurfaceIntent\n ] : [\n ]): SessionEvent;\n requestHeader(): EpochHeader | undefined;\n requestContext(): RequestContext | undefined;\n toolHistory(): ToolHistory;\n deriveMessages(): Message[];\n deriveEventMessage(event: SessionEvent): Message | null;\n}', + 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: T | readonly SessionEventInput[], data?: SessionEventMap[T], ...opts: [\n opts?: SurfaceIntent\n ]): SessionEvent | readonly SessionEvent[];\n requestHeader(): EpochHeader | undefined;\n reque /* …truncated — full shape in source */', }, { name: 'SessionAccess', @@ -8645,6 +8645,10 @@ export const TYPE_API: readonly TypeApiEntry[] = [ name: 'SessionEventFrame', declaration: 'export type SessionEventFrame = {\n readonly type: \'ready\';\n} | SessionEventEntry;', }, + { + name: 'SessionEventInput', + declaration: 'export type SessionEventInput = {\n [K in SessionEventType]: {\n type: K;\n data: SessionEventMap[K];\n } & (K extends SurfaceEventType ? SurfaceIntent : {\n surfaceOp?: never;\n sourceEventSeqs?: never;\n });\n}[SessionEventType];', + }, { name: 'SessionEventMap', declaration: 'export interface SessionEventMap {\n \'turn/start\': {\n turn: number;\n };\n \'turn/end\': {\n turn: number;\n reason: TurnEndReason;\n };\n \'step/start\': {\n turn: number;\n step: number;\n };\n \'step/end\': {\n turn: number;\n step: number;\n };\n \'user/message\': UserMessage;\n \'developer/message\': {\n turn: number;\n step: number;\n message: DeveloperMessage;\n headerSeq?: SessionSeq;\n };\n \'system/message\': {\n turn: number;\n step: number;\n message: SystemMessage;\n };\n \'assistant/message\': {\n turn: number;\n step: number;\n message: AssistantMessage;\n stream: AssistantStreamRecord[];\n usage?: TokenUsage;\n interrupted?: true;\n };\n \'assistant/attempt\': {\n turn: number;\n step: number;\n stream: AssistantStreamRecord[];\n };\n \'tool/call\': {\n turn: number;\n step: number;\n callId: ToolCallId;\n name: string;\n arguments: string;\n };\n \'tool/result\': {\n turn: number;\n step: number;\n message: ToolResultMessage;\n error?: {\n name: string;\n code: string;\n reason?: string;\n };\n meta?: JsonValue;\n };\n \'request/header\': {\n header: EpochHeader;\n reason: RequestHeaderReason;\n startsSeries?: true;\n };\n \'request/context\': RequestContext;\n \'session/end-seed\': {\n /* …truncated — full shape in source */', diff --git a/scripts/type-equiv.manifest.json b/scripts/type-equiv.manifest.json index eb54361c8d..79ded69344 100644 --- a/scripts/type-equiv.manifest.json +++ b/scripts/type-equiv.manifest.json @@ -2821,6 +2821,11 @@ "doc": "docs/subsystems/forge.md", "symbol": "ForgeReviewResult", "source": "packages/experimental/forge/src/types.ts" + }, + { + "doc": "docs/subsystems/session.md", + "symbol": "SessionEventInput", + "source": "packages/core/session/src/types.ts" } ] }