feat(session): accept multiple events through append

This commit is contained in:
liukx0205 authored and Kexin Liu committed 2026-11-07 17:02:19 +08:00
1 parent e49a941d65
commit b5969f9e8d
19 files changed
+380 -26

No files matched your search

+2 -2
View File
@@ -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
+1 -1
View File
@@ -161,7 +161,7 @@ export type SessionEvent<T extends SessionEventType = SessionEventType> = {
}[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
+1 -1
View File
@@ -163,7 +163,7 @@ export type SessionEvent<T extends SessionEventType = SessionEventType> = {
}[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)
## 事件
+5 -2
View File
@@ -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
+26
View File
@@ -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<K> : { 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.
<a id="plugin-owned-message-projections"></a>
### 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<T>] : []
): SessionEvent<T>;
append<T extends SessionEventType>(
type: T | readonly SessionEventInput[],
data?: SessionEventMap[T],
...opts: [opts?: SurfaceIntent]
): SessionEvent<T> | 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
+26
View File
@@ -363,6 +363,18 @@ type SurfaceIntent<T extends SurfaceEventType = SurfaceEventType> = {
`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<K> : { surfaceOp?: never; sourceEventSeqs?: never })
}[SessionEventType]
```
数组中的每个成员将事件类型与对应载荷配对,且仅对产生消息的事件要求消息视图意图。
<a id="plugin-owned-message-projections"></a>
### 插件拥有的消息投影
@@ -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<T>] : []
): SessionEvent<T>;
append<T extends SessionEventType>(
type: T | readonly SessionEventInput[],
data?: SessionEventMap[T],
...opts: [opts?: SurfaceIntent]
): SessionEvent<T> | 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
@@ -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) => {
+2 -2
View File
@@ -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
+2
View File
@@ -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.
+2
View File
@@ -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)` 分发需等待完成的持久性检查点:每个持久化监听器都会刷新,调用在所有监听器结算后完成。需要立即持久性屏障的生产方应等待它,而不是假定延后写入已经排空。
+64 -3
View File
@@ -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<T>] : []
): SessionEvent<T> {
): SessionEvent<T>
append<T extends SessionEventType>(
type: T | readonly SessionEventInput[],
data?: SessionEventMap[T],
...opts: [opts?: SurfaceIntent]
): SessionEvent<T> | 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)
}
+38
View File
@@ -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()
+6
View File
@@ -485,6 +485,12 @@ export type SurfaceIntent<T extends SurfaceEventType = SurfaceEventType> = {
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<K> : { surfaceOp?: never; sourceEventSeqs?: never })
}[SessionEventType]
/**
* One immutable entry in the session log.
*
+144
View File
@@ -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])
})
})
@@ -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<SessionEvent<'turn/start'>>()
const inputs: readonly SessionEventInput[] = [
{ type: 'turn/start', data: { turn: 1 } },
{ type: 'user/message', data: message, surfaceOp: 'append' },
]
expectTypeOf(session.append(inputs)).toEqualTypeOf<readonly SessionEvent[]>()
// @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 }])
}
@@ -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<string, (event: SessionEvent) => unknown> = {
@@ -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)
@@ -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<T extends SessionEventType>(type: T, data: SessionEventMap[T], ...opts: T extends SurfaceEventType ? [\n opts: SurfaceIntent<T>\n ] : [\n ]): SessionEvent<T>;\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<T extends SessionEventType>(type: T, data: SessionEventMap[T], ...opts: T extends SurfaceEventType ? [\n opts: SurfaceIntent<T>\n ] : [\n ]): SessionEvent<T>;\n append<T extends SessionEventType>(type: T | readonly SessionEventInput[], data?: SessionEventMap[T], ...opts: [\n opts?: SurfaceIntent\n ]): SessionEvent<T> | 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<K> : {\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 */',
+5
View File
@@ -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"
}
]
}