refactor(session): open journal streams from snapshots

This commit is contained in:
imccyu
2026-08-25 06:07:11 +08:00
parent 7fb2ca07e4
commit e7952d82ed
15 changed files with 938 additions and 601 deletions
@@ -29,7 +29,11 @@ export interface SessionSnapshot {
readonly sessionId: SessionId
readonly queue: readonly QueuedMessage[]
readonly running: boolean
readonly subagent: { readonly address: SubagentAddress; readonly parentAvailable: boolean } | null
readonly subagent: {
readonly address: SubagentAddress
/** Absent until the direct-parent catalog resolves. */
readonly parentAvailable?: boolean
} | null
readonly removed: boolean
readonly openState: OpenState
readonly openError: ClientFailure | null
@@ -45,7 +45,7 @@ export const PAGE_MESSAGES = 50
export interface SessionOptions {
/** Catalog-discovered address selecting non-activating subagent transport. */
address?: SubagentAddress
/** Whether the exact direct parent Agent was live at the latest catalog read. */
/** Whether the exact direct parent Agent was live at the latest catalog read; absent before that read. */
parentAvailable?: boolean
/**
* First ACCEPTED prompt on a blank session (fires at most once, on the
@@ -86,7 +86,7 @@ export class Session implements SessionFace {
private readonly queueMirror = new SessionQueueMirror()
private running = false
private address: SubagentAddress | undefined
private parentAvailable = false
private parentAvailable: boolean | undefined
/**
* Sticky send marker, private input of the composerPhase derivation: set
* synchronously before prompt()'s first await, never reset — the blank →
@@ -144,7 +144,7 @@ export class Session implements SessionFace {
) {
this.projections = options.projections ?? new ProjectionValueStore()
this.address = options.address
this.parentAvailable = options.parentAvailable ?? false
this.parentAvailable = options.parentAvailable
this.notifier = new Notifier(() => {
this.snapshotCache = this.buildSnapshot()
})
@@ -467,9 +467,9 @@ export class Session implements SessionFace {
* Install or clear the catalog-discovered transport address. A changed
* address rebuilds an already-open window through its new history route.
* @param address - direct parent/child address, or undefined for ordinary transport.
* @param parentAvailable - latest exact-parent availability hint.
* @param parentAvailable - latest exact-parent availability hint, or undefined before a catalog read.
*/
configureSubagent(address: SubagentAddress | undefined, parentAvailable = false): void {
configureSubagent(address: SubagentAddress | undefined, parentAvailable?: boolean): void {
const same = this.address?.parentSessionId === address?.parentSessionId
&& this.address?.childSessionId === address?.childSessionId
&& this.address?.mode === address?.mode
@@ -623,7 +623,10 @@ export class Session implements SessionFace {
running: this.running,
subagent: this.address === undefined
? null
: { address: this.address, parentAvailable: this.parentAvailable },
: {
address: this.address,
...(this.parentAvailable === undefined ? {} : { parentAvailable: this.parentAvailable }),
},
removed: this.removed,
openState: this.openState,
openError: this.openError,
@@ -17,6 +17,7 @@ import type {
SessionEventEntry,
SessionPage,
SessionPageRequest,
SessionProjectionBaseline,
} from '../types.ts'
export {
@@ -30,8 +31,13 @@ export type ClientSessionPageRequest = Omit<SessionPageRequest, 'address' | 'thr
/** Complete generated `ctx.remote.session` namespace. */
export type SessionRemote = ClientRemote['session']
/** Opening metadata carried only by a follow snapshot, never by loadOlder pages. */
interface SessionJournalPage extends SessionPage {
readonly projections?: SessionProjectionBaseline
}
/** One complete publication from the Session journal stream. */
export type SessionJournalChange = RemoteJournalChange<SessionPage, SessionEventEntry>
export type SessionJournalChange = RemoteJournalChange<SessionJournalPage, SessionEventEntry>
type SessionControlBaselineFrame = Extract<SessionControlFrame, { type: 'baseline' }>
type SessionControlDeltaFrame = Exclude<SessionControlFrame, SessionControlBaselineFrame>
@@ -93,7 +99,7 @@ export function createSessionControlStream(
/** Gateway-owned event journal bound to one ordinary or direct-subagent Session address. */
export class SessionEventStream extends RemoteJournalStream<
SessionPage,
SessionJournalPage,
SessionEventEntry,
number,
ClientSessionPageRequest
@@ -126,15 +132,23 @@ export class SessionEventStream extends RemoteJournalStream<
/** @inheritdoc */
protected override async * follow(
afterSeq: number | undefined,
request: ClientSessionPageRequest,
signal: AbortSignal,
): AsyncIterable<RemoteJournalFrame<SessionEventEntry, number>> {
const request = afterSeq === undefined
? { address: this.address }
: { address: this.address, afterSeq }
for await (const frame of this.remote.session.follow(request, signal)) {
if (frame.type === 'opened') {
yield frame
): AsyncIterable<RemoteJournalFrame<SessionEventEntry, number, SessionJournalPage>> {
for await (const frame of this.remote.session.follow({
address: this.address,
...(request.maxMessages === undefined ? {} : { maxMessages: request.maxMessages }),
}, signal)) {
if (frame.type === 'snapshot') {
yield {
type: 'opened',
cursor: frame.cursor,
page: {
events: frame.events,
hasMore: frame.hasMore,
projections: frame.projections,
},
}
continue
}
const { type: _type, ...entry } = frame
@@ -147,7 +161,7 @@ export class SessionEventStream extends RemoteJournalStream<
request: ClientSessionPageRequest,
throughSeq: number,
signal: AbortSignal,
): Promise<SessionPage> {
): Promise<SessionJournalPage> {
const result = await this.remote.session.page(
{ address: this.address, throughSeq, ...request },
signal,
+101 -98
View File
@@ -2,9 +2,9 @@
import type { Context } from '@deepseek-ai/cordis'
import { isAppendSurfaceEvent } from '@deepseek-ai/dsh-session'
import type { Session, SessionEvent, SessionHeader, SessionId } from '@deepseek-ai/dsh-session'
import type { SessionInspection } from '@deepseek-ai/dsh-session-persistence'
import { foldSubagentDescriptor } from '@deepseek-ai/dsh-subagent'
import type { SessionEvent, SessionHeader, SessionId } from '@deepseek-ai/dsh-session'
import { SessionQueryError, type SessionObservation } from '@deepseek-ai/dsh-session-query'
import type {} from '@deepseek-ai/dsh-subagent'
import { TypertRemoteFailure } from '@deepseek-ai/dsh-typert-protocol'
import type {
SessionAddress,
@@ -13,7 +13,7 @@ import type {
SessionFollowFrame,
SessionPage,
SessionPageRequest,
SessionProjectionsBlock,
SessionProjectionBaseline,
SessionProjectionValues,
SessionWireEvent,
} from './types.ts'
@@ -21,16 +21,18 @@ import type {
const DEFAULT_MAX_MESSAGES = 50
const MESSAGE_TYPES = new Set(['user/message', 'assistant/message'])
type SessionSource =
| { readonly kind: 'attached'; readonly session: Session }
| { readonly kind: 'detached'; readonly header: SessionHeader; readonly events: readonly SessionEvent[] }
/** Implements cold-safe history operations delegated by the Session Controller. */
export class SessionHistoryController {
private readonly closeFollowers = new Set<() => void>()
/** @param ctx - Host context carrying Session, persistence, and projection services. */
constructor(private readonly ctx: Context) {
/**
* @param ctx - Host context carrying Session query and projection services.
* @param promote - starts ordinary Session activation after snapshot delivery.
*/
constructor(
private readonly ctx: Context,
private readonly promote: (observation: SessionObservation) => void,
) {
ctx.effect(() => () => {
for (const close of this.closeFollowers) close()
this.closeFollowers.clear()
@@ -41,13 +43,13 @@ export class SessionHistoryController {
* Read one message-aligned history page without activating an Agent.
* @param request - durable address and backwards-page cursor.
* @param signal - caller cancellation for persistence reads.
* @returns a contiguous event page and a projection baseline on tail reads.
* @returns a contiguous event page.
*/
async page(request: SessionPageRequest, signal: AbortSignal): Promise<SessionPage> {
validatePageRequest(request)
const source = await this.sourceFor(request.address, signal)
using source = await this.sourceFor(request.address, signal, false)
signal.throwIfAborted()
const sourceLog = sourceEvents(source)
const sourceLog = source.events
const sourceCursor = sourceLog.at(-1)?.seq ?? -1
if (request.throughSeq > sourceCursor) {
reject(
@@ -56,19 +58,20 @@ export class SessionHistoryController {
{},
)
}
const events = sourceLog.filter(event => event.seq <= request.throughSeq)
if ((events.at(-1)?.seq ?? -1) !== request.throughSeq) {
/* v8 ignore next -- Session and persistence validation guarantee a dense zero-based event prefix. */
if (request.throughSeq >= 0 && sourceLog[request.throughSeq]?.seq !== request.throughSeq) {
reject('internal', `session log does not contain through seq ${String(request.throughSeq)}`, {})
}
const page = paginate(events, request.beforeSeq, request.maxMessages ?? DEFAULT_MAX_MESSAGES)
const page = paginate(
sourceLog,
request.beforeSeq,
request.maxMessages ?? DEFAULT_MAX_MESSAGES,
request.throughSeq,
)
const entries = page.events.map(entryFor)
const projections = request.beforeSeq === undefined
? this.projectionsFor(request.address, source, events)
: undefined
return {
events: entries,
hasMore: page.hasMore,
...(projections === undefined ? {} : { projections }),
}
}
@@ -76,13 +79,14 @@ export class SessionHistoryController {
* Follow events appended after an initial cursor on one durable address.
* @param request - durable address and last committed sequence already held by the caller.
* @param signal - stream cancellation owned by the Remote carrier.
* @returns an opened cursor followed by gap-free event frames.
* @returns a complete opening snapshot followed by gap-free event frames.
*/
async *follow(request: SessionFollowRequest, signal: AbortSignal): AsyncIterable<SessionFollowFrame> {
validateFollowRequest(request)
const { address, afterSeq } = request
const { address } = request
const target = addressId(address)
const buffered: SessionEvent[] = []
let snapshotCursor: number | undefined
let wake: (() => void) | undefined
const notify = (): void => {
const resume = wake
@@ -102,35 +106,44 @@ export class SessionHistoryController {
}, { global: true })
const disposeCreated = this.ctx.on('session/created', (session) => {
if (session.id !== target) return
// Session construction appends session/end-seed before attachment, so the
// marker has no session/event notification. Earlier session/created listeners
// may publish later setup events first; this suffix must precede those notifications.
const suffix = session.events.slice(session.firstLiveSeq)
// Constructor seed events have no session/event notification. Normally
// only the end-seed suffix is new; if persistence advanced after the
// opening observation, replay everything beyond that snapshot cursor.
const suffix = session.events.slice(snapshotCursor === undefined
? session.firstLiveSeq
: snapshotCursor + 1)
buffered.unshift(...suffix)
notify()
}, { global: true })
const onAbort = (): void => { notify() }
signal.addEventListener('abort', onAbort, { once: true })
try {
const source = await this.sourceFor(address, signal)
const events = [...sourceEvents(source)]
using source = await this.sourceFor(address, signal, true)
const events = source.events
signal.throwIfAborted()
const cursor = events.at(-1)?.seq ?? -1
if (afterSeq !== undefined && afterSeq > cursor) {
reject('bad-request', `session event resume seq ${String(afterSeq)} is past cursor ${String(cursor)}`, {})
const cursor = source.cursor
snapshotCursor = cursor
const page = paginate(events, undefined, request.maxMessages ?? DEFAULT_MAX_MESSAGES)
yield {
type: 'snapshot',
header: source.header,
cursor,
events: page.events.map(entryFor),
hasMore: page.hasMore,
projections: source.projections === undefined
? { asOfSeq: cursor, values: {} }
: projectionBlock(source.projections),
}
let nextSeq = (afterSeq ?? cursor) + 1
yield { type: 'opened', cursor }
if (afterSeq !== undefined) {
for (const event of events) {
if (event.seq < nextSeq) continue
if (event.seq !== nextSeq) {
reject('internal', `session event replay skipped seq ${String(nextSeq)}`, {})
}
nextSeq++
yield { type: 'event', ...entryFor(event) }
if (address.kind === 'session' && source.source === 'prepared') {
const promotion = source.retain()
try {
this.promote(promotion)
} catch (error: unknown) {
promotion[Symbol.dispose]()
throw error
}
}
let nextSeq = cursor + 1
while (!follower.closed && !signal.aborted) {
const item = buffered.shift()
if (item === undefined) {
@@ -152,49 +165,44 @@ export class SessionHistoryController {
}
}
private async sourceFor(address: SessionAddress, signal: AbortSignal): Promise<SessionSource> {
private async sourceFor(
address: SessionAddress,
signal: AbortSignal,
withProjections: boolean,
): Promise<SessionObservation> {
const sessionId = addressId(address)
const attached = this.ctx.sessions.get(sessionId)
if (attached !== undefined) {
validateAddress(address, attached.header, attached.events)
return { kind: 'attached', session: attached }
try {
const observation = await this.ctx.sessionQuery.observeSession(sessionId, {
signal,
projectionMode: withProjections || address.kind === 'subagent' ? 'all' : 'none',
})
if (observation.header.cwd === undefined) {
observation[Symbol.dispose]()
rejectNotFound(address)
}
try {
validateAddress(address, observation.header, observation.projections)
} catch (error: unknown) {
observation[Symbol.dispose]()
throw error
}
return observation
} catch (error: unknown) {
if (error instanceof SessionQueryError
&& error.code === 'SESSION_QUERY_SESSION_NOT_FOUND') rejectNotFound(address)
throw error
}
const persistence = this.ctx.get('sessionPersistence')
if (persistence === undefined) {
reject('internal', 'session persistence is not configured', {})
}
signal.throwIfAborted()
const header = (await persistence.list(signal)).find(candidate => candidate.id === sessionId)
if (header === undefined || header.cwd === undefined) rejectNotFound(address)
const inspected: SessionInspection = await persistence.inspect(sessionId, signal)
signal.throwIfAborted()
if (inspected.meta.cwd === undefined) rejectNotFound(address)
validateAddress(address, inspected.meta, inspected.events)
return { kind: 'detached', header: inspected.meta, events: inspected.events }
}
private projectionsFor(
address: SessionAddress,
source: SessionSource,
events: readonly SessionEvent[],
): SessionProjectionsBlock | undefined {
const registry = this.ctx.get('sessionProjections')
if (registry === undefined) return undefined
try {
const throughSeq = events.at(-1)?.seq ?? -1
const snapshot = source.kind === 'attached' && source.session.seq - 1 === throughSeq
? registry.snapshot(source.session)
: registry.restore({}, events, 0).snapshot
return {
asOfSeq: snapshot.asOfSeq,
// Projection definitions validate whole JSON values before snapshot publication.
values: snapshot.values as SessionProjectionValues,
}
} catch (error) {
if (address.kind === 'session') throw error
this.ctx.logger.warn(`session.page: projections for "${address.childSessionId}" failed: ${String(error)}`)
return undefined
}
}
function projectionBlock(
snapshot: NonNullable<SessionObservation['projections']>,
): SessionProjectionBaseline {
return {
asOfSeq: snapshot.asOfSeq,
// Projection definitions validate whole JSON values before snapshot publication.
values: snapshot.values as SessionProjectionValues,
}
}
@@ -213,9 +221,9 @@ function validatePageRequest(request: SessionPageRequest): void {
}
function validateFollowRequest(request: SessionFollowRequest): void {
if (request.afterSeq !== undefined
&& (!Number.isSafeInteger(request.afterSeq) || request.afterSeq < -1)) {
reject('bad-request', 'afterSeq must be an integer greater than or equal to -1', {})
if (request.maxMessages !== undefined
&& (!Number.isSafeInteger(request.maxMessages) || request.maxMessages <= 0)) {
reject('bad-request', 'maxMessages must be a positive safe integer', {})
}
}
@@ -226,7 +234,7 @@ function addressId(address: SessionAddress): SessionId {
function validateAddress(
address: SessionAddress,
header: SessionHeader,
events: readonly SessionEvent[],
projections: SessionObservation['projections'],
): void {
if (address.kind === 'session') {
if (header.origin === 'subagent') {
@@ -241,24 +249,22 @@ function validateAddress(
childSessionId: address.childSessionId,
})
}
let descriptor
try {
descriptor = foldSubagentDescriptor(events.slice(header.seedLength ?? 0))
} catch {
const identity = projections?.values.subagent
if (identity === null) {
reject('subagent-catalog-diagnostic', 'subagent descriptor is corrupt', {
parentSessionId: address.parentSessionId,
childSessionId: address.childSessionId,
reason: 'corrupt',
})
}
if (descriptor === undefined) {
if (identity === undefined || identity.seq < (header.seedLength ?? 0)) {
reject('subagent-catalog-diagnostic', 'subagent descriptor is unavailable', {
parentSessionId: address.parentSessionId,
childSessionId: address.childSessionId,
reason: 'unsupported',
})
}
if (descriptor.mode !== address.mode) {
if (identity.mode !== address.mode) {
reject('subagent-unauthorized', 'subagent mode does not match the supplied address', {
childSessionId: address.childSessionId,
})
@@ -279,20 +285,17 @@ function reject(code: string, message: string, details: object): never {
throw new TypertRemoteFailure({ code, message, details })
}
function sourceEvents(source: SessionSource): readonly SessionEvent[] {
return source.kind === 'attached' ? source.session.events : source.events
}
function paginate(
events: readonly SessionEvent[],
beforeSeq: number | undefined,
maxMessages: number,
throughSeq = events.at(-1)?.seq ?? -1,
): { readonly events: SessionEvent[]; readonly hasMore: boolean } {
const window = beforeSeq === undefined ? [...events] : events.filter(event => event.seq < beforeSeq)
const end = Math.min(throughSeq + 1, beforeSeq ?? throughSeq + 1)
let count = 0
let cut = 0
for (let index = window.length - 1; index >= 0; index--) {
const event = window[index] as SessionEvent
for (let index = end - 1; index >= 0; index--) {
const event = events[index] as SessionEvent
if (!MESSAGE_TYPES.has(event.type) || !isAppendSurfaceEvent(event)) continue
count++
const sources = (event as { readonly sourceEventSeqs?: readonly number[] }).sourceEventSeqs
@@ -305,7 +308,7 @@ function paginate(
break
}
}
return { events: window.filter(event => event.seq >= cut), hasMore: cut > 0 }
return { events: events.slice(cut, end), hasMore: cut > 0 }
}
function entryFor(event: SessionEvent): SessionEventEntry {
+59 -21
View File
@@ -6,7 +6,7 @@ import type {
import type { Branded } from '@deepseek-ai/dsh-brand'
import type { MessageId } from '@deepseek-ai/dsh-llm/brand'
import type { ContentBlock } from '@deepseek-ai/dsh-llm/types'
import type { JsonValue, SessionId, SurfaceOp } from '@deepseek-ai/dsh-session/types'
import type { JsonValue, SessionHeader, SessionId, SurfaceOp } from '@deepseek-ai/dsh-session/types'
import type { SessionProjectionMap } from '@deepseek-ai/dsh-session-projection/types'
import type { JobId } from '@deepseek-ai/dsh-jobs/brand'
import type { WorkspaceId } from '@deepseek-ai/dsh-workspace/types'
@@ -17,12 +17,26 @@ declare module '@deepseek-ai/dsh-session-projection/types' {
sessionListMetadata: SessionListMetadata
/** Host state for the boot-constant image-limit view. */
imageLimits: null
/** Durable model selection already used by a request and still pending for a later request. */
modelSelection: ModelSelectionProjectionState
}
interface SessionProjectionMap {
/** Persisted facts used to summarize a Session without activating it. */
sessionListMetadata: SessionListMetadata
/** Image-intake limits enforced by the Session prompt endpoint. */
imageLimits: ImageAttachmentLimits
/** Durable model selection already used and selected for the next request. */
modelSelection: ModelSelectionProjection
}
}
declare module '@deepseek-ai/dsh-session/types' {
interface SessionEventMap {
/**
* Complete validated model selection requested for subsequent prompt
* assembly. Log-only: it never enters derived model history.
*/
'model/selection': ModelSelection
}
}
@@ -34,10 +48,17 @@ export interface SessionListMetadata {
readonly lastPromptAt: number | null
}
/** Projection values and the durable event position they represent. */
export interface SessionProjectionsBlock {
/** Every available cached wire value used as partial, possibly stale Session-list hints. */
export interface SessionProjectionHints {
readonly asOfSeq: number
/** Provider-validated values across the merge-extensible projection key space. */
/** Provider-validated values present in the cache; omitted keys remain unknown. */
readonly values: SessionProjectionValues
}
/** Complete projection values at an exact Session event cursor. */
export interface SessionProjectionBaseline {
readonly asOfSeq: number
/** Provider-validated values; omitted keys are absent capabilities at this cut. */
readonly values: SessionProjectionValues
}
@@ -62,6 +83,22 @@ export interface ModelSelection {
readonly reasoningEffort?: string
}
/** Host fold state for durable model selection. */
export interface ModelSelectionProjectionState {
/** Selection consumed by the latest recorded model request. */
readonly lastUsed: ModelSelection | null
/** Later user selection not yet consumed by a matching model request. */
readonly pending: ModelSelection | null
}
/** Client view of the durable model-selection fold. */
export interface ModelSelectionProjection {
/** Selection consumed by the latest recorded model request. */
readonly lastUsed: ModelSelection | null
/** Selection the next request should use, falling back to {@link lastUsed}. */
readonly next: ModelSelection | null
}
/** One adapter-owned reasoning effort for an exact model route. */
export interface ModelReasoningEffort {
readonly id: string
@@ -97,10 +134,11 @@ export interface ModelCatalogFailure {
readonly message: string
}
/** Detached model-directory snapshot for one Session. */
export interface SessionModels {
readonly current: ModelSelection
readonly routable: boolean
/** Host-generation model catalog and the default used by unconfigured Sessions. */
export interface ModelCatalog {
readonly default: ModelSelection
/** Provider routes currently able to serve a request, including empty catalogs. */
readonly routableProviders: readonly string[]
readonly groups: readonly ModelProviderGroup[]
readonly failures: readonly ModelCatalogFailure[]
}
@@ -120,8 +158,7 @@ export interface SessionSummary {
readonly parentSessionId?: SessionId
readonly origin?: 'subagent'
readonly cwd?: string
readonly agentPreset?: string
readonly projections?: SessionProjectionsBlock
readonly projections?: SessionProjectionHints
}
/** One session-content search result. */
@@ -220,11 +257,6 @@ export interface SessionCreateValue {
readonly agentPreset?: string
}
/** Model-directory request. */
export interface SessionModelsRequest {
readonly sessionId: SessionId
}
/** Session model-selection request. */
export interface SessionSelectModelRequest extends ModelSelection {
readonly sessionId: SessionId
@@ -352,22 +384,28 @@ export interface SessionPageRequest {
readonly maxMessages?: number
}
/** One live event request, optionally resuming after an already-applied event. */
/** One live event request for a durable Session address. */
export interface SessionFollowRequest {
readonly address: SessionAddress
readonly afterSeq?: number
readonly maxMessages?: number
}
/** One contiguous backwards page of a Session log. */
export interface SessionPage {
readonly events: readonly SessionEventEntry[]
readonly hasMore: boolean
readonly projections?: SessionProjectionsBlock
}
/** Initial cursor followed by ordered events appended after that cursor. */
/** Complete opening window followed by ordered events appended after its cursor. */
export type SessionFollowFrame =
| { readonly type: 'opened'; readonly cursor: number }
| {
readonly type: 'snapshot'
readonly header: SessionHeader
readonly cursor: number
readonly events: readonly SessionEventEntry[]
readonly hasMore: boolean
readonly projections: SessionProjectionBaseline
}
| ({ readonly type: 'event' } & SessionEventEntry)
/** One pending inbox occurrence in the authoritative queue snapshot. */
@@ -396,7 +434,7 @@ export interface SessionJob {
export interface SessionControlBaseline {
readonly queues: Readonly<Record<SessionId, readonly SessionQueuedItem[]>>
readonly jobs: Readonly<Record<SessionId, readonly SessionJob[]>>
readonly projections: Readonly<Record<SessionId, SessionProjectionsBlock>>
readonly projections: Readonly<Record<SessionId, SessionProjectionBaseline>>
}
/** One finished projection value and its durable watermark. */