1454 lines
64 KiB
TypeScript
1454 lines
64 KiB
TypeScript
/**
|
|
* Event-sourced session service: append-only session log, in-memory store, and
|
|
* the derived LLM message history. Persistence is a plugin concern (subscribe
|
|
* to `session/event`, drain on `session/flush`).
|
|
*
|
|
* @module @deepseek-ai/dsh-session
|
|
*/
|
|
|
|
import { Context, Service } from '@deepseek-ai/cordis'
|
|
import { isAbsolute } from 'node:path'
|
|
import { brandString } from '@deepseek-ai/dsh-brand'
|
|
import { assertNever, deepFreeze, snapshotJsonValue } from '@deepseek-ai/dsh-util-values'
|
|
import { scopeOf, scopeTarget } from '@deepseek-ai/dsh-scope'
|
|
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 { SurfaceManager, validateSessionEventData, validateSurfaceMetadata } from './surface.ts'
|
|
import type { SessionSurface, SessionMessageProjection } from './surface.ts'
|
|
import { foldRequestHeader } from './request-header.ts'
|
|
import { ToolHistoryProjection } from './tool-history.ts'
|
|
import type { ToolHistory } from '@deepseek-ai/dsh-llm'
|
|
|
|
import { buildForkSeed } from './fork.ts'
|
|
|
|
export { buildForkSeed } from './fork.ts'
|
|
export * from './types.ts'
|
|
export { SessionPreparation } from './preparation.ts'
|
|
export type { SessionPreparationOptions } from './preparation.ts'
|
|
export type { AssistantMessage, DeveloperMessage, SystemMessage, ToolResultMessage, UserMessage } from '@deepseek-ai/dsh-llm'
|
|
export { interruptedTurnClosers, ToolCallRecovery, TOOL_NOT_STARTED, TOOL_OUTCOME_UNKNOWN } from './repair.ts'
|
|
export type { SessionSurface, SurfaceFoldReplacement, SurfaceFoldResult, SessionMessageProjection, SessionMessageProjectionContext } from './surface.ts'
|
|
export { deriveEventMessage, foldSurface, isAppendSurfaceEvent, isReplacementSurfaceEvent, isSurfaceEvent, isSurfaceEligibleType } from './surface.ts'
|
|
export { canonicalHeader, foldRequestHeader, headerEquals } from './request-header.ts'
|
|
export { KNOWN_SESSION_EVENT_TYPES } from './known-event-types.ts'
|
|
|
|
declare module '@deepseek-ai/cordis' {
|
|
interface Context {
|
|
sessions: SessionStore
|
|
}
|
|
|
|
interface Events {
|
|
/**
|
|
* Creation announcement during session publication. A synchronous throw vetoes and rolls
|
|
* back with a paired disposal; detach requested during dispatch is deferred.
|
|
* A returned-promise rejection is logged but cannot retroactively veto this
|
|
* synchronous boundary.
|
|
* Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): agent-scoped listeners
|
|
* receive only sessions entered through that agent's context.
|
|
* @param session - the session just entered and announced.
|
|
* @mode emit
|
|
*/
|
|
'session/created'(this: Scoped<Session>, session: Session): void
|
|
/**
|
|
* Emitted once when an announced session leaves the store, including
|
|
* publication rollback, but never for an entry whose creation announcement
|
|
* did not begin. Listener failures are logged and contained.
|
|
* Scope-filtered dispatch (`@deepseek-ai/dsh-scope`) reuses the owner scope.
|
|
* @param session - the session that is no longer live in the store.
|
|
* @mode emit
|
|
*/
|
|
'session/disposed'(this: Scoped<Session>, session: Session): void
|
|
/**
|
|
* Post-commit, fire-and-forget append feed. The listener snapshot resolves
|
|
* before the log push, but callbacks run after it; observer failures are
|
|
* logged and contained without making the committed append fail.
|
|
* Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): agent-scoped listeners
|
|
* receive only events from sessions entered through that agent's context.
|
|
* @param session - the session whose log grew.
|
|
* @param event - the appended event, exactly as recorded.
|
|
* @mode emit
|
|
*/
|
|
'session/event'(this: Scoped<Session>, session: Session, event: SessionEvent): void
|
|
/**
|
|
* Awaited parallel durability checkpoint: every listener runs and the
|
|
* caller awaits all of them, with no waterfall veto. Scope-filtered dispatch
|
|
* (`@deepseek-ai/dsh-scope`) reuses the session's owner scope.
|
|
* @param session - the session whose buffered events must reach durable storage.
|
|
* @mode parallel
|
|
*/
|
|
'session/flush'(this: Scoped<Session>, session: Session): Promise<void> | void
|
|
}
|
|
}
|
|
|
|
declare module '@deepseek-ai/dsh-typert-protocol' {
|
|
interface TypertLookupMap {
|
|
session: TypertLookup<Session, SessionId>
|
|
}
|
|
}
|
|
|
|
/** Validate and freeze one detached creation header in place. */
|
|
function validateSessionHeader(id: SessionId, input: unknown): SessionHeader {
|
|
if (input === null || typeof input !== 'object' || Array.isArray(input)) {
|
|
throw new Error('session header is not a plain JSON record')
|
|
}
|
|
const record = input as Record<string, unknown>
|
|
if (Object.hasOwn(record, 'seedLength')) {
|
|
throw new Error('session header has invalid field "seedLength"')
|
|
}
|
|
if (record.version !== SESSION_FORMAT_VERSION) {
|
|
throw new Error(`session header version must be ${SESSION_FORMAT_VERSION}, got ${String(record.version)}`)
|
|
}
|
|
if (record.id !== id) {
|
|
throw new Error(`session header id "${String(record.id)}" does not match session id "${id}"`)
|
|
}
|
|
if (typeof record.createdAt !== 'number'
|
|
|| !Number.isSafeInteger(record.createdAt)
|
|
|| record.createdAt < 0) {
|
|
throw new Error('session header createdAt must be a non-negative safe integer')
|
|
}
|
|
if (record.cwd !== undefined) {
|
|
if (typeof record.cwd !== 'string') throw new Error('session header cwd must be a string')
|
|
if (!isAbsolute(record.cwd)) {
|
|
throw new Error(`session header cwd must be an absolute path, got "${record.cwd}"`)
|
|
}
|
|
}
|
|
if (record.parentSession !== undefined && typeof record.parentSession !== 'string') {
|
|
throw new Error('session header parentSession must be a string')
|
|
}
|
|
if (typeof record.isSeeded !== 'boolean') {
|
|
throw new Error('session header isSeeded must be a boolean')
|
|
}
|
|
if (record.origin !== undefined && record.origin !== 'subagent') {
|
|
throw new Error('session header origin must be "subagent"')
|
|
}
|
|
if (record.delegationDepth !== undefined
|
|
&& (typeof record.delegationDepth !== 'number' || !Number.isSafeInteger(record.delegationDepth) || record.delegationDepth < 0)) {
|
|
throw new Error('session header delegationDepth must be a non-negative safe integer')
|
|
}
|
|
if (record.agentPreset !== undefined && typeof record.agentPreset !== 'string') {
|
|
throw new Error('session header agentPreset must be a string')
|
|
}
|
|
return deepFreeze(record as unknown as SessionHeader)
|
|
}
|
|
|
|
/** Validate and freeze one exclusively owned persistence header in place. */
|
|
function validateRestoredSessionHeader(id: SessionId, input: unknown): SessionHeader {
|
|
if (input !== null && typeof input === 'object' && !Array.isArray(input)) {
|
|
const prototype = Reflect.getPrototypeOf(input)
|
|
if (prototype !== Object.prototype && prototype !== null) {
|
|
throw new Error('session header is not a plain JSON record')
|
|
}
|
|
}
|
|
return validateSessionHeader(id, input)
|
|
}
|
|
|
|
/** Detach, validate, and freeze the creation metadata published by a session. */
|
|
function snapshotSessionHeader(id: SessionId, source?: SessionHeader): SessionHeader {
|
|
const input: unknown = source === undefined
|
|
? { version: SESSION_FORMAT_VERSION, id, createdAt: Date.now(), isSeeded: false }
|
|
: source
|
|
const snapshot = snapshotJsonValue(input)
|
|
if (snapshot === undefined) throw new Error('session header is not losslessly JSON-serializable')
|
|
return validateSessionHeader(id, snapshot)
|
|
}
|
|
|
|
/**
|
|
* Validate an exclusively owned event and deeply freeze its identified message
|
|
* without copying the event. The caller transfers an object graph that no
|
|
* producer retains and that shares no mutable children with another event.
|
|
* Use {@link snapshotSessionEvent} when exclusive ownership is not guaranteed.
|
|
* @param event - exclusively owned event imported across a trusted boundary.
|
|
* @returns the same event object with a validated, deeply frozen message.
|
|
* @throws when event-local surface metadata, request-header fields, or message invariants are invalid; history relations are not checked.
|
|
*/
|
|
export function adoptSessionEvent<T extends SessionEvent>(event: T): T {
|
|
validateSessionEventData(event, `session event at seq ${event.seq}`)
|
|
validateSurfaceMetadata(event)
|
|
assertMessageEventShape(
|
|
event,
|
|
`session event at seq ${event.seq}`,
|
|
)
|
|
switch (event.type) {
|
|
case 'user/message':
|
|
deepFreeze(event.data)
|
|
break
|
|
case 'developer/message':
|
|
case 'system/message':
|
|
case 'assistant/message':
|
|
case 'tool/result':
|
|
deepFreeze(event.data.message)
|
|
break
|
|
default:
|
|
// SessionEventMap is merge-extensible; plugin-owned events carry no core message.
|
|
break
|
|
}
|
|
return event
|
|
}
|
|
|
|
/**
|
|
* Detach one event while preserving deep immutability for its identified message.
|
|
* @param event - event imported across a query or persistence boundary.
|
|
* @returns a detached event snapshot with a validated, deeply frozen message.
|
|
*/
|
|
export function snapshotSessionEvent<T extends SessionEvent>(event: T): T {
|
|
return adoptSessionEvent(structuredClone(event))
|
|
}
|
|
|
|
/** Validate the fixed event envelope after one-pass JSON materialization. */
|
|
function assertSessionEventEnvelope(value: unknown, index: number): asserts value is SessionEvent {
|
|
if (value === null || typeof value !== 'object' || Array.isArray(value)) {
|
|
throw new Error(`seed event at index ${index} has an invalid event envelope`)
|
|
}
|
|
const event = value as Record<string, unknown>
|
|
for (const key in event) {
|
|
switch (key) {
|
|
case 'type':
|
|
case 'seq':
|
|
case 'time':
|
|
case 'data':
|
|
case 'surfaceOp':
|
|
case 'sourceEventSeqs':
|
|
case 'ignorable':
|
|
break
|
|
default:
|
|
throw new Error(`seed event at index ${index} has an invalid event envelope`)
|
|
}
|
|
}
|
|
const type = event['type']
|
|
const seq = event['seq']
|
|
const time = event['time']
|
|
if (typeof type !== 'string'
|
|
|| typeof seq !== 'number' || !Number.isSafeInteger(seq) || seq < 0 || Object.is(seq, -0)
|
|
|| typeof time !== 'number' || !Number.isSafeInteger(time)
|
|
|| event['data'] === undefined
|
|
|| (event['ignorable'] !== undefined && event['ignorable'] !== true)) {
|
|
throw new Error(`seed event at index ${index} has an invalid event envelope`)
|
|
}
|
|
validateSessionEventData(event as SessionEvent, `seed ${type} at index ${index}`)
|
|
switch (type) {
|
|
case 'request/header':
|
|
case 'developer/message':
|
|
case 'system/message':
|
|
case 'user/message':
|
|
case 'assistant/attempt':
|
|
case 'assistant/message':
|
|
case 'tool/result':
|
|
assertCurrentLlmShape(event, index)
|
|
break
|
|
}
|
|
}
|
|
|
|
/** Reject obsolete request headers and malformed messages at the seed/load boundary. */
|
|
function assertCurrentLlmShape(event: Record<string, unknown>, index: number): void {
|
|
const data = event['data']
|
|
const record = typeof data === 'object' && data !== null
|
|
? data as Record<string, unknown>
|
|
: undefined
|
|
if (event['type'] === 'request/header') {
|
|
const headerRecord = record?.['header'] as Record<string, unknown>
|
|
const config = headerRecord['config']
|
|
if (!hasProviderModel(config)) throw new Error(`seed request/header at index ${index} lacks provider/model`)
|
|
const configRecord = config as Record<string, unknown>
|
|
const reasoningEffort = configRecord['reasoningEffort']
|
|
if (reasoningEffort !== undefined
|
|
&& (typeof reasoningEffort !== 'string' || reasoningEffort.length === 0)) {
|
|
throw new Error(`seed request/header at index ${index} has an invalid reasoningEffort`)
|
|
}
|
|
assertAdapterDefaults(headerRecord['adapterDefaults'], configRecord, index)
|
|
const reason = record?.['reason']
|
|
if (reason !== 'initial' && reason !== 'resume' && reason !== 'change' && reason !== 'series') {
|
|
throw new Error(`seed request/header at index ${index} has an invalid reason`)
|
|
}
|
|
if (record?.['startsSeries'] !== undefined && record['startsSeries'] !== true) {
|
|
throw new Error(`seed request/header at index ${index} has an invalid startsSeries marker`)
|
|
}
|
|
}
|
|
const type = event['type']
|
|
if (type === 'assistant/attempt') {
|
|
assertAssistantSettlementShape(record, type, index)
|
|
return
|
|
}
|
|
if (!isMessageEventType(type)) return
|
|
assertMessageEventShape(event, `seed ${type} at index ${index}`)
|
|
if (type === 'assistant/message') {
|
|
assertAssistantSettlementShape(record, type, index)
|
|
}
|
|
}
|
|
|
|
/** Validate fields used directly by restored Session lifecycle logic without replaying the embedded stream. */
|
|
function assertAssistantSettlementShape(
|
|
data: Record<string, unknown> | undefined,
|
|
type: 'assistant/attempt' | 'assistant/message',
|
|
index: number,
|
|
): void {
|
|
const turn = data?.['turn']
|
|
const step = data?.['step']
|
|
if (typeof turn !== 'number' || !Number.isSafeInteger(turn) || turn < 0 || Object.is(turn, -0)
|
|
|| typeof step !== 'number' || !Number.isSafeInteger(step) || step < 0 || Object.is(step, -0)
|
|
|| !Array.isArray(data?.['stream'])) {
|
|
throw new Error(`seed ${type} at index ${index} has invalid settlement fields`)
|
|
}
|
|
}
|
|
|
|
const allowedAdapterKeys = new Set(['reasoningEffort', 'maxTokens'])
|
|
|
|
/** Validate adapter-default markers imported from a durable request header. */
|
|
function assertAdapterDefaults(
|
|
value: unknown,
|
|
config: Record<string, unknown>,
|
|
index: number,
|
|
): void {
|
|
if (value === undefined) return
|
|
if (typeof value !== 'object' || value === null || Array.isArray(value)) {
|
|
throw new Error(`seed request/header at index ${index} has invalid adapterDefaults`)
|
|
}
|
|
const defaults = value as Record<string, unknown>
|
|
if (Object.keys(defaults).some(key => !allowedAdapterKeys.has(key))
|
|
|| Object.values(defaults).some(marker => marker !== true)
|
|
|| defaults['reasoningEffort'] === true && config['reasoningEffort'] === undefined
|
|
|| defaults['maxTokens'] === true && config['maxTokens'] === undefined) {
|
|
throw new Error(`seed request/header at index ${index} has invalid adapterDefaults`)
|
|
}
|
|
}
|
|
|
|
/** The surface event types whose payload carries an identified message. */
|
|
function isMessageEventType(type: unknown): type is SurfaceEventType {
|
|
return type === 'developer/message' || type === 'system/message' || type === 'user/message'
|
|
|| type === 'assistant/message' || type === 'tool/result'
|
|
}
|
|
|
|
const MESSAGE_ROLE_BY_TYPE: Record<SurfaceEventType, Message['role']> = {
|
|
'system/message': 'system',
|
|
'developer/message': 'developer',
|
|
'user/message': 'user',
|
|
'assistant/message': 'assistant',
|
|
'tool/result': 'tool',
|
|
}
|
|
|
|
/** Validate only the event-specific invariants needed to safely replay a message. */
|
|
function assertMessageEventShape(event: Record<string, unknown>, subject: string): void {
|
|
const type = event['type']
|
|
if (!isMessageEventType(type)) return
|
|
const data = event['data']
|
|
const record = typeof data === 'object' && data !== null
|
|
? data as Record<string, unknown>
|
|
: undefined
|
|
const message = type === 'user/message' ? record : record?.['message']
|
|
if (typeof message !== 'object' || message === null
|
|
|| typeof (message as Record<string, unknown>)['id'] !== 'string'
|
|
|| (message as Record<string, unknown>)['id'] === '') {
|
|
throw new Error(`${subject} lacks an identified message`)
|
|
}
|
|
const messageRecord = message as Record<string, unknown>
|
|
const expectedRole = MESSAGE_ROLE_BY_TYPE[type]
|
|
if (messageRecord['role'] !== expectedRole) {
|
|
throw new Error(`${subject} message must have role "${expectedRole}"`)
|
|
}
|
|
const source = messageRecord['source']
|
|
if (typeof source !== 'object' || source === null
|
|
|| typeof (source as Record<string, unknown>)['kind'] !== 'string'
|
|
|| (source as Record<string, unknown>)['kind'] === '') {
|
|
throw new Error(`${subject} message has invalid source`)
|
|
}
|
|
if (!Array.isArray(messageRecord['content'])) {
|
|
throw new Error(`${subject} message has invalid content`)
|
|
}
|
|
const sourceRecord = source as Record<string, unknown>
|
|
if (type === 'system/message') {
|
|
if (sourceRecord['kind'] !== 'system-prompt') {
|
|
throw new Error(`${subject} message must have system-prompt source`)
|
|
}
|
|
return
|
|
}
|
|
if (type === 'assistant/message') {
|
|
if (sourceRecord['kind'] !== 'model' || !hasProviderModel(sourceRecord)) {
|
|
throw new Error(`${subject} message must have model source`)
|
|
}
|
|
return
|
|
}
|
|
if (type !== 'tool/result') return
|
|
if (sourceRecord['kind'] !== 'tool'
|
|
|| typeof sourceRecord['callId'] !== 'string'
|
|
|| sourceRecord['callId'] === '') {
|
|
throw new Error(`${subject} message must have tool source`)
|
|
}
|
|
if (messageRecord['toolCallId'] !== sourceRecord['callId']) {
|
|
throw new Error(`${subject} message has mismatched tool call ids`)
|
|
}
|
|
}
|
|
|
|
/** Whether an unknown value carries the current provider/model pair. */
|
|
function hasProviderModel(value: unknown): boolean {
|
|
if (typeof value !== 'object' || value === null) return false
|
|
const pair = value as Record<string, unknown>
|
|
return typeof pair['provider'] === 'string' && pair['provider'].length > 0
|
|
&& typeof pair['model'] === 'string' && pair['model'].length > 0
|
|
}
|
|
|
|
type SessionCallback = (...args: unknown[]) => unknown
|
|
|
|
/** Resolve one listener snapshot, including Cordis's internal dispatch checks. */
|
|
function collectSessionCallbacks(ctx: Context, args: unknown[]): SessionCallback[] {
|
|
return [...ctx.events.dispatch('emit', args)] as SessionCallback[]
|
|
}
|
|
|
|
/** Invoke one resolved observe-only listener snapshot with per-listener containment. */
|
|
function invokeContainedSessionObservers(
|
|
ctx: Context,
|
|
name: 'session/event' | 'session/disposed',
|
|
id: SessionId,
|
|
args: unknown[],
|
|
callbacks: SessionCallback[],
|
|
): void {
|
|
for (const callback of callbacks) {
|
|
try {
|
|
const returned: unknown = callback(...args)
|
|
void Promise.resolve(returned).catch((error: unknown) => {
|
|
ctx.logger.warn(`session "${id}": ${name} listener rejected: ${String(error)}`)
|
|
})
|
|
} catch (error: unknown) {
|
|
ctx.logger.warn(`session "${id}": ${name} listener threw: ${String(error)}`)
|
|
}
|
|
}
|
|
}
|
|
|
|
/** All mutable lifecycle state for one exact store entry. */
|
|
interface SessionEntry {
|
|
readonly id: SessionId
|
|
readonly session: Session
|
|
readonly carrier: Scoped<Session>
|
|
readonly emitCtx: Context
|
|
announced: boolean
|
|
announcing: boolean
|
|
appending: boolean
|
|
detachRequested: boolean
|
|
detach(): void
|
|
}
|
|
|
|
/** Store attachment for the append path; module-private to keep Session store-agnostic publicly. */
|
|
const attachments = new WeakMap<Session, SessionEntry>()
|
|
|
|
/**
|
|
* Grammar of a {@link PluginRecordType}: `plugin:` followed by
|
|
* slash-separated segments of lowercase letters, digits, `.`, `_`, and `-`,
|
|
* each segment starting with a letter or digit.
|
|
*/
|
|
const PLUGIN_RECORD_TYPE = /^plugin:[a-z0-9][a-z0-9._-]*(?:\/[a-z0-9][a-z0-9._-]*)*$/u
|
|
|
|
/**
|
|
* Whether one event type name follows the plugin record grammar.
|
|
* @param type - an event type name.
|
|
* @returns whether the name is a {@link PluginRecordType}.
|
|
*/
|
|
function isPluginRecordType(type: string): type is PluginRecordType {
|
|
return PLUGIN_RECORD_TYPE.test(type)
|
|
}
|
|
|
|
/**
|
|
* Type one constructed plugin record as a log event after checking its type
|
|
* against the plugin record grammar. A restored Session already holds events
|
|
* whose types are outside `SessionEventMap` when they are marked ignorable;
|
|
* a plugin record is such an event.
|
|
* @param value - the frozen record envelope built by {@link Session}.
|
|
* @throws when the record type is outside the plugin record grammar.
|
|
*/
|
|
function assertPluginRecordEvent(value: unknown): asserts value is SessionEvent {
|
|
const { type } = value as { readonly type: string }
|
|
if (!isPluginRecordType(type)) {
|
|
throw new Error(`plugin record type "${type}" must be "plugin:" followed by lowercase slash-separated segments`)
|
|
}
|
|
}
|
|
|
|
/** Module-private access to {@link Session}'s record commit, assigned once by its static block. */
|
|
let commitPluginRecord: (session: Session, type: PluginRecordType, data: unknown) => SessionSeq
|
|
|
|
/**
|
|
* Append one experimental plugin record: an event marked `ignorable` whose
|
|
* type is declared in {@link PluginRecordMap}, outside {@link SessionEventMap}.
|
|
* The plugin record catalog lists it separately from released event schemas.
|
|
* A reader that does not recognize the type retains and skips the record, which never
|
|
* enters the model-visible surface, and fork and resume carry it with the rest
|
|
* of the log. A Session format migration retains records on a best-effort
|
|
* basis, so a record holds plugin-owned state that its owner can lose without
|
|
* changing how the rest of the log is interpreted.
|
|
*
|
|
* Only production source under `packages/experimental/` may call this
|
|
* function; the `verify-plugin-record-callers` gate rejects any other
|
|
* production caller in this repository. Read records back with
|
|
* {@link pluginRecordOf}.
|
|
* @param session - the Session whose log receives the record.
|
|
* @param type - declared record name: `plugin:` followed by slash-separated
|
|
* segments of lowercase letters, digits, `.`, `_`, and `-`, each starting
|
|
* with a letter or digit.
|
|
* @param data - JSON payload, snapshotted before it enters the log.
|
|
* @returns the sequence number of the committed record.
|
|
* @throws when `type` is outside that grammar, when `data` is not losslessly
|
|
* JSON-serializable, or when the call reenters another append's publication;
|
|
* a rejected record does not change the log.
|
|
*/
|
|
export function appendPluginRecord<K extends Extract<keyof PluginRecordMap, PluginRecordType>>(
|
|
session: Session,
|
|
type: K,
|
|
data: NoInfer<PluginRecordMap[K]>,
|
|
): SessionSeq {
|
|
return commitPluginRecord(session, type, data)
|
|
}
|
|
|
|
/**
|
|
* Read one committed event as a plugin record. The V3-to-V4 format edge also
|
|
* renames each unknown ignorable V3 event into the `plugin:` namespace, so an
|
|
* owner names its records under its own package name and validates `data`.
|
|
* @param event - any committed Session event, live or restored.
|
|
* @returns the record, or undefined when the event is not an ignorable event
|
|
* whose type follows the plugin record grammar.
|
|
*/
|
|
export function pluginRecordOf(event: SessionEvent): PluginRecord | undefined {
|
|
const type: string = event.type
|
|
if (event.ignorable !== true || !isPluginRecordType(type)) return undefined
|
|
return { type, seq: event.seq, time: event.time, data: event.data }
|
|
}
|
|
|
|
/**
|
|
* An event-sourced session: an append-only log of {@link SessionEvent}s.
|
|
*
|
|
* Plain class (not a Service) — create live instances via
|
|
* `ctx.sessions.create()` and detached instances via {@link create}.
|
|
* Seeding with an existing event log replays/forks a session.
|
|
* @typert object
|
|
*/
|
|
export class Session {
|
|
private log: SessionEvent[] = []
|
|
/** Single incremental owner of surface acceptance and projection state. */
|
|
private readonly surfaceManager: SurfaceManager
|
|
|
|
/** The ordered surface over this session's event log. */
|
|
get surface(): SessionSurface {
|
|
return this.surfaceManager
|
|
}
|
|
|
|
/**
|
|
* Detached, deep-frozen creation metadata (format version, cwd, lineage,
|
|
* and whether fork history exists). Supplied by the store via `ctx.sessions.create()`. When a
|
|
* `Session` is created without a store-owned header, a minimal header is
|
|
* synthesized (stamped with the current {@link SESSION_FORMAT_VERSION}) so
|
|
* `session.header` is always present. Kept out of the event log — it is a
|
|
* storage concern, not replayable conversation state.
|
|
*/
|
|
readonly header: SessionHeader
|
|
|
|
/** Number of leading events inherited from this Session's fork parent. */
|
|
readonly inheritedEventCount: SessionLogOffset
|
|
|
|
/** The session identity, derived from its durable header's single copy. */
|
|
get id(): SessionId {
|
|
return this.header.id
|
|
}
|
|
|
|
/**
|
|
* The constructor seed length (0 without one), before any marker appended
|
|
* during construction. Seed events never publish on `session/event`. A
|
|
* marker appended before the store attaches occupies this seq without
|
|
* publishing either; otherwise this seq is available for the next append.
|
|
*
|
|
* This in-process offset is not persisted. A fork seed can already contain
|
|
* the child's inherited marker and synthetic closers, so its child-owned
|
|
* history starts at {@link inheritedEventCount}, before this offset. A
|
|
* resumed Session's seed contains its full stored log, while its inherited
|
|
* count keeps the durable fork cut. Consumers needing complete canonical
|
|
* history start at seq 0.
|
|
*/
|
|
readonly firstLiveSeq: SessionLogOffset
|
|
|
|
/**
|
|
* First event produced for this object lifecycle. A new fork includes its
|
|
* child-owned seed marker and closers; a restored Session starts after its
|
|
* complete stored prefix. This in-process capture offset is not persisted.
|
|
*/
|
|
readonly firstLifecycleSeq: SessionLogOffset
|
|
|
|
/**
|
|
* Create a detached session by validating and snapshotting borrowed seed
|
|
* events and storage metadata.
|
|
* @param id - session identity.
|
|
* @param seed - optional borrowed replay or fork events.
|
|
* @param header - optional borrowed storage metadata.
|
|
* @param inheritedEventCount - exact fork-inherited prefix length for a seeded header.
|
|
* @param projections - pure interpreters for plugin-owned message changes.
|
|
* @returns a detached session.
|
|
* @throws when a seed event requires a missing message interpreter or fails validation.
|
|
*/
|
|
static create(
|
|
id: SessionId,
|
|
seed?: readonly SessionEvent[],
|
|
header?: SessionHeader,
|
|
inheritedEventCount?: SessionLogOffset,
|
|
projections?: readonly SessionMessageProjection[],
|
|
): Session {
|
|
return new Session(id, seed, header, 'snapshot', inheritedEventCount, projections)
|
|
}
|
|
|
|
/**
|
|
* Restore a detached session by adopting an independently owned or deeply frozen seed.
|
|
* Runtime-required event fields, event envelopes, sequence continuity, surface
|
|
* transitions, and header fields are validated without copying or freezing events.
|
|
* Embedded Assistant streams remain opaque until a stream consumer or storage
|
|
* verifier reads them.
|
|
* @param id - restored session identity.
|
|
* @param seed - independently owned or deeply frozen events.
|
|
* @param header - independently owned storage metadata.
|
|
* @param inheritedEventCount - exact fork-inherited prefix length decoded from storage.
|
|
* @param eventState - aliasing state carried from the operation that produced the seed.
|
|
* @param projections - pure interpreters for plugin-owned message changes.
|
|
* @returns a restored detached session.
|
|
* @throws when a seed event requires a missing message interpreter or fails validation.
|
|
*/
|
|
static fromRestore(
|
|
id: SessionId,
|
|
seed: readonly SessionEvent[],
|
|
header: SessionHeader,
|
|
inheritedEventCount: SessionLogOffset,
|
|
eventState: SessionSeedEventState,
|
|
projections?: readonly SessionMessageProjection[],
|
|
): Session {
|
|
return new Session(
|
|
id,
|
|
seed,
|
|
header,
|
|
eventState,
|
|
inheritedEventCount,
|
|
projections,
|
|
)
|
|
}
|
|
|
|
private constructor(
|
|
id: SessionId,
|
|
seed?: readonly SessionEvent[],
|
|
header?: SessionHeader,
|
|
mode: 'snapshot' | SessionSeedEventState = 'snapshot',
|
|
suppliedInheritedEventCount?: SessionLogOffset,
|
|
projections: readonly SessionMessageProjection[] = [],
|
|
) {
|
|
this.surfaceManager = new SurfaceManager(this.log, SessionLogOffset(0), projections)
|
|
const restoredHeader = mode === 'snapshot' ? undefined : validateRestoredSessionHeader(id, header)
|
|
if (seed !== undefined) {
|
|
// Validate the seed to the SAME invariants `append` enforces, so a
|
|
// replay/fork (`ctx.sessions.create(id, { seed })`) cannot construct a
|
|
// live log that no persistence backend could store: each event's `data`
|
|
// must be JSON-serializable, and `seq` must be contiguous from 0 (the
|
|
// `seq = log.length` contract the whole system relies on). Without this,
|
|
// a bad seed would surface only later as a backend rejection or a silent
|
|
// divergence between the live log and disk.
|
|
for (const [index, source] of seed.entries()) {
|
|
// The seed is a persistence/replay boundary: validate and detach the
|
|
// complete event in one lossless-JSON pass.
|
|
const snapshot = mode === 'snapshot' ? snapshotJsonValue(source) : source
|
|
if (snapshot === undefined) {
|
|
throw new Error(`seed event at index ${index} is not losslessly JSON-serializable`)
|
|
}
|
|
assertSessionEventEnvelope(snapshot, index)
|
|
if (snapshot.seq !== index) {
|
|
throw new Error(`seed event at index ${index} has seq ${snapshot.seq} (expected ${index}); seed must be contiguous from 0`)
|
|
}
|
|
// A seed is accepted incrementally through the same transition as a
|
|
// live append and a full-log fold. The candidate is planned before it
|
|
// enters `log`, so a failure cannot partially mutate the surface.
|
|
try {
|
|
this.surfaceManager.validateNext(snapshot)
|
|
} catch (error: unknown) {
|
|
throw new Error(`invalid seed event at index ${index}: ${error instanceof Error ? error.message : 'invalid surface metadata'}`)
|
|
}
|
|
this.log.push(mode === 'snapshot' ? deepFreeze(snapshot) : snapshot)
|
|
}
|
|
}
|
|
this.firstLiveSeq = SessionLogOffset(this.log.length)
|
|
this.header = restoredHeader ?? snapshotSessionHeader(id, header)
|
|
if (this.header.isSeeded && seed === undefined) {
|
|
throw new Error('seeded session requires an explicit constructor seed')
|
|
}
|
|
if (this.header.isSeeded && suppliedInheritedEventCount === undefined) {
|
|
throw new Error('seeded session requires an inherited event count')
|
|
}
|
|
const inheritedEventCount = SessionLogOffset(suppliedInheritedEventCount ?? 0)
|
|
if (!this.header.isSeeded && inheritedEventCount !== 0) {
|
|
throw new Error('unseeded session inherited event count must be 0')
|
|
}
|
|
if (inheritedEventCount > this.log.length) {
|
|
throw new Error('session inherited event count exceeds its event log')
|
|
}
|
|
const seedMarker = this.log[inheritedEventCount]
|
|
const markedSeed = seedMarker?.type === 'session/end-seed' && seedMarker.data.inherited === true
|
|
if (mode === 'snapshot' && this.header.isSeeded && inheritedEventCount !== this.log.length && !markedSeed) {
|
|
throw new Error('seeded session constructor seed must equal its inherited prefix or mark its inherited cut')
|
|
}
|
|
if (markedSeed && this.log.slice(inheritedEventCount + 1).some(event => event.type === 'session/end-seed' && event.data.inherited === true)) {
|
|
throw new Error('session inherited event count must identify the final inherited marker')
|
|
}
|
|
this.inheritedEventCount = inheritedEventCount
|
|
this.firstLifecycleSeq = mode === 'snapshot' && this.header.isSeeded ? inheritedEventCount : this.firstLiveSeq
|
|
// A fresh seeded child always owns one tagged marker at its inherited cut,
|
|
// even when the copied prefix already ends in an ancestor marker. Restore
|
|
// retains that durable marker and appends only the ordinary resume marker.
|
|
if (seed !== undefined && mode === 'snapshot' && this.header.isSeeded && !markedSeed) {
|
|
this.append('session/end-seed', { inherited: true })
|
|
} else if (seed !== undefined && !(mode === 'snapshot' && this.header.isSeeded) && this.log.at(-1)?.type !== 'session/end-seed') {
|
|
this.append('session/end-seed', {})
|
|
}
|
|
}
|
|
|
|
/** Cached immutable full snapshot of the private append-only log. */
|
|
private eventsSnapshot: readonly SessionEvent[] | undefined
|
|
|
|
/**
|
|
* Return the immutable event stored at one exact sequence number.
|
|
* @deprecated Existing logic may remain unmigrated for now, but new calls are prohibited.
|
|
* See the [Agent Note](../../../../.agents/notes/implemented/architecture/2026-09-09-deprecate-synchronous-session-event-reads.md).
|
|
* @param seq - event sequence number.
|
|
* @returns the accepted event, or undefined when the log does not contain it.
|
|
*/
|
|
eventAt(seq: SessionSeq): SessionEvent | undefined {
|
|
return this.log[seq]
|
|
}
|
|
|
|
/**
|
|
* Materialize an immutable snapshot of a half-open event sequence range.
|
|
* A full current snapshot is reused until the next append; every previously
|
|
* returned snapshot remains stable after later appends.
|
|
* @deprecated Existing logic may remain unmigrated for now, but new calls are prohibited.
|
|
* See the [Agent Note](../../../../.agents/notes/implemented/architecture/2026-09-09-deprecate-synchronous-session-event-reads.md).
|
|
* @param fromSeq - non-negative inclusive sequence number; defaults to the log start.
|
|
* @param toSeqExclusive - non-negative exclusive sequence number; defaults to the current end.
|
|
* @returns a frozen array of the selected deeply frozen events.
|
|
*/
|
|
snapshotEvents(
|
|
fromSeq: SessionLogOffset = SessionLogOffset(0),
|
|
toSeqExclusive: SessionLogOffset = this.seq,
|
|
): readonly SessionEvent[] {
|
|
if (fromSeq === 0 && toSeqExclusive === this.log.length) {
|
|
this.eventsSnapshot ??= Object.freeze([...this.log])
|
|
return this.eventsSnapshot
|
|
}
|
|
return Object.freeze(this.log.slice(fromSeq, toSeqExclusive))
|
|
}
|
|
|
|
/**
|
|
* Return this Session's events after its fork-inherited prefix.
|
|
* @deprecated Existing logic may remain unmigrated for now, but new calls are prohibited.
|
|
* See the [Agent Note](../../../../.agents/notes/implemented/architecture/2026-09-09-deprecate-synchronous-session-event-reads.md).
|
|
* @returns a fresh array containing child-owned events in log order.
|
|
*/
|
|
ownEvents(): readonly SessionEvent[] {
|
|
// oxlint-disable-next-line typescript/no-deprecated -- Deprecated reader delegates to the deprecated range read.
|
|
return this.snapshotEvents(this.inheritedEventCount)
|
|
}
|
|
|
|
/**
|
|
* Whether one existing event position is outside the fork-inherited prefix.
|
|
* @param seq - event position in this Session.
|
|
* @returns true when the event belongs to this Session rather than its parent.
|
|
*/
|
|
isOwnSeq(seq: SessionSeq): boolean {
|
|
return seq >= this.inheritedEventCount && seq < this.seq
|
|
}
|
|
|
|
/** The next event's sequence number — always the log length (the `seq = log.length` contiguity contract). */
|
|
get seq(): SessionLogOffset {
|
|
return SessionLogOffset(this.log.length)
|
|
}
|
|
|
|
/**
|
|
* Append one typed event to the log and synchronously notify observers via
|
|
* the store-owned, module-private publication hooks. The hot path never blocks
|
|
* on I/O — persistence plugins buffer asynchronously. Once the event enters
|
|
* the log, the append is committed: observer failures are logged and
|
|
* contained per listener, so they do not change the return value or prevent
|
|
* later listeners from observing the same accepted event.
|
|
*
|
|
* @param type - The event type (key of {@link SessionEventMap}).
|
|
* @param data - The event payload; must be JSON-serializable.
|
|
* @param opts - Surface metadata: `surfaceOp` controls how the event enters
|
|
* the ordered surface; `sourceEventSeqs` lists the seq numbers of earlier
|
|
* events this one derives from. REQUIRED for
|
|
* {@link SurfaceEventType} events (every message-producing event must
|
|
* declare how it joins the surface, the sole source of derived model
|
|
* history) and
|
|
* rejected by the compiler for non-surface types like `turn/start` or
|
|
* `assistant/attempt`. Assistant messages embed their exact provider
|
|
* stream and cannot cite top-level source events.
|
|
* @returns the logged event — its assigned `seq`/`time` plus the SNAPSHOT of
|
|
* `data` that entered the log, so reading `event.data` back sees the logged
|
|
* value, never the caller's still-mutable input.
|
|
* @throws if `data` or surface metadata is not losslessly JSON-serializable
|
|
* (BigInt, function, symbol, undefined, negative zero, non-finite number,
|
|
* circular reference, sparse array, or an exotic object such as
|
|
* Map/Set/Date/class instance), or when the candidate violates the
|
|
* request-header empty-field or tool-error consistency rules, or the
|
|
* canonical surface contract (marker shape and eligibility, unique
|
|
* earlier source-event references, positional replacement validity, and complete
|
|
* shadowed-node coverage). One iterative pass reads, validates, and
|
|
* copies each nested value once, so a stateful getter cannot supply one value
|
|
* to validation and another to storage. The event log is the durable source
|
|
* of truth, so a bad event fails at the append site rather than later during
|
|
* a backend flush. A synchronous internal dispatch validation failure or an
|
|
* append reentered while this acceptance/publication boundary is open also
|
|
* rejects before the log changes.
|
|
*/
|
|
append<T extends SessionEventType>(
|
|
type: T,
|
|
data: SessionEventMap[T],
|
|
...opts: T extends SurfaceEventType ? [opts: SurfaceIntent<T>] : []
|
|
): SessionEvent<T> {
|
|
const surfaceOpts: SurfaceIntent | undefined = 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 entry = this.publicationEntry()
|
|
const event = deepFreeze({
|
|
type,
|
|
seq: SessionSeq(this.log.length),
|
|
time: Date.now(),
|
|
data: dataSnapshot,
|
|
...(surfaceMetadataSnapshot as { surfaceOp?: unknown; sourceEventSeqs?: unknown }),
|
|
} as unknown as SessionEvent<T>)
|
|
validateSessionEventData(event, `session event "${type}" at seq ${event.seq}`)
|
|
this.commit(event as SessionEvent, entry)
|
|
return event
|
|
}
|
|
|
|
static {
|
|
commitPluginRecord = (session, type, data) => session.#appendRecord(type, data)
|
|
}
|
|
|
|
/**
|
|
* Commit one plugin record; {@link appendPluginRecord} is the only caller.
|
|
* @param type - record type, checked against the plugin record grammar.
|
|
* @param data - JSON payload, snapshotted before it enters the log.
|
|
* @returns the record's sequence number.
|
|
*/
|
|
#appendRecord(type: PluginRecordType, data: unknown): SessionSeq {
|
|
const dataSnapshot = snapshotJsonValue(data)
|
|
if (dataSnapshot === undefined) {
|
|
throw new Error(`plugin record "${type}" carries non-JSON-serializable data`)
|
|
}
|
|
const entry = this.publicationEntry()
|
|
const record: unknown = deepFreeze({
|
|
type,
|
|
seq: SessionSeq(this.log.length),
|
|
time: Date.now(),
|
|
data: dataSnapshot,
|
|
ignorable: true,
|
|
})
|
|
assertPluginRecordEvent(record)
|
|
this.commit(record, entry)
|
|
return record.seq
|
|
}
|
|
|
|
/**
|
|
* Read the store attachment for one append, refusing an append that
|
|
* reenters while another append is being published.
|
|
* @returns the attachment, or undefined for a detached Session.
|
|
*/
|
|
private publicationEntry(): SessionEntry | undefined {
|
|
const entry = attachments.get(this)
|
|
if (entry?.appending) {
|
|
throw new Error('session append cannot reenter while another append is being published')
|
|
}
|
|
return entry
|
|
}
|
|
|
|
/**
|
|
* Accept one frozen candidate at the next sequence number, enter it into the
|
|
* log, and notify `session/event` observers with per-listener containment.
|
|
* @param event - the candidate built at the current log length.
|
|
* @param entry - the attachment read before the candidate was built.
|
|
*/
|
|
private commit(event: SessionEvent, entry: SessionEntry | undefined): void {
|
|
this.surfaceManager.validateNext(event)
|
|
|
|
if (entry !== undefined) entry.appending = true
|
|
try {
|
|
let callbacks: SessionCallback[] | undefined
|
|
const callbackArgs: unknown[] = [this, event]
|
|
if (entry !== undefined) {
|
|
callbacks = collectSessionCallbacks(entry.emitCtx, [entry.carrier, 'session/event', ...callbackArgs])
|
|
}
|
|
this.log.push(event)
|
|
this.eventsSnapshot = undefined
|
|
if (callbacks !== undefined && entry !== undefined) {
|
|
invokeContainedSessionObservers(entry.emitCtx, 'session/event', entry.id, callbackArgs, callbacks)
|
|
}
|
|
} finally {
|
|
if (entry !== undefined) {
|
|
entry.appending = false
|
|
if (entry.detachRequested && !entry.announcing) entry.detach()
|
|
}
|
|
}
|
|
}
|
|
|
|
/** Cached fold of the request-header events — see {@link requestHeader}. */
|
|
private headerFold: EpochHeader | undefined
|
|
/** Log position (events consumed) the header fold has reached. */
|
|
private headerFoldSeq = 0
|
|
|
|
/**
|
|
* The {@link EpochHeader} in force after the log's last header event — the
|
|
* header the NEXT request will be compared against — or undefined before
|
|
* the first `request/header` snapshot. The live, incrementally-maintained
|
|
* form of `foldRequestHeader(session.snapshotEvents())`: each header event is folded
|
|
* once, when first seen, so a per-step read costs O(new events).
|
|
* @returns the folded header, or undefined when no header event exists yet.
|
|
*/
|
|
requestHeader(): EpochHeader | undefined {
|
|
if (this.headerFoldSeq < this.log.length) {
|
|
// Frozen on update: the fold is session state exposed by reference — a
|
|
// consumer mutating it in place (instead of building a replacement)
|
|
// would desync every later comparison against the log, so mutation
|
|
// throws instead.
|
|
this.headerFold = deepFreeze(foldRequestHeader(this.log.slice(this.headerFoldSeq), this.headerFold))
|
|
this.headerFoldSeq = this.log.length
|
|
}
|
|
return this.headerFold
|
|
}
|
|
|
|
/** Cached fold of `request/context` events. */
|
|
private contextFold: RequestContext | undefined
|
|
private contextFoldSeq = 0
|
|
|
|
/**
|
|
* Return the latest resolved route metadata, or `undefined` before the first
|
|
* `request/context` event. Each event is folded once.
|
|
* @returns the latest immutable route metadata.
|
|
*/
|
|
requestContext(): RequestContext | undefined {
|
|
if (this.contextFoldSeq < this.log.length) {
|
|
for (const event of this.log.slice(this.contextFoldSeq)) {
|
|
if (event.type === 'request/context') this.contextFold = deepFreeze({ ...event.data })
|
|
}
|
|
this.contextFoldSeq = this.log.length
|
|
}
|
|
return this.contextFold
|
|
}
|
|
|
|
/** Cached historical tool definitions and updates for request projection. */
|
|
private readonly toolHistoryProjection = new ToolHistoryProjection()
|
|
/** Index of the next committed event not yet consumed by the tool-history fold. */
|
|
private toolHistorySeq = 0
|
|
|
|
/**
|
|
* Fold unseen committed events into capability-independent tool history.
|
|
* Initial access reconstructs inherited history; later reads consume only new events.
|
|
* @returns an immutable snapshot for LLM request projection, including historical addition definitions.
|
|
*/
|
|
toolHistory(): ToolHistory {
|
|
for (const event of this.log.slice(this.toolHistorySeq)) this.toolHistoryProjection.apply(event)
|
|
this.toolHistorySeq = this.log.length
|
|
return this.toolHistoryProjection.snapshot()
|
|
}
|
|
|
|
/** The derived-message cache: frozen projections, extended per unseen node. */
|
|
private derived: Message[] = []
|
|
/** Surface position (nodes projected) the cache has reached. */
|
|
private derivedNodes = 0
|
|
/** {@link SurfaceManager.contentGeneration} the cache was built under. */
|
|
private derivedGeneration = 0
|
|
|
|
/**
|
|
* Derive the LLM message history by walking the ordered sequences of
|
|
* message-producing events maintained by `surfaceOp` markers. The
|
|
* surface is the single source of derived history: every message-producing
|
|
* append records its `surfaceOp`, so a raw event with no marker (a chunk, a
|
|
* turn boundary) is correctly absent, and a compaction `replace` deletes the
|
|
* shadowed nodes from the derivation. The projection rules are
|
|
* {@link deriveEventMessage}, with logged message projections applied
|
|
* without changing node membership or message identity.
|
|
*
|
|
* CACHED: pure tail growth costs O(new nodes); a replacement or message projection
|
|
* ({@link SessionSurface.contentGeneration}) rebuilds. The returned array is
|
|
* a fresh snapshot per call (later appends never grow an array a caller
|
|
* already holds); the `Message` objects in it are SHARED and **deep-frozen**.
|
|
* Unchanged content reuses frozen event data; projected blocks are frozen
|
|
* derived copies. Consumers cannot mutate the log through either form.
|
|
* @returns a fresh array of the shared, frozen derived history.
|
|
*/
|
|
deriveMessages(): Message[] {
|
|
const surface = this.surface
|
|
const nodes = surface.nodes
|
|
const generation = surface.contentGeneration
|
|
if (generation !== this.derivedGeneration) {
|
|
this.derived = []
|
|
this.derivedNodes = 0
|
|
this.derivedGeneration = generation
|
|
}
|
|
for (const seq of nodes.slice(this.derivedNodes)) {
|
|
// Surface sequences are built from this.log — seq is always a valid
|
|
// index by construction. The non-null assertion expresses that invariant.
|
|
// oxlint-disable-next-line typescript/no-non-null-assertion
|
|
const msg = this.deriveEventMessage(this.log[seq]!)
|
|
// A surface node is one of the five message-producing types, but an
|
|
// empty-content assistant/message (a max-tokens step that hosts only
|
|
// usage) derives to null and must not enter the transcript.
|
|
if (msg) this.derived.push(msg)
|
|
}
|
|
this.derivedNodes = nodes.length
|
|
return [...this.derived]
|
|
}
|
|
|
|
/**
|
|
* Project one event with all committed message projections applied.
|
|
* The original durable event remains unchanged.
|
|
* @param event - the event to project.
|
|
* @returns the derived message, or null when the event produces none.
|
|
*/
|
|
deriveEventMessage(event: SessionEvent): Message | null {
|
|
return this.surfaceManager.deriveEventMessage(event)
|
|
}
|
|
}
|
|
|
|
/** A fork source: either the live session object or its live store id. */
|
|
export type SessionForkSource = Session | SessionId
|
|
|
|
/**
|
|
* Rejection codes for session forking: the fork source id is unknown to the
|
|
* live store (`SESSION_NOT_FOUND`) or names a session object that is not the
|
|
* store's live instance (`SESSION_NOT_LIVE`); the requested child id is
|
|
* already taken (`SESSION_ALREADY_EXISTS`); or the boundary is not a contiguous
|
|
* existing seq (`INVALID_BOUNDARY`).
|
|
*/
|
|
export type SessionForkErrorCode =
|
|
| 'SESSION_NOT_FOUND'
|
|
| 'SESSION_NOT_LIVE'
|
|
| 'SESSION_ALREADY_EXISTS'
|
|
| 'INVALID_BOUNDARY'
|
|
|
|
/** Typed error for session fork rejections. */
|
|
export class SessionForkError extends Error {
|
|
constructor(message: string, public readonly code: SessionForkErrorCode) {
|
|
super(message)
|
|
this.name = 'SessionForkError'
|
|
}
|
|
}
|
|
|
|
/**
|
|
* In-memory session store (`ctx.sessions`).
|
|
*
|
|
* Persistence is intentionally not implemented here — the agent lifecycle
|
|
* attaches a session-log writer to each published session's write handle;
|
|
* a session published outside that lifecycle persists nothing.
|
|
*/
|
|
export class SessionStore extends Service {
|
|
private store = new Map<SessionId, SessionEntry>()
|
|
private counter = 0
|
|
private readonly projections: SessionMessageProjection[] = []
|
|
|
|
/** Borrowed definitions for detached replay; contributions live until their registering fibers unload. */
|
|
get messageProjections(): readonly SessionMessageProjection[] {
|
|
return this.projections
|
|
}
|
|
|
|
/**
|
|
* Register one event interpreter for live creation, restore, and fork.
|
|
* Disposing the contribution makes sessions that used it refuse further derivation.
|
|
* @param projection - pure definition owned by the event's plugin.
|
|
* @returns the fiber-owned disposer.
|
|
* @throws when another definition already owns this event type.
|
|
*/
|
|
registerMessageProjection(projection: SessionMessageProjection): () => Promise<void> {
|
|
if (this.projections.some(item => item.type === projection.type)) {
|
|
throw new Error(`session message projection "${projection.type}" is already registered`)
|
|
}
|
|
return this.ctx.effect(() => {
|
|
this.projections.push(projection)
|
|
return () => { this.projections.splice(this.projections.indexOf(projection), 1) }
|
|
}, 'sessions.registerMessageProjection()')
|
|
}
|
|
|
|
constructor(ctx: Context) {
|
|
super(ctx, 'sessions')
|
|
ctx.inject(['typert'], (typeCtx) => {
|
|
typeCtx.typert.lookups.register('session', {
|
|
parameter: 'session',
|
|
wire: 'sessionId',
|
|
resource: 'session',
|
|
hostTypeSymbol: '@deepseek-ai/dsh-session#Session',
|
|
wireTypeSymbol: '@deepseek-ai/dsh-session/types#SessionId',
|
|
resolve: sessionId => this.get(sessionId),
|
|
})
|
|
})
|
|
}
|
|
|
|
/**
|
|
* Create a session owned by the calling fiber: disposing that fiber stops
|
|
* event notification and removes the session from the store. `options.seed`
|
|
* populates the session with a copy of those events (replay/fork);
|
|
* `options.meta` attaches creation metadata (validated absolute `cwd`, seed
|
|
* and parent lineage, and delegation depth) as the immutable
|
|
* {@link SessionHeader} (the store fills `version`/`id`/`createdAt`).
|
|
*
|
|
* For an agent whose session must be torn down IN ORDER with its loop (so the
|
|
* loop's final events are published before the store attachment ends), do NOT use this
|
|
* — fold the session lifecycle into the agent's own effect via
|
|
* {@link prepare} + {@link enter} + {@link announce} (see
|
|
* `dsh-agent-loop`'s creation transaction).
|
|
*
|
|
* @param id - the session id; omitted, the store mints `session-<n>`.
|
|
* @param options - seed events and/or creation metadata for the header.
|
|
* @returns the live session, already entered and announced.
|
|
* @throws if a session with `id` already exists, metadata is not a plain
|
|
* lossless-JSON record with valid scalar fields, or `meta.cwd` is a
|
|
* non-absolute path (storage backends key directories off it).
|
|
*/
|
|
create(id?: SessionId, options?: CreateSessionOptions): Session {
|
|
const session = this.prepare(id, options)
|
|
// Single effect owned by the calling fiber. Yield the detach BEFORE
|
|
// announcing so a throwing `session/created` listener rolls the attach back
|
|
// (the generator effect disposes already-yielded disposers on a throw)
|
|
// instead of leaking the store entry and its publication hooks.
|
|
this.ctx.effect(function* (this: SessionStore) {
|
|
yield this.enter(session)
|
|
this.announce(session)
|
|
}.bind(this), 'sessions.create()')
|
|
return session
|
|
}
|
|
|
|
/**
|
|
* Build a session WITHOUT entering it into the store — validate the id/cwd and
|
|
* construct the {@link Session} (with its immutable {@link SessionHeader}).
|
|
* Pairs with {@link enter} + {@link announce}: a caller that owns a composite
|
|
* `ctx.effect` (the agent factory) folds the session lifecycle into that ONE
|
|
* effect so a fiber unload tears the session + agent down as a single ORDERED
|
|
* chain rather than as racing sibling effects — which would remove the publication hooks
|
|
* before the driver's closing events commit, dropping them.
|
|
*
|
|
* @param id - the session id; omitted, the store mints `session-<n>`.
|
|
* @param options - seed events and/or creation metadata for the header. With
|
|
* `eventState`, every seed event is either independently owned or any
|
|
* shared value is deeply frozen; {@link Session.fromRestore} validates and
|
|
* adopts those values without copying or freezing them.
|
|
* @returns the constructed session, NOT yet in the store.
|
|
* @throws if a session with `id` already exists, metadata is not a plain
|
|
* lossless-JSON record with valid scalar fields, or `meta.cwd` is a
|
|
* non-absolute path.
|
|
*/
|
|
prepare(id?: SessionId, options?: PrepareSessionOptions): Session {
|
|
let sessionId: SessionId
|
|
if (id === undefined) {
|
|
do sessionId = brandString<SessionId>(`session-${++this.counter}`)
|
|
while (this.store.has(sessionId))
|
|
} else {
|
|
sessionId = brandString<SessionId>(id)
|
|
}
|
|
if (this.store.has(sessionId)) throw new Error(`session "${sessionId}" already exists`)
|
|
if (options !== undefined) {
|
|
const { eventState } = options
|
|
switch (eventState) {
|
|
case 'detached':
|
|
case 'shared-frozen':
|
|
return Session.fromRestore(
|
|
sessionId,
|
|
options.seed,
|
|
options.meta,
|
|
options.inheritedEventCount,
|
|
eventState,
|
|
this.projections,
|
|
)
|
|
case undefined:
|
|
break
|
|
/* v8 ignore next -- closed-union exhaustiveness guard */
|
|
default:
|
|
assertNever(eventState, 'SessionStore.prepare event state')
|
|
}
|
|
}
|
|
const seed = options?.seed
|
|
const meta = options?.meta
|
|
const header: SessionHeader = {
|
|
version: SESSION_FORMAT_VERSION,
|
|
id: sessionId,
|
|
createdAt: meta?.createdAt ?? Date.now(),
|
|
...meta?.cwd === undefined ? {} : { cwd: meta.cwd },
|
|
...meta?.parentSession === undefined ? {} : { parentSession: meta.parentSession },
|
|
isSeeded: meta?.isSeeded ?? false,
|
|
...meta?.origin === undefined ? {} : { origin: meta.origin },
|
|
...meta?.delegationDepth === undefined ? {} : { delegationDepth: meta.delegationDepth },
|
|
...meta?.agentPreset === undefined ? {} : { agentPreset: meta.agentPreset },
|
|
}
|
|
return Session.create(sessionId, seed, header, options?.inheritedEventCount, this.projections)
|
|
}
|
|
|
|
/**
|
|
* Enter a {@link prepare}d session into the store: install the module-private
|
|
* append publication hooks and add it to the store. Returns the DETACH
|
|
* disposer (hooks + store removal). Does NOT emit `session/created` —
|
|
* the caller yields this disposer inside its effect and THEN calls
|
|
* {@link announce}, so a throwing `session/created` listener rolls the attach
|
|
* back instead of leaking it.
|
|
*
|
|
* Re-checks the id for a duplicate: `prepare` and `enter` are public
|
|
* cross-package primitives and a caller may interleave arbitrary work (or
|
|
* another create) between them, so a stale prepared session must NOT overwrite
|
|
* a live store entry of the same id — its detach disposer would later delete
|
|
* the REAL session. The {@link create} convenience and the agent factory call
|
|
* the two back-to-back so they never trip this, but the public API cannot
|
|
* assume that.
|
|
*
|
|
* @param session - a {@link prepare}d session not yet in the store.
|
|
* @returns the detach disposer (publication hooks + store removal). When called from
|
|
* a synchronous `session/created` listener, removal and disposal wait until
|
|
* that creation dispatch unwinds.
|
|
* @throws if a session with this id is already in the store.
|
|
*/
|
|
enter(session: Session): () => void {
|
|
const id = session.id
|
|
const carrier = scopeTarget(session, scopeOf(this.ctx))
|
|
// This is the authoritative collision boundary after arbitrary unpublished
|
|
// preparation. Only one exact same-id transaction can publish.
|
|
if (this.store.has(id)) throw new Error(`session "${id}" already exists`)
|
|
if (attachments.has(session)) throw new Error(`session "${id}" is already attached to a store`)
|
|
const entry: SessionEntry = {
|
|
id,
|
|
session,
|
|
carrier,
|
|
emitCtx: this.ctx,
|
|
announced: false,
|
|
announcing: false,
|
|
appending: false,
|
|
detachRequested: false,
|
|
detach: () => { this.detachEntered(entry) },
|
|
}
|
|
this.store.set(id, entry)
|
|
attachments.set(session, entry)
|
|
let entered = true
|
|
const detach = (): void => {
|
|
if (!entered) return
|
|
entered = false
|
|
// A lifecycle listener may own the advanced detach capability. Keep the
|
|
// entry and its publication hooks live until synchronous creation or append
|
|
// publication unwinds, then publish the paired disposal edge.
|
|
if (entry.announcing || entry.appending) {
|
|
entry.detachRequested = true
|
|
return
|
|
}
|
|
entry.detach()
|
|
}
|
|
return detach
|
|
}
|
|
|
|
/** Remove one exact entered session and emit its paired disposal when announced. */
|
|
private detachEntered(entry: SessionEntry): void {
|
|
entry.detachRequested = false
|
|
// A stale capability cannot remove observers or storage belonging to a
|
|
// later same-id lifecycle.
|
|
/* v8 ignore next -- enter() rejects replacement while this single-shot detach capability is live. */
|
|
if (this.store.get(entry.id) !== entry) return
|
|
this.store.delete(entry.id)
|
|
attachments.delete(entry.session)
|
|
if (entry.announced) this.emitDisposed(entry)
|
|
}
|
|
|
|
/** Emit `session/created` exactly once for an {@link enter}ed session (with
|
|
* the carrier {@link enter} captured). Separate from {@link enter} so the
|
|
* caller can yield the detach disposer first (rollback safety — see
|
|
* {@link enter}).
|
|
* @param session - the entered session to announce to listeners.
|
|
* @throws if the session is not live or its announcement already began,
|
|
* including a reentrant call from a creation listener. */
|
|
announce(session: Session): void {
|
|
const entry = this.liveEntryFor(session)
|
|
if (entry.announced || entry.announcing) {
|
|
throw new Error(`session "${entry.id}" was already announced`)
|
|
}
|
|
// Mark before emit: Cordis emit may deliver to earlier listeners and then
|
|
// throw. Rollback must still pair that partial creation with disposal, and
|
|
// a listener cannot recursively create a second lifecycle edge.
|
|
entry.announced = true
|
|
const callbackArgs: unknown[] = [session]
|
|
entry.announcing = true
|
|
try {
|
|
const callbacks = collectSessionCallbacks(this.ctx, [entry.carrier, 'session/created', session])
|
|
for (const callback of callbacks) {
|
|
// Synchronous throws intentionally propagate and veto publication; the
|
|
// yielded detach then emits the paired disposal edge. An async function
|
|
// is nevertheless assignable to a void listener, so observe its returned
|
|
// promise: rejection is too late to roll back and must be logged instead
|
|
// of becoming unhandled.
|
|
const returned: unknown = callback(...callbackArgs)
|
|
void Promise.resolve(returned).catch((error: unknown) => {
|
|
this.ctx.logger.warn(`session "${entry.id}": session/created listener rejected: ${String(error)}`)
|
|
})
|
|
}
|
|
} finally {
|
|
entry.announcing = false
|
|
if (entry.detachRequested && !entry.appending) entry.detach()
|
|
}
|
|
}
|
|
|
|
/** Emit the paired teardown notification with per-listener containment. */
|
|
private emitDisposed(entry: SessionEntry): void {
|
|
const callbackArgs: unknown[] = [entry.session]
|
|
try {
|
|
const callbacks = collectSessionCallbacks(this.ctx, [entry.carrier, 'session/disposed', entry.session])
|
|
invokeContainedSessionObservers(this.ctx, 'session/disposed', entry.id, callbackArgs, callbacks)
|
|
} catch (error: unknown) {
|
|
this.ctx.logger.warn(`session "${entry.id}": session/disposed dispatch threw: ${String(error)}`)
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Dispatch the awaited `session/flush` durability checkpoint for `session`,
|
|
* with the carrier captured at {@link enter}. THE flush entry point: the
|
|
* store owns the carrier, so callers (the checkpoint policy's per-request
|
|
* barrier, goal-round-driver's idle checkpoint, teardown drains, and consumers
|
|
* that flush themselves before reading storage) must come through here
|
|
* rather than dispatch a raw `ctx.parallel('session/flush', …)` — one owner
|
|
* and one spelling.
|
|
* @param session - the session whose buffered events must reach durable storage.
|
|
* @returns whether at least one durability listener participated, after every
|
|
* listener has settled successfully.
|
|
* @throws the first registered listener failure after every listener settles.
|
|
*/
|
|
async flush(session: Session): Promise<boolean> {
|
|
const { carrier } = this.liveEntryFor(session)
|
|
const callbackArgs: unknown[] = [session]
|
|
const callbacks = collectSessionCallbacks(this.ctx, [carrier, 'session/flush', session])
|
|
const results = await Promise.allSettled(callbacks.map((callback) => {
|
|
try {
|
|
return callback(...callbackArgs)
|
|
} catch (error: unknown) {
|
|
// Preserve the listener's exact rejection value; flush is a caller-owned
|
|
// failure boundary, and Cordis listeners may throw arbitrary values.
|
|
// oxlint-disable-next-line typescript/prefer-promise-reject-errors
|
|
return Promise.reject(error)
|
|
}
|
|
}))
|
|
const failure = results.find((result): result is PromiseRejectedResult => result.status === 'rejected')
|
|
if (failure !== undefined) throw failure.reason
|
|
return callbacks.length > 0
|
|
}
|
|
|
|
/** Return the exact live entry; detached/prepared objects reject. */
|
|
private liveEntryFor(session: Session): SessionEntry {
|
|
const entry = attachments.get(session)
|
|
if (entry === undefined || this.store.get(entry.id) !== entry) {
|
|
throw new Error(`session "${session.id}" is not live in this store`)
|
|
}
|
|
return entry
|
|
}
|
|
|
|
/**
|
|
* Look up a live session.
|
|
* @param id - the session id to look up.
|
|
* @returns the session, or undefined when no live session has that id.
|
|
*/
|
|
get(id: SessionId): Session | undefined {
|
|
return this.store.get(id)?.session
|
|
}
|
|
|
|
/**
|
|
* All live sessions, in creation order.
|
|
* @returns a fresh array; mutating it does not affect the store.
|
|
*/
|
|
list(): Session[] {
|
|
return [...this.store.values()].map(entry => entry.session)
|
|
}
|
|
|
|
/**
|
|
* Create a live child session from an exact prefix of a live source.
|
|
* `boundary` is an inclusive source event seq; omitted means the source's
|
|
* current last event. An open tail receives synthetic tool results and
|
|
* step/turn closers with the forked cause. Closed steps and turns remain
|
|
* unchanged, including any failed tool calls already missing results.
|
|
* `inheritedEventCount` counts only copied source events, excluding these closers.
|
|
*
|
|
* @param source - Live source session object or id.
|
|
* @param boundary - Inclusive source event seq to fork through; omitted means
|
|
* the source's current last event, and omitted on an empty source forks an
|
|
* empty child.
|
|
* @param childSessionId - Optional child session id; omitted delegates to
|
|
* `SessionStore`'s id policy.
|
|
* @returns The created live child session.
|
|
*/
|
|
fork(source: SessionForkSource, boundary?: SessionSeq, childSessionId?: SessionId): Session {
|
|
if (childSessionId !== undefined && this.get(childSessionId) !== undefined) {
|
|
throw new SessionForkError(`session "${childSessionId}" already exists`, 'SESSION_ALREADY_EXISTS')
|
|
}
|
|
const liveSource = this._resolveForkSource(source)
|
|
// oxlint-disable-next-line typescript/no-deprecated -- Existing fork snapshot read; migration deferred.
|
|
const events = liveSource.snapshotEvents()
|
|
const resolved = this._forkBoundary(liveSource.id, events, boundary)
|
|
const seed = resolved === undefined ? [] : buildForkSeed(events, resolved)
|
|
return this.create(childSessionId, {
|
|
seed,
|
|
inheritedEventCount: SessionLogOffset(resolved === undefined ? 0 : resolved + 1),
|
|
meta: {
|
|
...liveSource.header.cwd !== undefined ? { cwd: liveSource.header.cwd } : {},
|
|
parentSession: liveSource.id,
|
|
isSeeded: true,
|
|
},
|
|
})
|
|
}
|
|
|
|
private _forkBoundary(
|
|
sessionId: SessionId, events: readonly SessionEvent[], requestedBoundary: SessionSeq | undefined,
|
|
): SessionSeq | undefined {
|
|
const lastEvent = events.at(-1)
|
|
let boundary: SessionSeq
|
|
if (requestedBoundary !== undefined) {
|
|
boundary = requestedBoundary
|
|
} else {
|
|
if (lastEvent === undefined) return undefined
|
|
boundary = lastEvent.seq
|
|
}
|
|
if (!Number.isSafeInteger(boundary) || boundary < 0) {
|
|
throw new SessionForkError(
|
|
`fork boundary for session "${sessionId}" must be a non-negative safe integer, got ${String(boundary)}`,
|
|
'INVALID_BOUNDARY',
|
|
)
|
|
}
|
|
if (boundary >= events.length) {
|
|
const lastSeq = lastEvent?.seq
|
|
throw new SessionForkError(
|
|
`fork boundary ${boundary} does not exist in session "${sessionId}" (last seq: ${lastSeq ?? 'none'})`,
|
|
'INVALID_BOUNDARY',
|
|
)
|
|
}
|
|
|
|
const boundaryEvent = events[boundary]
|
|
if (boundaryEvent === undefined || boundaryEvent.seq !== boundary) {
|
|
throw new SessionForkError(
|
|
`fork boundary ${boundary} does not match a contiguous event seq in session "${sessionId}"`,
|
|
'INVALID_BOUNDARY',
|
|
)
|
|
}
|
|
return boundary
|
|
}
|
|
|
|
private _resolveForkSource(source: SessionForkSource): Session {
|
|
if (typeof source === 'string') {
|
|
const session = this.get(source)
|
|
if (session === undefined) throw new SessionForkError(`session "${source}" not found`, 'SESSION_NOT_FOUND')
|
|
return session
|
|
}
|
|
|
|
const live = this.get(source.id)
|
|
if (live === undefined) {
|
|
throw new SessionForkError(`session "${source.id}" not found`, 'SESSION_NOT_FOUND')
|
|
}
|
|
if (live !== source) throw new SessionForkError(`session "${source.id}" is not the live store instance`, 'SESSION_NOT_LIVE')
|
|
return source
|
|
}
|
|
|
|
}
|
|
|
|
export { decodeSeqRanges, encodeSeqRanges } from './seq-ranges.ts'
|
|
export default SessionStore
|