mward4/deepseek-harness
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 29 Wiki Activity

feat(session): split history, state and web controllers

Browse Source 
History, State and Web become ordinary classes sharing the Session scope.
Subscribe-before-read removes the tail repair, events and page expose the
cursor-bearing history read, observeEvent keeps a pure event subscription,
and follow stays passive unless activate is requested.
This commit is contained in:
imccyu committed 2026-10-25 18:14:18 +08:00
1 parent a1473ddec6
commit 2f2caf88b3
13 files changed
+1325 -365
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

packages/api/session-controller/package.json
+1
View File
@@ -73,6 +73,7 @@
"dependencies": {
"@deepseek-ai/dsh-brand": "workspace:*",
"@deepseek-ai/dsh-deque": "workspace:*",
"@deepseek-ai/dsh-timeout": "workspace:*",
"@deepseek-ai/schemastery": "workspace:~",
"mime-types": "^3.0.2",
"zod": "4.4.3"
packages/api/session-controller/src/client/transport.ts
+1
View File
@@ -178,6 +178,7 @@ export class SessionEventStream extends RemoteJournalStream<
let assistantRevision: number | undefined
for await (const frame of this.remote.session.follow({
address: this.address,
activate: true,
assistantStream: true,
...this.repairRequest(request),
}, signal)) {
packages/api/session-controller/src/history.ts
+52 -308
View File
@@ -1,12 +1,8 @@
/** Cold Session history pagination and live-event source. */
/** Shared Session history pagination, address validation and wire conversion. */
import type { Context } from '@deepseek-ai/cordis'
import { Deque } from '@deepseek-ai/dsh-deque'
import type { AssistantStreamFrame } from '@deepseek-ai/dsh-agent'
import {
isAppendSurfaceEvent,
SessionLogOffset,
SessionSeq,
} from '@deepseek-ai/dsh-session'
import type {
SessionEvent,
@@ -15,309 +11,26 @@ import type {
SessionLogOffset as SessionLogOffsetType,
SessionSeqCursor,
} from '@deepseek-ai/dsh-session'
import { SessionQueryError, type SessionObservation } from '@deepseek-ai/dsh-session-query'
import type { SessionObservation } from '@deepseek-ai/dsh-session-query'
import type {} from '@deepseek-ai/dsh-subagent'
import { RemoteError } from '@deepseek-ai/dsh-typert-protocol'
import type { JsonValue } from '@deepseek-ai/dsh-util-values'
import type {
SessionAddress,
SessionAssistantStreamFrame,
SessionEventEntry,
SessionFollowRequest,
SessionFollowFrame,
SessionHistoryRecord,
SessionPage,
SessionPageRequest,
SessionProjectionBaseline,
SessionProjectionValues,
SessionWireHeader,
SessionWireEvent,
} from './types.ts'
import { SessionAssistantStreamAccumulator } from './assistant-stream.ts'
const DEFAULT_MAX_MESSAGES = 50
/** Message budget applied when a history request omits `maxMessages`. */
export const DEFAULT_MAX_MESSAGES = 50
const MESSAGE_TYPES = new Set(['user/message', 'assistant/message'])
/** Implements cold-safe history operations delegated by the Session Controller. */
export class SessionHistoryController {
private readonly closeFollowers = new Set<() => void>()
private readonly assistantStreams = new Map<SessionId, SessionAssistantStreamAccumulator>()
/**
* @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.on('agent/assistant-stream', ({ agent, frame }) => {
let stream = this.assistantStreams.get(agent.session.id)
if (stream === undefined) {
stream = new SessionAssistantStreamAccumulator()
this.assistantStreams.set(agent.session.id, stream)
}
stream.accept(frame, cursorBeforeNext(agent.session.seq))
}, { global: true })
ctx.on('agent/disposed', ({ agent }) => {
this.assistantStreams.delete(agent.session.id)
}, { global: true })
ctx.effect(() => () => {
for (const close of this.closeFollowers) close()
this.closeFollowers.clear()
}, 'session-controller.history')
}
/**
* 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.
*/
async page(request: SessionPageRequest, signal: AbortSignal): Promise<SessionPage> {
validatePageRequest(request)
const throughSeq: SessionSeqCursor = request.throughSeq === -1
? -1
: SessionSeq(request.throughSeq)
const beforeSeq = request.beforeSeq === undefined
? undefined
: SessionLogOffset(request.beforeSeq)
using source = await this.sourceFor(request.address, signal, false)
signal.throwIfAborted()
const sourceLog = source.events
const sourceCursor: SessionSeqCursor = sourceLog.at(-1)?.seq ?? -1
if (throughSeq > sourceCursor) {
throw new RemoteError(
'gateway/bad-request',
`session page through seq ${String(throughSeq)} is past cursor ${String(sourceCursor)}`,
{},
)
}
/* v8 ignore next -- Session and persistence validation guarantee a dense zero-based event prefix. */
if (throughSeq >= 0 && sourceLog[throughSeq]?.seq !== throughSeq) {
throw new RemoteError('gateway/internal', `session log does not contain through seq ${String(throughSeq)}`, {})
}
const page = paginate(
sourceLog,
beforeSeq,
request.maxMessages ?? DEFAULT_MAX_MESSAGES,
throughSeq,
request.turnWindow,
)
const records = pageRecords(page.events)
return {
records,
hasMore: page.hasMore,
}
}
/**
* 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 a complete opening snapshot followed by gap-free durable events and opted-in assistant frames.
*/
async *follow(request: SessionFollowRequest, signal: AbortSignal): AsyncIterable<SessionFollowFrame> {
validateHistoryWindow(request)
const { address } = request
const target = addressId(address)
const buffered = new Deque<
| { readonly type: 'event'; readonly event: SessionEvent }
| {
readonly type: 'assistant-stream'
readonly frame: SessionAssistantStreamFrame
readonly ordinal: number
}
>()
let snapshotCursor: SessionSeqCursor | undefined
let assistantStreamOrdinal = 0
let wake: (() => void) | undefined
const notify = (): void => {
const resume = wake
wake = undefined
resume?.()
}
const follower = { closed: false }
const close = (): void => {
follower.closed = true
notify()
}
this.closeFollowers.add(close)
const disposeEvent = this.ctx.on('session/event', (session, event) => {
if (session.id !== target) return
buffered.pushBack({ type: 'event', event })
notify()
}, { global: true })
const disposeCreated = this.ctx.on('session/created', (session) => {
if (session.id !== target) return
// 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.
// oxlint-disable-next-line typescript/no-deprecated -- Existing Session history read; migration deferred.
const suffix = session.snapshotEvents(snapshotCursor === undefined
? session.firstLiveSeq
: SessionLogOffset(snapshotCursor + 1))
for (let index = suffix.length - 1; index >= 0; index -= 1) {
buffered.pushFront({ type: 'event', event: suffix[index] as SessionEvent })
}
notify()
}, { global: true })
const disposeAssistantStream = request.assistantStream !== true
? undefined
: this.ctx.on('agent/assistant-stream', ({ agent, frame }) => {
if (agent.session.id !== target) return
buffered.pushBack({
type: 'assistant-stream',
frame: wireAssistantStreamFrame(frame, cursorBeforeNext(agent.session.seq)),
ordinal: ++assistantStreamOrdinal,
})
notify()
}, { global: true })
const onAbort = (): void => { notify() }
signal.addEventListener('abort', onAbort, { once: true })
try {
using source = await this.sourceFor(address, signal, true)
const events = source.events
signal.throwIfAborted()
const cursor = source.cursor
snapshotCursor = cursor
const page = paginate(events, undefined, request.maxMessages ?? DEFAULT_MAX_MESSAGES, cursor, request.turnWindow)
const assistantStream = request.assistantStream === true
? this.assistantStreams.get(target)?.snapshot() ?? { revision: 0 }
: undefined
// The accumulator snapshot and this watermark are synchronous. Frames
// through the cut are represented or superseded by that baseline,
// including larger revisions from a retired Agent; later revision
// resets reach Client continuity validation.
const assistantStreamOrdinalCut = assistantStreamOrdinal
yield {
type: 'snapshot',
header: wireHeader(source.header),
cursor,
records: pageRecords(page.events),
hasMore: page.hasMore,
projections: source.projections === undefined
? { asOfSeq: cursor, values: {} }
: projectionBlock(source.projections),
...assistantStream === undefined ? {} : { assistantStream },
}
if (address.kind === 'session' && source.source === 'prepared') {
const promotion = source.retain()
try {
this.promote(promotion)
} catch (error: unknown) {
promotion[Symbol.dispose]()
throw error
}
}
let nextOffset = SessionLogOffset(cursor + 1)
while (!follower.closed && !signal.aborted) {
const item = buffered.popFront()
if (item === undefined) {
await new Promise<void>((resolve) => { wake = resolve })
continue
}
if (item.type === 'assistant-stream') {
if (item.ordinal > assistantStreamOrdinalCut) {
yield { type: 'assistant-stream', frame: item.frame }
}
continue
}
const expectedSeq = SessionSeq(nextOffset)
if (item.event.seq < expectedSeq) continue
if (item.event.seq !== expectedSeq) {
throw new RemoteError('gateway/internal', `session event stream skipped seq ${String(expectedSeq)}`, {})
}
nextOffset = SessionLogOffset(nextOffset + 1)
yield entryFor(item.event)
}
} finally {
this.closeFollowers.delete(close)
signal.removeEventListener('abort', onAbort)
disposeCreated()
disposeEvent()
disposeAssistantStream?.()
}
}
private async sourceFor(
address: SessionAddress,
signal: AbortSignal,
withProjections: boolean,
): Promise<SessionObservation> {
const sessionId = addressId(address)
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.inheritedEventCount,
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
}
}
}
function cursorBeforeNext(nextSeq: SessionLogOffsetType): SessionSeqCursor {
return nextSeq === 0 ? -1 : SessionSeq(nextSeq - 1)
}
function wireAssistantStreamFrame(
frame: AssistantStreamFrame,
durableCursor: SessionSeqCursor,
): SessionAssistantStreamFrame {
if (frame.type === 'start') return { ...frame, startedAfterSeq: durableCursor }
if (frame.type === 'end') return frame
return {
...frame,
chunk: frame.chunk as JsonValue,
}
}
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,
}
}
function validatePageRequest(request: SessionPageRequest): void {
if (!Number.isSafeInteger(request.throughSeq)
|| request.throughSeq < -1
|| Object.is(request.throughSeq, -0)) {
throw new RemoteError('gateway/bad-request', 'throughSeq must be an integer greater than or equal to -1', {})
}
if (request.beforeSeq !== undefined
&& (!Number.isSafeInteger(request.beforeSeq)
|| request.beforeSeq < 0
|| Object.is(request.beforeSeq, -0))) {
throw new RemoteError('gateway/bad-request', 'beforeSeq must be a non-negative safe integer', {})
}
validateHistoryWindow(request)
}
function validateHistoryWindow(request: Pick<SessionPageRequest, 'maxMessages' | 'turnWindow'>): void {
/**
* Reject an invalid message budget or turn window.
* @param request - history window fields.
*/
export function validateHistoryWindow(request: Pick<SessionPageRequest, 'maxMessages' | 'turnWindow'>): void {
if (request.maxMessages !== undefined
&& (!Number.isSafeInteger(request.maxMessages) || request.maxMessages <= 0)) {
throw new RemoteError('gateway/bad-request', 'maxMessages must be a positive safe integer', {})
@@ -334,11 +47,23 @@ function validateHistoryWindow(request: Pick<SessionPageRequest, 'maxMessages' |
}
}
function addressId(address: SessionAddress): SessionId {
/**
* Select the Session whose log an address names.
* @param address - ordinary Session or direct-subagent address.
* @returns the Session whose log the address reads.
*/
export function addressId(address: SessionAddress): SessionId {
return address.kind === 'session' ? address.sessionId : address.childSessionId
}
function validateAddress(
/**
* Reject an address that does not match the observed Session.
* @param address - ordinary Session or direct-subagent address.
* @param header - observed Session header.
* @param inheritedEventCount - observed fork-inherited event count.
* @param projections - observed projections; subagent addresses require them.
*/
export function validateAddress(
address: SessionAddress,
header: SessionHeader,
inheritedEventCount: SessionLogOffsetType,
@@ -379,7 +104,13 @@ function validateAddress(
}
}
function rejectNotFound(address: SessionAddress): never {
/**
* Reject a missing Session using its address-specific error.
* @param address - ordinary Session or direct-subagent address.
* @returns never; this operation always throws.
* @throws RemoteError `session/not-found` or `subagent/not-found`.
*/
export function rejectNotFound(address: SessionAddress): never {
if (address.kind === 'session') {
throw new RemoteError('session/not-found', `session "${address.sessionId}" not found`, { sessionId: address.sessionId })
}
@@ -389,7 +120,16 @@ function rejectNotFound(address: SessionAddress): never {
})
}
function paginate(
/**
* Cut one message-aligned backwards page from a contiguous event prefix.
* @param events - contiguous events starting at seq 0.
* @param beforeSeq - exclusive upper offset of an older page.
* @param maxMessages - message budget.
* @param throughSeq - inclusive log cut.
* @param turnWindow - optional turn-boundary stop rule.
* @returns the page events and whether older events remain.
*/
export function paginate(
events: readonly SessionEvent[],
beforeSeq: SessionLogOffsetType | undefined,
maxMessages: number,
@@ -426,12 +166,12 @@ function paginate(
return { events: events.slice(cut, end), hasMore: cut > 0 }
}
/** Translate current logical Session metadata to the browser wire. */
function wireHeader(header: SessionHeader): SessionWireHeader {
return { ...header }
}
function entryFor(event: SessionEvent): SessionEventEntry {
/**
* Convert a committed event to its wire entry.
* @param event - committed Session event.
* @returns the wire event entry.
*/
export function entryFor(event: SessionEvent): SessionEventEntry {
return {
type: 'event',
// Session.append validates and freezes event data as JSON before publication.
@@ -439,7 +179,11 @@ function entryFor(event: SessionEvent): SessionEventEntry {
}
}
/** Encode one bounded logical page without changing its pagination cut. */
function pageRecords(events: readonly SessionEvent[]): SessionHistoryRecord[] {
/**
* Encode one bounded logical page without changing its pagination cut.
* @param events - page events.
* @returns the wire records.
*/
export function pageRecords(events: readonly SessionEvent[]): SessionHistoryRecord[] {
return events.map(entryFor)
}
packages/api/session-controller/src/history/assistant-stream.ts
+102
View File
@@ -0,0 +1,102 @@
/** Assistant stream baselines and per-connection subscriptions. */
import type { Context } from '@deepseek-ai/cordis'
import { SessionSeq } from '@deepseek-ai/dsh-session'
import type { Session, SessionId } from '@deepseek-ai/dsh-session'
import type { JsonValue } from '@deepseek-ai/dsh-util-values'
import { SessionAssistantStreamAccumulator } from '../assistant-stream.ts'
import type { SessionAssistantStreamBaseline, SessionAssistantStreamFrame } from '../types.ts'
import { ListenerBuffer } from './listener.ts'
/** One accumulator per live Session, shared by every subscription bound to it. */
export class SessionAssistantStreamsListener {
private readonly streams = new Map<SessionId, SessionAssistantStreamAccumulator>()
/** @param ctx - service context owning the accumulator listeners. */
constructor(private readonly ctx: Context) {
ctx.on('agent/assistant-stream', ({ agent, frame }) => {
let stream = this.streams.get(agent.session.id)
if (stream === undefined) {
stream = new SessionAssistantStreamAccumulator()
this.streams.set(agent.session.id, stream)
}
stream.accept(frame, agent.session.seq === 0 ? -1 : SessionSeq(agent.session.seq - 1))
}, { global: true })
ctx.on('agent/disposed', ({ agent }) => { this.streams.delete(agent.session.id) }, { global: true })
ctx.effect(() => () => { this.streams.clear() }, 'session-controller.assistant-streams')
}
/**
* Bind one connection to a Session assistant stream.
* @param sessionId - Session to subscribe.
* @param signal - cancellation owned by the caller.
* @param queue - caller-owned raw notification buffer.
* @returns a listener the caller disposes.
*/
bind(sessionId: SessionId, signal: AbortSignal,
queue: Pick<ListenerBuffer<AssistantEntry>, 'createInput' | 'capture'>): AssistantStreamListener {
return new AssistantStreamListener(this.ctx, sessionId, signal, () => {
return this.streams.get(sessionId)?.snapshot() ?? { revision: 0 }
}, queue)
}
}
type AssistantEntry = { type: 'assistant-stream'; frame: SessionAssistantStreamFrame }
/** An assistant notification and its connection-local arrival position. */
interface AssistantItem {
readonly type: 'assistant-stream'
readonly session: Session
readonly frame: SessionAssistantStreamFrame
readonly ordinal: number
}
/** Writes assistant notifications into the connection's shared queue. */
class AssistantStreamListener implements Disposable {
private readonly disposes: (() => void)[] = []
private ordinal = 0
private cut = 0
constructor(
ctx: Context,
sessionId: SessionId,
signal: AbortSignal,
private _snapshot: () => SessionAssistantStreamBaseline,
output: Pick<ListenerBuffer<AssistantEntry>, 'createInput' | 'capture'>) {
signal.throwIfAborted()
const input = output.createInput<AssistantItem>(item => this.transform(item))
try {
this.disposes.push(ctx.on('agent/assistant-stream', ({ agent, frame }) => {
if (agent.session.id !== sessionId) return
output.capture(() => {
const wire: SessionAssistantStreamFrame = frame.type === 'start'
? { ...frame, startedAfterSeq: agent.session.seq === 0 ? -1 : SessionSeq(agent.session.seq - 1) }
: frame.type === 'chunk' ? { ...frame, chunk: frame.chunk as JsonValue } : frame
input.append({ type: 'assistant-stream', session: agent.session, frame: wire, ordinal: ++this.ordinal })
})
}, { global: true }))
} catch (error: unknown) {
this[Symbol.dispose]()
throw error
}
}
/** @returns the baseline and local arrival cut captured synchronously together. */
snapshot(): { baseline: SessionAssistantStreamBaseline; ordinal: number } {
return { baseline: this._snapshot(), ordinal: this.ordinal }
}
/** @param ordinal - local snapshot cut set before queue consumption. */
setOrdinalCut(ordinal: number): void {
this.cut = ordinal
}
private *transform(item: AssistantItem): Iterable<AssistantEntry> {
if (this.disposes.length === 0 || item.ordinal <= this.cut) return
yield { type: 'assistant-stream', frame: item.frame }
}
[Symbol.dispose](): void {
for (const dispose of this.disposes.splice(0).reverse()) dispose()
}
}
packages/api/session-controller/src/history/controller.ts
+256
View File
@@ -0,0 +1,256 @@
/** Independent Session snapshots and passive event subscriptions. */
import type { Context } from '@deepseek-ai/cordis'
import { SessionLogOffset, SessionSeq, isAppendSurfaceEvent } from '@deepseek-ai/dsh-session'
import type { SessionId, SessionEvent } from '@deepseek-ai/dsh-session'
import type { SessionObservation } from '@deepseek-ai/dsh-session-query'
import { RemoteError, remoteErrorOf } from '@deepseek-ai/dsh-typert-protocol'
import {
DEFAULT_MAX_MESSAGES, paginate, pageRecords, entryFor, addressId,
validateHistoryWindow, validateAddress, rejectNotFound,
} from '../history.ts'
import { EventListener, ProjectionListener } from './listener.ts'
import { SessionAssistantStreamsListener } from './assistant-stream.ts'
import type {
SessionEventFrame, SessionEventsPage, SessionPageRequest,
SessionObserveEventRequest, SessionProjectionsRequest, SessionProjectionFrame,
SessionAddress, SessionProjectionBaseline, SessionProjectionValues, SessionProjectionsValue, SessionWaitOutput, SessionHistoryRecord,
} from '../types.ts'
/** Session snapshots do not share lifetime with passive subscriptions. */
export class HistoryController {
/** Reconnect state for this controller's assistant subscriptions. */
readonly streamsListener: SessionAssistantStreamsListener
/** @param ctx - Host context carrying Session reads and projections. */
constructor(private readonly ctx: Context) {
this.streamsListener = new SessionAssistantStreamsListener(this.ctx)
}
/**
* Read a passive history page at an exact event cut.
* @param request - Session identity and optional historical cut and page window.
* @param signal - read cancellation.
* @returns a page at the supplied cut, or the observed log end when omitted.
*/
async history(request: SessionPageRequest, signal: AbortSignal): Promise<SessionEventsPage> {
const spec = resolveHistoryRequest(request)
try {
return await this.usingObservation({ sessionId: addressId(request.address), projection: request.address.kind === 'subagent' }, signal, (observation) => {
validateObservationAddress(request.address, observation)
return history(observation, spec)
})
} catch (error: unknown) {
if (remoteErrorOf(error)?.code === 'session/not-found') rejectNotFound(request.address)
throw error
}
}
/**
* Subscribe to Session events without reading or activating the Session.
* @param request - Session address to subscribe.
* @param signal - stream cancellation.
* @returns subscription readiness followed by event frames; readiness does not establish existence.
*/
async *observeEvent(request: SessionObserveEventRequest, signal: AbortSignal): AsyncIterable<SessionEventFrame> {
using listener = new EventListener(this.ctx, addressId(request.address), signal)
yield { type: 'ready' }
yield* listener.events()
}
/**
* Read exact projections or migration cache hints without activating the Session.
* @param request - Session identity whose exact projection baseline is required.
* @param signal - read cancellation.
* @returns exact sequenced values, migration cache hints, or null when absent.
*/
async projections(request: SessionProjectionsRequest, signal: AbortSignal): Promise<SessionProjectionsValue> {
const { sessionId } = request
if (sessionId.length === 0) {
throw new RemoteError('gateway/bad-request', 'sessionId must not be empty', {})
}
try {
signal.throwIfAborted()
const live = this.ctx.sessions.get(sessionId)
if (live !== undefined) {
const snapshot = this.ctx.sessionProjections.snapshot(live)
return { kind: 'sequenced', asOfSeq: snapshot.asOfSeq, values: snapshot.values as SessionProjectionValues }
} else {
const stored = await this.ctx.get('sessionPersistence')?.stat(sessionId, { signal })
signal.throwIfAborted()
if (stored?.formatStatus === 'migration-required') {
const cached = this.ctx.get('sessionProjectionCache')?.cachedSnapshot(stored.header)
return { kind: 'migration-required', values: { ...cached?.values } as SessionProjectionValues }
}
return await this.usingObservation({ sessionId: request.sessionId, projection: true }, signal, observation => ({
kind: 'sequenced' as const,
...projections(observation),
}))
}
} catch (error: unknown) {
const code = typeof error === 'object' && error !== null && 'code' in error ? error.code : undefined
if (code === 'SESSION_QUERY_SESSION_NOT_FOUND' || code === 'session/not-found') {
return null
}
if (signal.aborted || code === 'SESSION_QUERY_ABORTED') {
throw new RemoteError('gateway/cancelled', 'Session projection read was cancelled', {}, { cause: error })
}
if (remoteErrorOf(error) !== undefined) throw error
throw new RemoteError('gateway/internal', 'Session projection read failed', {}, { cause: error })
}
}
/**
* Subscribe to sparse projection replacements without reading a baseline.
* @param request - Session identity to subscribe without reading a baseline.
* @param signal - stream cancellation.
* @returns readiness followed by sparse per-key replacement values.
*/
async *observeProjections(request: SessionProjectionsRequest, signal: AbortSignal): AsyncIterable<SessionProjectionFrame> {
using listener = new ProjectionListener(this.ctx, request.sessionId, signal)
yield { type: 'ready' }
yield* listener.changes()
}
/**
* Read bounded committed assistant messages through a stopping cursor.
* @param sessionId - Session whose committed assistant messages are requested.
* @param afterSeq - exclusive lower bound.
* @param throughSeq - inclusive stopping-state cursor.
* @param maxMessages - positive assistant record limit.
* @param signal - caller cancellation, independent of a completed wait deadline.
* @returns newest-first assistant messages and whether further matching records were omitted.
*/
async assistantOutput(
sessionId: SessionId, afterSeq: number, throughSeq: number, maxMessages: number, signal: AbortSignal,
): Promise<SessionWaitOutput> {
return this.usingObservation({ sessionId, projection: false }, signal, (observation) => {
if (throughSeq > observation.cursor) {
throw new RemoteError('gateway/internal', 'Session history no longer covers the wait cursor', {})
}
const records: SessionHistoryRecord[] = []
const events = observation.events
for (let index = throughSeq; index > afterSeq; index--) {
const event = events[index] as SessionEvent
if (event.type !== 'assistant/message' || !isAppendSurfaceEvent(event)) continue
if (records.length === maxMessages) return { records, truncated: true }
records.push(entryFor(event))
}
return { records, truncated: false }
})
}
/**
* Acquire a read without activating a Session; projections are opt-in.
* @param request - Session identity and whether to compute projections.
* @param signal - read cancellation.
* @param callback - omitted to transfer lease ownership to the caller.
* @returns a lease the caller must dispose.
*/
usingObservation(
request: { readonly sessionId: SessionId; readonly projection: boolean }, signal: AbortSignal, callback?: undefined,
): Promise<SessionObservation>
/**
* Keep the borrowed lease alive until the callback settles, then dispose it.
* @param request - Session identity and whether to compute projections.
* @param signal - read cancellation.
* @param callback - synchronous or asynchronous computation; retain a separate lease if it must escape.
* @returns the callback result; read and callback failures propagate.
*/
usingObservation<T>(
request: { readonly sessionId: SessionId; readonly projection: boolean }, signal: AbortSignal,
callback: (observation: SessionObservation) => T,
): Promise<Awaited<T>>
async usingObservation<T>(
{ sessionId, projection }: { readonly sessionId: SessionId; readonly projection: boolean }, signal: AbortSignal,
callback?: (observation: SessionObservation) => T,
): Promise<SessionObservation | Awaited<T>> {
let observation: SessionObservation
try {
observation = await this.ctx.sessionQuery.observeSession(sessionId, { signal, projectionMode: projection ? 'all' : 'none' })
} catch (error: unknown) {
const code = (error as { readonly code?: unknown } | null | undefined)?.code
if (code === 'SESSION_QUERY_SESSION_NOT_FOUND') {
throw new RemoteError('session/not-found', `session "${sessionId}" not found`, { sessionId })
}
if (code === 'SESSION_QUERY_ABORTED') {
// Keep the query's own message: cancellation detail was visible before this mapping.
throw new RemoteError('gateway/cancelled',
error instanceof Error ? error.message : 'Session read was cancelled', {}, { cause: error })
}
throw error
}
try {
signal.throwIfAborted()
} catch (error: unknown) {
observation[Symbol.dispose]()
throw error
}
if (callback === undefined) return observation
using ownedObservation = observation
return await callback(ownedObservation)
}
}
/**
* Validate an Address against the exact read used by a Controller.
* @param address - caller-supplied Session address.
* @param observation - borrowed read; subagent addresses require projections.
*/
export function validateObservationAddress(address: SessionAddress, observation: SessionObservation): void {
if (observation.header.cwd === undefined) rejectNotFound(address)
validateAddress(address, observation.header, observation.inheritedEventCount, observation.projections)
}
/**
* Convert a borrowed observation to one page without acquiring or retaining it.
* @param observation - exact read cut owned by the caller.
* @param request - validated page window and optional inclusive cut.
* @returns a chronological page at the requested cut.
*/
export function history(observation: SessionObservation, request: Omit<SessionPageRequest, 'address'> & { readonly maxMessages: number }): SessionEventsPage {
const cursor = request.throughSeq === undefined ? observation.cursor
: request.throughSeq === -1 ? -1 : SessionSeq(request.throughSeq)
if (cursor > observation.cursor) {
throw new RemoteError('gateway/bad-request', `session history cut ${String(cursor)} exceeds ${String(observation.cursor)}`, {})
}
const page = paginate(
observation.events,
request.beforeSeq === undefined ? undefined : SessionLogOffset(request.beforeSeq),
request.maxMessages,
cursor,
request.turnWindow,
)
return { cursor, records: pageRecords(page.events), hasMore: page.hasMore }
}
/**
* Convert the projection baseline without acquiring or retaining its observation.
* @param observation - projected read cut owned by the caller.
* @returns the exact baseline; missing projection support fails explicitly.
*/
export function projections(observation: SessionObservation): SessionProjectionBaseline {
const snapshot = observation.projections
if (snapshot === undefined) throw new RemoteError('session/projections-unavailable', 'Session projections are unavailable', {})
return { asOfSeq: snapshot.asOfSeq, values: snapshot.values as SessionProjectionValues }
}
/**
* Validate and resolve the history page window.
* @param request - external history window.
* @returns the validated request with an explicit message budget.
*/
export function resolveHistoryRequest(request: SessionPageRequest): SessionPageRequest & { readonly maxMessages: number } {
if (request.throughSeq !== undefined
&& (!Number.isSafeInteger(request.throughSeq) || request.throughSeq < -1 || Object.is(request.throughSeq, -0))) {
throw new RemoteError('gateway/bad-request', 'throughSeq must be an integer greater than or equal to -1', {})
}
if (request.beforeSeq !== undefined
&& (!Number.isSafeInteger(request.beforeSeq) || request.beforeSeq < 0 || Object.is(request.beforeSeq, -0))) {
throw new RemoteError('gateway/bad-request', 'beforeSeq must be a non-negative safe integer', {})
}
validateHistoryWindow(request)
return { ...request, maxMessages: request.maxMessages ?? DEFAULT_MAX_MESSAGES }
}
packages/api/session-controller/src/history/listener.ts
+303
View File
@@ -0,0 +1,303 @@
/** Passive per-Session streams; subscriptions never acquire Session observations. */
import type { Context } from '@deepseek-ai/cordis'
import { Deque } from '@deepseek-ai/dsh-deque'
import { SessionLogOffset } from '@deepseek-ai/dsh-session'
import type { Session, SessionEvent, SessionId } from '@deepseek-ai/dsh-session'
import { RemoteError } from '@deepseek-ai/dsh-typert-protocol'
import type { JsonValue } from '@deepseek-ai/dsh-util-values'
import { entryFor } from '../history.ts'
import type { SessionProjectionChange, SessionEventEntry } from '../types.ts'
/** Raw history notifications buffered before an opening cut is known. */
export type EventItem =
| { readonly type: 'event'; readonly session: Session; readonly event: SessionEvent }
| { readonly type: 'created'; readonly session: Session; readonly end: number }
/** A single subscription buffers before either consumption mode starts. */
export class EventListener implements Disposable {
private readonly buffer: ListenerBuffer<SessionEventEntry> | undefined
private readonly disposes: (() => void)[] = []
private firstReceivedSeq: number | undefined
private nextExpectedSeq: number | undefined
/**
* @param ctx - context owning the event registrations.
* @param sessionId - Session whose notifications are emitted.
* @param signal - cancellation that wakes consumers.
* @param queue - optional caller-owned buffer; only an internally created buffer is disposed here.
*/
constructor(ctx: Context, sessionId: SessionId, signal: AbortSignal,
queue?: Pick<ListenerBuffer<SessionEventEntry>, 'createInput' | 'capture'>) {
signal.throwIfAborted()
const buffer = queue ?? new ListenerBuffer(ctx, signal)
this.buffer = queue === undefined ? buffer as ListenerBuffer<SessionEventEntry> : undefined
const input = buffer.createInput<EventItem>(item => this.transform(item))
const target = sessionId
try {
this.disposes.push(ctx.on('session/event', (session, event) => {
if (session.id !== target) return
buffer.capture(() => {
this.recordFirst(event.seq)
input.append({ type: 'event', session, event })
})
}, { global: true }))
this.disposes.push(ctx.on('session/created', (session) => {
if (session.id !== target) return
buffer.capture(() => {
if (session.firstLiveSeq < session.seq) this.recordFirst(session.firstLiveSeq)
input.prepend({ type: 'created', session, end: session.seq })
})
}, { global: true }))
this.buffer?.own(() => { this[Symbol.dispose]() })
} catch (error: unknown) {
this[Symbol.dispose]()
throw error
}
}
/** Consume notifications without reading history.
* @returns event frames from the first received event, without loading a baseline. */
events(): AsyncIterable<SessionEventEntry> {
return this.iterate()
}
/** Set the inclusive opening cut for buffered history notifications.
* @param cursor - snapshot cut set before consuming the queue. */
setCursor(cursor: number): void {
this.nextExpectedSeq = cursor + 1
}
private *transform(item: EventItem): Iterable<SessionEventEntry> {
if (this.disposes.length === 0) return
this.nextExpectedSeq ??= this.firstReceivedSeq
const result = eventItem(item, this.nextExpectedSeq)
this.nextExpectedSeq = result.next
for (const event of result.events) {
if (this.disposes.length === 0) return
yield entryFor(event)
}
}
private async *iterate(): AsyncIterable<SessionEventEntry> {
if (this.buffer === undefined) throw new Error('Consume the externally owned history buffer directly')
try {
yield* this.buffer.items()
} finally {
this[Symbol.dispose]()
}
}
private recordFirst(seq: number): void {
this.firstReceivedSeq = this.firstReceivedSeq === undefined ? seq : Math.min(this.firstReceivedSeq, seq)
}
[Symbol.dispose](): void {
for (const dispose of this.disposes.splice(0).reverse()) dispose()
this.buffer?.[Symbol.dispose]()
}
}
/**
* Select the uncovered events represented by one notification.
* @param item - one buffered notification.
* @param next - next required seq, or undefined before the first event.
* @returns uncovered events and the next required seq without acquiring an observation.
*/
export function eventItem(item: EventItem, next?: number): {
events: readonly SessionEvent[]
next: number | undefined
} {
if (item.type === 'event') {
if (next !== undefined && item.event.seq < next) return { events: [], next }
if (next !== undefined && item.event.seq !== next) {
throw new RemoteError('gateway/internal', `session event stream skipped seq ${String(next)}`, {})
}
return { events: [item.event], next: item.event.seq + 1 }
}
const start = next ?? item.session.firstLiveSeq
if (start >= item.end) return { events: [], next }
// oxlint-disable-next-line typescript/no-deprecated -- Existing Session-created history adaptation.
const events = item.session.snapshotEvents(SessionLogOffset(start), SessionLogOffset(item.end))
for (const event of events) {
if (next !== undefined && event.seq !== next) {
throw new RemoteError('gateway/internal', `session event stream skipped seq ${String(next)}`, {})
}
next = event.seq + 1
}
return { events, next }
}
/** Sparse projection changes retain every key, including keys sharing a seq. */
export class ProjectionListener implements Disposable {
private readonly buffer: ListenerBuffer<{ session: Session; change: SessionProjectionChange }>
/**
* @param ctx - context owning projection notifications.
* @param sessionId - Session whose notifications are emitted.
* @param signal - cancellation that removes registrations and wakes consumers.
*/
constructor(ctx: Context, sessionId: SessionId, signal: AbortSignal) {
this.buffer = new ListenerBuffer(ctx, signal)
const input = this.buffer.createInput<{ session: Session; change: SessionProjectionChange }>(item => [item])
const target = sessionId
try {
this.buffer.own(ctx.sessionProjections.onChanged((session, key, value, seq) => {
if (session.id !== target) return
this.buffer.capture(() => {
input.append({
session, change: { type: 'projection', key, value: value as JsonValue, seq } satisfies SessionProjectionChange,
})
})
}))
} catch (error: unknown) {
this.buffer[Symbol.dispose]()
throw error
}
}
/** Consume sparse per-key projection replacements.
* @returns validated replacement frames; projection seqs need not be contiguous. */
async *changes(): AsyncIterable<SessionProjectionChange> {
try {
for await (const { change } of this.buffer.items()) {
yield change
}
} finally {
this[Symbol.dispose]()
}
}
[Symbol.dispose](): void {
this.buffer[Symbol.dispose]()
}
}
/** Owns buffering, callback failure and registration cleanup for one consumer. */
export class ListenerBuffer<T> implements Disposable {
private readonly values = new Deque<unknown>()
private readonly transformIds = new Deque<number>()
private readonly transforms: ((value: unknown) => Iterable<T>)[] = []
private readonly cleanup: (() => void)[] = []
private wake: (() => void) | undefined
private closed = false
private failure: { readonly error: unknown } | undefined
private consumed = false
constructor(ctx: Context, signal: AbortSignal) {
signal.throwIfAborted()
const close = (): void => { this[Symbol.dispose]() }
signal.addEventListener('abort', close, { once: true })
this.own(() => { signal.removeEventListener('abort', close) })
try {
const dispose = ctx.effect(() => close, 'session-controller.listener-buffer')
this.own(() => { void dispose() })
} catch (error: unknown) {
this[Symbol.dispose]()
throw error
}
}
/**
* Register cleanup, or run it immediately when already closed.
* @param dispose - synchronous cleanup.
*/
own(dispose: () => void): void {
if (this.closed) dispose()
else this.cleanup.push(dispose)
}
/**
* Contain producer failure and wake the consumer with that failure.
* @param action - synchronous producer operation.
*/
capture(action: () => void): void {
if (this.closed) return
try {
action()
} catch (error: unknown) {
this.failure = { error }
this[Symbol.dispose]()
}
}
/**
* Register a dequeue transform and obtain its write-only input.
* @param transform - conversion bound once and evaluated when each input is dequeued.
* @returns a write-only input sharing this buffer's storage and lifetime.
* @throws when this buffer is already closed.
*/
createInput<Input>(transform: (item: Input) => Iterable<T>): { append(item: Input): void; prepend(item: Input): void } {
if (this.closed) throw new Error('Cannot register a transform on a closed listener buffer')
const transformId = this.transforms.length
// Only this typed input can enqueue values with its private transform id.
this.transforms.push(value => transform(value as Input))
return {
append: (item) => {
if (this.closed) return
this.values.pushBack(item)
this.transformIds.pushBack(transformId)
this.notify()
},
prepend: (item) => {
if (this.closed) return
this.values.pushFront(item)
this.transformIds.pushFront(transformId)
this.notify()
},
}
}
/**
* Consume queued values once, releasing the buffer when iteration ends.
* @returns transformed output values.
*/
async *items(): AsyncIterable<T> {
if (this.consumed) throw new RemoteError('gateway/internal', 'Session listener already has a consumer', {})
this.consumed = true
try {
for (;;) {
if (this.failure !== undefined) throw this.failure.error
if (this.closed) return
if (this.transformIds.size === 0) {
await new Promise<void>((resolve) => { this.wake = resolve })
continue
}
const transformId = this.transformIds.popFront() as number
const value = this.values.popFront()
for (const item of (this.transforms[transformId] as (value: unknown) => Iterable<T>)(value)) {
this.throwIfFailed()
if (this.isClosed()) return
yield item
}
}
} finally {
this[Symbol.dispose]()
}
}
private throwIfFailed(): void {
if (this.failure !== undefined) throw this.failure.error
}
/** @returns whether disposal ended this buffer while a transform was yielding. */
private isClosed(): boolean {
return this.closed
}
private notify(): void {
const wake = this.wake
this.wake = undefined
wake?.()
}
[Symbol.dispose](): void {
if (this.closed) return
this.closed = true
this.values.clear()
this.transformIds.clear()
this.transforms.length = 0
this.notify()
for (const dispose of this.cleanup.splice(0).reverse()) dispose()
}
}
packages/api/session-controller/src/index.ts
+94 -55
View File
@@ -11,7 +11,7 @@ import type {} from '@deepseek-ai/dsh-client-file-upload'
import { canOpenNativePath, nativeFileManager, nativeFileApplications, openNativeFileApplication, openNativeAssociatedPath, revealNativePath } from '@deepseek-ai/dsh-native-command'
import type { SessionId } from '@deepseek-ai/dsh-session'
import type {} from '@deepseek-ai/dsh-session-projection-cache'
import { SessionQueryError, type SessionObservation } from '@deepseek-ai/dsh-session-query'
import type { SessionObservation } from '@deepseek-ai/dsh-session-query'
import { Remote, RemoteError, TypertRemoteService, principalOf, type TypertResourceResolveInput } from '@deepseek-ai/dsh-typert-protocol'
import { canonicalClientTimeZone } from '@deepseek-ai/dsh-util-time'
import { assertNever } from '@deepseek-ai/dsh-util-values'
@@ -22,7 +22,9 @@ import {
} from './agent.ts'
import { SessionCommandController, hasPromptContent } from './commands.ts'
import { SessionControlController } from './control.ts'
import { SessionHistoryController } from './history.ts'
import { HistoryController } from './history/controller.ts'
import { StateController } from './state/controller.ts'
import { WebController } from './web/controller.ts'
import { SessionFileReferences } from './file-references.ts'
import { ApiSessionList } from './list.ts'
import { buildModelCatalog, hasProviderApiKey } from './catalog.ts'
@@ -48,8 +50,8 @@ import type {
SessionListValue,
SessionOpenWorkspacePathRequest,
SessionOpenWorkspacePathValue,
SessionPage,
SessionPageRequest,
SessionEventsPage, SessionPageRequest, SessionEventFrame, SessionObserveEventRequest,
SessionProjectionFrame, SessionStateRequest, SessionState, SessionWaitRequest, SessionWaitValue,
SessionPromptRequest,
SessionPromptValue,
SessionRenameRequest,
@@ -60,7 +62,6 @@ import type {
SessionSelectModelValue,
SessionProjectionsRequest,
SessionProjectionsValue,
SessionProjectionValues,
SessionUpdateQueueRequest,
SessionUpdateQueueValue,
} from './types.ts'
@@ -144,6 +145,7 @@ export class SessionController extends TypertRemoteService {
'sessionProjections',
'sessionQuery',
'typert',
'typertGateway',
'workspaceRegistry',
]
@@ -152,10 +154,13 @@ export class SessionController extends TypertRemoteService {
listWorkSliceMs: z.natural().min(1).default(16),
})
private historyController!: HistoryController
private stateController!: StateController
private webController!: WebController
private agents!: ApiSessionAgentController
private commands!: SessionCommandController
private controlState!: SessionControlController
private history!: SessionHistoryController
private listState!: ApiSessionList
private readonly openPath: (path: string, signal: AbortSignal) => Promise<void>
private readonly fileApplications: typeof nativeFileApplications
@@ -198,7 +203,11 @@ export class SessionController extends TypertRemoteService {
ctx.effect(() => async () => {
await Promise.allSettled([...this.promotions])
}, 'session-controller.promotions')
this.history = new SessionHistoryController(ctx, (observation) => { this.promote(observation) })
// Assistant-stream observation must be live before the list read yields: an Agent's
// opening frame arriving during that await would otherwise have no listener.
this.historyController = new HistoryController(ctx)
this.stateController = new StateController(ctx, this.historyController)
this.webController = new WebController(ctx, this.historyController, (observation) => { this.promote(observation) })
this.listState = new ApiSessionList(ctx, this.listWorkSliceMs)
await this.listState.initialize()
ctx.plugin(SessionFileReferences)
@@ -234,6 +243,10 @@ export class SessionController extends TypertRemoteService {
})
}
/**
* Request background activation through the existing Session Agent controller.
* @param observation - owned lease disposed by the background task; callers retain their own lease separately.
*/
private promote(observation: SessionObservation): void {
const sessionId = observation.header.id
const task = (async () => {
@@ -507,13 +520,80 @@ export class SessionController extends TypertRemoteService {
/**
* Read one cold-safe, message-aligned Session history page.
* @param request - durable address, backward cursor, and page budget.
* @deprecated Use `events`; this entry point remains for existing Web callers.
* @param request - durable address, optional fixed cut, backward cursor, and page budget.
* @param signal - cancellation for persistence reads.
* @returns one chronological page.
* @returns one chronological page with its inclusive cursor; an omitted `throughSeq` selects the observed log end.
*/
@Remote('page')
page(request: SessionPageRequest, signal: AbortSignal): Promise<SessionPage> {
return this.history.page(request, signal)
page(request: SessionPageRequest, signal: AbortSignal): Promise<SessionEventsPage> {
return this.historyController.history(request, signal)
}
/**
* Read one cold-safe, message-aligned Session history page.
* @param request - durable address, optional fixed cut, backward cursor, and page budget.
* @param signal - cancellation for persistence reads.
* @returns one chronological page with its inclusive cursor.
*/
@Remote
events(request: SessionPageRequest, signal: AbortSignal): Promise<SessionEventsPage> {
return this.historyController.history(request, signal)
}
/**
* Subscribe to Session events without reading a snapshot.
* @param request - Session address.
* @param signal - subscription cancellation.
* @returns readiness and subsequent events, without a snapshot.
*/
@Remote({ mode: 'stream' })
observeEvent(request: SessionObserveEventRequest, signal: AbortSignal): AsyncIterable<SessionEventFrame> {
return this.historyController.observeEvent(request, signal)
}
/**
* Subscribe to sparse Session projection changes.
* @param request - target Session.
* @param signal - subscription cancellation.
* @returns readiness and sparse projection changes.
*/
@Remote({ mode: 'stream' })
observeProjections(request: SessionProjectionsRequest, signal: AbortSignal): AsyncIterable<SessionProjectionFrame> {
return this.historyController.observeProjections(request, signal)
}
/**
* Read passive Session state.
* @param request - target Session.
* @param signal - read cancellation.
* @returns current passive state.
*/
@Remote
state(request: SessionStateRequest, signal: AbortSignal): Promise<SessionState> {
return this.stateController.state(request, signal)
}
/**
* Subscribe to complete Session state replacements.
* @param request - target Session.
* @param signal - subscription cancellation.
* @returns complete passive state replacements.
*/
@Remote({ mode: 'stream' })
observeState(request: SessionStateRequest, signal: AbortSignal): AsyncIterable<SessionState> {
return this.stateController.observeState(request, signal)
}
/**
* Wait for Session quiescence or pending input with bounded assistant output.
* @param request - Session, wait bounds and output window.
* @param signal - caller cancellation.
* @returns stopping state and bounded assistant output.
*/
@Remote
wait(request: SessionWaitRequest, signal: AbortSignal): Promise<SessionWaitValue> {
return this.stateController.wait(request, signal)
}
/**
@@ -525,7 +605,7 @@ export class SessionController extends TypertRemoteService {
*/
@Remote({ mode: 'stream' })
follow(request: SessionFollowRequest, signal: AbortSignal): AsyncIterable<SessionFollowFrame> {
return this.history.follow(request, signal)
return this.webController.follow(request, signal)
}
/**
@@ -535,49 +615,8 @@ export class SessionController extends TypertRemoteService {
* @returns a sequenced baseline, cached hints when migration is deferred, or null when absent.
*/
@Remote('projections')
async projections(request: SessionProjectionsRequest, signal: AbortSignal): Promise<SessionProjectionsValue> {
const { sessionId } = request
if (sessionId.length === 0) {
throw new RemoteError('gateway/bad-request', 'sessionId must not be empty', {})
}
try {
signal.throwIfAborted()
const live = this.liveProjections(sessionId)
if (live !== undefined) return live
const stored = await this.ctx.get('sessionPersistence')?.stat(sessionId, { signal })
signal.throwIfAborted()
const attached = this.liveProjections(sessionId)
if (attached !== undefined) return attached
if (stored === undefined) return null
if (stored.formatStatus === 'migration-required') {
const cached = this.ctx.get('sessionProjectionCache')?.cachedSnapshot(stored.header)
return { kind: 'migration-required', values: { ...cached?.values } as SessionProjectionValues }
}
using observation = await this.ctx.sessionQuery.observeSession(sessionId, { signal })
const projections = observation.projections
if (projections === undefined) {
throw new RemoteError('session/projections-unavailable', 'Session projections are unavailable', {})
}
return { kind: 'sequenced', asOfSeq: projections.asOfSeq, values: projections.values as SessionProjectionValues }
} catch (error: unknown) {
if (error instanceof SessionQueryError && error.code === 'SESSION_QUERY_SESSION_NOT_FOUND') {
return this.liveProjections(sessionId) ?? null
}
if (signal.aborted || (error instanceof SessionQueryError && error.code === 'SESSION_QUERY_ABORTED')) {
throw new RemoteError('gateway/cancelled', 'Session projection read was cancelled', {}, { cause: error })
}
if (error instanceof RemoteError) throw error
throw new RemoteError('gateway/internal', 'Session projection read failed', {}, { cause: error })
}
}
private liveProjections(
sessionId: SessionId,
): Extract<SessionProjectionsValue, { readonly kind: 'sequenced' }> | undefined {
const session = this.ctx.sessions.get(sessionId)
if (session === undefined) return undefined
const snapshot = this.ctx.sessionProjections.snapshot(session)
return { kind: 'sequenced', asOfSeq: snapshot.asOfSeq, values: snapshot.values as SessionProjectionValues }
projections(request: SessionProjectionsRequest, signal: AbortSignal): Promise<SessionProjectionsValue> {
return this.historyController.projections(request, signal)
}
/**
packages/api/session-controller/src/state/controller.ts
+221
View File
@@ -0,0 +1,221 @@
/** Passive Session state, coalesced observations and quiescence waits. */
import type { Context } from '@deepseek-ai/cordis'
import type { Agent } from '@deepseek-ai/dsh-agent'
import type {} from '@deepseek-ai/dsh-api-gateway'
import type { UserMessage } from '@deepseek-ai/dsh-llm'
import type { SessionId } from '@deepseek-ai/dsh-session'
import { deadline, MAX_TIMER_DELAY_MS, timeoutOf } from '@deepseek-ai/dsh-timeout'
import { RemoteError } from '@deepseek-ai/dsh-typert-protocol'
import type { JsonValue } from '@deepseek-ai/dsh-util-values'
import type { HistoryController } from '../history/controller.ts'
import { DEFAULT_MAX_MESSAGES, rejectNotFound } from '../history.ts'
import type { SessionPendingInput, SessionQueuedItem, SessionState, SessionStateRequest, SessionWaitRequest, SessionWaitValue } from '../types.ts'
import { StateListener } from './listener.ts'
import type { VersionedSessionState } from './listener.ts'
const DEFAULT_WAIT_TIMEOUT_MS = 30_000
/** Reads and waits without activating or cancelling Session execution. */
export class StateController {
/**
* @param ctx - live Session and pending Remote event services.
* @param historyController - passive read entry used for cold state and its bounded output.
*/
constructor(private readonly ctx: Context, private readonly historyController: HistoryController) {}
/**
* Read passive execution and pending-input state.
* @param request - Session identity.
* @param signal - read cancellation.
* @returns current passive state.
*/
async state(request: SessionStateRequest, signal: AbortSignal): Promise<SessionState> {
return this.readState(request.sessionId, signal)
}
/**
* Observe complete passive state replacements.
* @param request - Session identity.
* @param signal - observation cancellation.
* @returns complete state replacements, starting with the current state.
*/
async *observeState(request: SessionStateRequest, signal: AbortSignal): AsyncIterable<SessionState> {
using listener = this.listen(request.sessionId, signal)
for await (const { state } of listener.observeStates()) yield state
}
/**
* Wait for quiescence, pending input, or the deadline without cancelling execution.
* @param request - Session identity, timeout and assistant output window.
* @param signal - cancellation of this wait only, never of Agent execution.
* @returns the stopping reason and last complete state; no state before timeout is an error.
*/
async wait(request: SessionWaitRequest, signal: AbortSignal): Promise<SessionWaitValue> {
const timeoutMs = request.timeoutMs ?? DEFAULT_WAIT_TIMEOUT_MS
const afterSeq = request.afterSeq ?? -1
const maxMessages = request.maxMessages ?? DEFAULT_MAX_MESSAGES
if (!Number.isSafeInteger(afterSeq) || afterSeq < -1 || Object.is(afterSeq, -0)) {
throw new RemoteError('gateway/bad-request', 'afterSeq must be an integer greater than or equal to -1', {})
}
if (!Number.isSafeInteger(maxMessages) || maxMessages <= 0) {
throw new RemoteError('gateway/bad-request', 'maxMessages must be a positive safe integer', {})
}
if (!Number.isSafeInteger(timeoutMs) || timeoutMs <= 0 || timeoutMs > MAX_TIMER_DELAY_MS) {
throw new RemoteError('gateway/bad-request', 'timeoutMs must be a positive supported timer delay', {})
}
using timer = deadline(signal, timeoutMs, 'session/wait-timeout')
using listener = this.listen(request.sessionId, timer.signal)
const sessionId = request.sessionId
const statesGenerator = listener.observeStates(1_000)
let nextStatePromise = statesGenerator.next()
let currentState: SessionState | undefined
let reason: SessionWaitValue['reason'] = 'timeout'
try {
for (;;) {
const observedState = await nextStatePromise
if (observedState.done) {
timer.signal.throwIfAborted()
throw new RemoteError('session/observation-ended', 'Session observation ended before a stopping state', {})
}
nextStatePromise = statesGenerator.next()
// Observe rejection on immediate exits without replacing the original demand.
void nextStatePromise.catch(() => undefined)
// Only a state the listener still confirms may drive this wait; an invalidated
// sample never becomes the state a timeout reports.
if (!listener.isCurrent(observedState.value.revision)) continue
currentState = observedState.value.state
const action = computeWaitAction(currentState)
if (action === 'continue') continue
if (action === 'input-required' || action === 'idle') {
reason = action
break
}
const agent = this.ctx.agents.get(sessionId)
if (agent === undefined || this.ctx.sessions.get(sessionId) !== agent.session) continue
if (await waitForIdleOrStateChange(agent, nextStatePromise) === 'state-changed') continue
if (!listener.isCurrent(observedState.value.revision) || this.ctx.agents.get(sessionId) !== agent
|| this.ctx.sessions.get(sessionId) !== agent.session) continue
reason = 'idle'
break
}
} catch (error: unknown) {
// A caller cancellation is never a wait outcome; the deadline absorbs only the read
// it cancelled, and any other failure keeps its own error.
if (signal.aborted) throw signal.reason
if (!expired(timer.signal, error)) throw error
} finally {
// Wake/cancel a pending read before asking the generator to return.
listener[Symbol.dispose]()
await nextStatePromise.catch(() => undefined)
await statesGenerator.return(undefined)
}
signal.throwIfAborted()
if (currentState === undefined) throw new RemoteError('session/wait-timeout', 'No Session state arrived before the wait deadline', {})
const output = await this.historyController.assistantOutput(
sessionId, afterSeq, currentState.cursor, maxMessages, signal,
)
return { reason, state: currentState, output }
}
private listen(sessionId: SessionId, signal: AbortSignal): StateListener {
return new StateListener(this.ctx, sessionId, signal, readSignal => this.readState(sessionId, readSignal))
}
/**
* @param sessionId - Session to sample.
* @param signal - read cancellation.
* @param retried - whether a live source already redirected this read once.
* @returns one passive state sample.
*/
private async readState(sessionId: SessionId, signal: AbortSignal, retried = false): Promise<SessionState> {
signal.throwIfAborted()
const agent = this.ctx.agents.get(sessionId)
if (agent !== undefined) {
if (agent.session.header.cwd === undefined) rejectNotFound({ kind: 'session', sessionId })
return {
cursor: agent.session.seq - 1,
active: true,
running: agent.status === 'running',
queue: queueItems(agent),
pendingInput: this.pendingInput(agent.session.id),
}
}
const state = await this.historyController.usingObservation({ sessionId, projection: false }, signal, (observation) => {
// A live source means the Session attached during this read; report its live state
// rather than a live cursor paired with an inactive sample.
if (!retried && observation.source === 'live') return undefined
if (observation.header.cwd === undefined) rejectNotFound({ kind: 'session', sessionId })
return {
cursor: observation.cursor,
active: false,
running: false,
queue: null,
pendingInput: this.pendingInput(observation.header.id),
}
})
signal.throwIfAborted()
return state ?? this.readState(sessionId, signal, true)
}
private pendingInput(sessionId: SessionId): readonly SessionPendingInput[] {
const kinds = new Set<SessionPendingInput>()
for (const { event, agentId } of this.ctx.typertGateway.getPendingRemoteEvents()) {
if (String(agentId) !== sessionId) continue
kinds.add(event === 'approval/request' ? 'approval' : event === 'user-questions/request' ? 'question' : 'other')
}
return [...kinds]
}
}
/**
* @param signal - wait deadline signal.
* @param error - failure raised by the in-flight read.
* @returns whether the deadline expired and cancelled that read.
*/
function expired(signal: AbortSignal, error: unknown): boolean {
const reason = timeoutOf(signal, 'session/wait-timeout')
if (reason === undefined) return false
// Cancellation carries the deadline reason along its cause chain; any other failure does not.
let current: unknown = error
for (let depth = 0; depth < 4 && current instanceof Error; depth++) {
if (current === reason || current.cause === reason) return true
current = current.cause
}
return error === reason
}
/** Choose the next wait action; check-idle requires the caller's whenIdle and identity checks. */
function computeWaitAction(state: SessionState): 'input-required' | 'continue' | 'idle' | 'check-idle' {
if (state.pendingInput.length > 0) return 'input-required'
if (state.running) return 'continue'
if (!state.active) return 'idle'
return 'check-idle'
}
/** Wait for either driver quiescence or the already-armed next state; preserve rejections from both. */
function waitForIdleOrStateChange(
agent: Agent,
nextStatePromise: Promise<IteratorResult<VersionedSessionState>>,
): Promise<'idle' | 'state-changed'> {
return Promise.race([
agent.whenIdle().then(() => 'idle' as const),
nextStatePromise.then(() => 'state-changed' as const),
])
}
function queueItems(agent: Agent): SessionQueuedItem[] {
const item = (message: UserMessage, placement: SessionQueuedItem['placement']): SessionQueuedItem => ({
id: message.id,
placement,
...message.source.kind === 'user' && 'rpcId' in message.source ? { rpcId: message.source.rpcId } : {},
content: message.content as typeof message.content & readonly JsonValue[],
})
return [
...agent.inbox.nextTurn.map(message => item(message, 'queued')),
...agent.inbox.nextStep.map(message => item(message, message.source.kind === 'user' ? 'steering' : 'context')),
]
}
packages/api/session-controller/src/state/listener.ts
+116
View File
@@ -0,0 +1,116 @@
/** Coalesced Session state observations with Host-local revisions. */
import type { Context } from '@deepseek-ai/cordis'
import type { SessionId } from '@deepseek-ai/dsh-session'
import type { SessionState } from '../types.ts'
/** One complete state with its listener-local revision. */
export interface VersionedSessionState { state: SessionState; revision: number }
/** Coalesces Session/Agent invalidations with local revisions; pending events are sampled on reads. */
export class StateListener implements Disposable {
private readonly disposes: (() => void)[] = []
private readonly cancellation = new AbortController()
private readonly signal: AbortSignal
private revision = 0
private resolver = createDeferredSignal()
private closed = false
/**
* @param ctx - lifecycle notification owner.
* @param target - Session to observe.
* @param signal - caller cancellation.
* @param readState - passive complete-state reader.
*/
constructor(
ctx: Context, target: SessionId, signal: AbortSignal,
private readonly readState: (signal: AbortSignal) => Promise<SessionState>,
) {
signal.throwIfAborted()
this.signal = AbortSignal.any([signal, this.cancellation.signal])
const notify = (id: string | undefined): void => {
if (id !== target || this.closed) return
this.revision++
const resolver = this.resolver
this.resolver = createDeferredSignal()
resolver.resolve()
}
const close = (): void => { this[Symbol.dispose]() }
signal.addEventListener('abort', close, { once: true })
this.disposes.push(() => { signal.removeEventListener('abort', close) })
try {
this.disposes.push(ctx.on('session/event', (session) => { notify(session.id) }, { global: true }))
this.disposes.push(ctx.on('session/created', (session) => { notify(session.id) }, { global: true }))
this.disposes.push(ctx.on('session/disposed', (session) => { notify(session.id) }, { global: true }))
this.disposes.push(ctx.on('agent/status', ({ agent }) => { notify(agent.id) }, { global: true }))
this.disposes.push(ctx.on('agent/created', ({ agent }) => { notify(agent.id) }, { global: true }))
this.disposes.push(ctx.on('agent/disposed', ({ agent }) => { notify(agent.id) }, { global: true }))
const dispose = ctx.effect(() => close, 'session-controller.state-listener')
this.disposes.push(() => { void dispose() })
} catch (error: unknown) {
this[Symbol.dispose]()
throw error
}
}
/**
* Check whether a completed read still represents the latest invalidation.
* @param revision - revision captured with a completed read.
* @returns whether no intervening invalidation occurred.
*/
isCurrent(revision: number): boolean { return !this.closed && revision === this.revision }
/**
* Read coalesced state replacements with optional periodic refresh.
* @param refreshMs - optional idle recheck interval; omitted for event-only observation.
* @returns complete state replacements with internal consistency revisions.
*/
async *observeStates(refreshMs?: number): AsyncGenerator<VersionedSessionState> {
let lastRevision = -1
try {
while (!this.closed) {
if (lastRevision === this.revision) {
if (refreshMs === undefined) {
await this.resolver.promise
} else {
let timer: ReturnType<typeof setTimeout> | undefined
try {
await Promise.race([
this.resolver.promise,
new Promise<void>((resolve) => { timer = setTimeout(resolve, refreshMs) }),
])
} finally {
clearTimeout(timer)
}
if (this.isCurrent(lastRevision)) {
this.revision++
const resolver = this.resolver
this.resolver = createDeferredSignal()
resolver.resolve()
}
}
}
if (!this.isCurrent(this.revision)) return
lastRevision = this.revision
const state = await this.readState(this.signal)
if (this.isCurrent(lastRevision)) yield { state, revision: lastRevision }
}
} finally {
this[Symbol.dispose]()
}
}
[Symbol.dispose](): void {
if (this.closed) return
this.closed = true
this.cancellation.abort()
this.resolver.resolve()
for (const dispose of this.disposes.splice(0).reverse()) dispose()
}
}
function createDeferredSignal(): { promise: Promise<void>; resolve: () => void } {
let resolve!: () => void
const promise = new Promise<void>((settle) => { resolve = settle })
return { promise, resolve }
}
packages/api/session-controller/src/types.ts
+86 -2
View File
@@ -175,6 +175,10 @@ export const SESSION_SEARCH_SNIPPET_MAX_CODE_POINTS = 240
declare module '@deepseek-ai/dsh-typert-protocol' {
interface RemoteErrorDetailsMap {
/** A wait observation ended before a stopping state; the caller should reread. */
'session/observation-ended': Record<string, never>
/** No state arrived before the deadline; a timeout after a state returns its value instead. */
'session/wait-timeout': Record<string, never>
'session/provider-credentials-unavailable': Record<string, never>
'session/provider-models-unavailable': { readonly provider: string }
'session/model-unavailable': { readonly provider: string; readonly model: string }
@@ -461,8 +465,11 @@ export interface SessionWireEvent {
/** One message-aligned backwards-history request. */
export interface SessionPageRequest {
readonly address: SessionAddress
/** Inclusive log cut obtained from the corresponding follow opening frame. */
readonly throughSeq: number
/**
* Inclusive log cut; omission selects the observed log end. A follow opening
* reports the cut it covered so a caller can pin the same point.
*/
readonly throughSeq?: number
readonly beforeSeq?: number
readonly maxMessages?: number
/** Stop at a Turn start after both minima, unless maxMessages or history exhaustion wins. */
@@ -477,6 +484,8 @@ export interface SessionPageRequest {
/** One live event request for a durable Session address. */
export interface SessionFollowRequest extends Pick<SessionPageRequest, 'maxMessages' | 'turnWindow'> {
readonly address: SessionAddress
/** Request background activation; omission leaves the read passive. */
readonly activate?: boolean
/** Include process-local assistant presentation frames for the Web client. */
readonly assistantStream?: true
}
@@ -571,6 +580,81 @@ export type SessionControlFrame =
| { readonly type: 'baseline'; readonly value: SessionControlBaseline }
| ({ readonly type: 'projection' } & SessionProjectionUpdate)
/** One pending live-Agent inbox item, preserving placement and content order. */
export interface SessionQueuedItem {
readonly id: MessageId
readonly placement: 'queued' | 'steering' | 'context'
readonly rpcId?: SessionRequestId
readonly content: readonly JsonValue[]
}
/** State-owned classification of pending forwarded Remote events. */
export type SessionPendingInput = 'approval' | 'question' | 'other'
/** Passive Session state; a missing live Agent has no readable queue. */
export interface SessionState {
readonly cursor: number
readonly active: boolean
readonly running: boolean
readonly queue: readonly SessionQueuedItem[] | null
/**
* Pending Remote requests sampled when this state was read. A pending change alone does not
* trigger `observeState`; `wait` refreshes on its own interval.
*/
readonly pendingInput: readonly SessionPendingInput[]
}
/** Session identity for a passive state read or subscription. */
export interface SessionStateRequest {
readonly sessionId: SessionId
}
/** Wait bounds and assistant output window; timeout omission uses the fixed 30-second default. */
export interface SessionWaitRequest extends SessionStateRequest {
readonly timeoutMs?: number
/** Exclusive output cursor; omission includes all available assistant records. */
readonly afterSeq?: number
/** Positive assistant-record limit; omission uses the history message budget. */
readonly maxMessages?: number
}
/** Assistant messages after afterSeq through the stopping state's cursor, newest first. */
export interface SessionWaitOutput {
readonly records: readonly SessionHistoryRecord[]
/** More eligible assistant messages exist in the selected interval. */
readonly truncated: boolean
}
/** A stopping condition, the last complete passive state and its bounded assistant output. */
export interface SessionWaitValue {
readonly reason: 'idle' | 'input-required' | 'timeout'
readonly state: SessionState
readonly output: SessionWaitOutput
}
/** One `session.history` page and its inclusive cut: the supplied `throughSeq`, otherwise the log end it was read at. */
export interface SessionEventsPage extends SessionPage {
readonly cursor: number
}
/** Address for a passive event subscription. */
export interface SessionObserveEventRequest {
readonly address: SessionAddress
}
/** Readiness confirms subscription only; event continuity starts with the first delivered event. */
export type SessionEventFrame =
| { readonly type: 'ready' }
| SessionEventEntry
/** One sparse per-key replacement; different keys can share a seq. */
export type SessionProjectionChange = { readonly type: 'projection' } & Omit<SessionProjectionUpdate, 'sessionId'>
/** Subscription readiness without a baseline, followed by per-key changes. */
export type SessionProjectionFrame =
| { readonly type: 'ready' }
| SessionProjectionChange
declare module '@deepseek-ai/cordis' {
interface Events {
/**
packages/api/session-controller/src/web/controller.ts
+78
View File
@@ -0,0 +1,78 @@
/** Web opening aggregation over one observation and a shared transforming buffer. */
import type { Context } from '@deepseek-ai/cordis'
import { remoteErrorOf } from '@deepseek-ai/dsh-typert-protocol'
import type { SessionObservation } from '@deepseek-ai/dsh-session-query'
import type { HistoryController } from '../history/controller.ts'
import { addressId, rejectNotFound } from '../history.ts'
import { EventListener, ListenerBuffer } from '../history/listener.ts'
import { history, projections, resolveHistoryRequest, validateObservationAddress } from '../history/controller.ts'
import type { SessionFollowFrame, SessionFollowRequest } from '../types.ts'
/** Combines opening data and consumes transformed listener output from one queue. */
export class WebController {
/**
* @param ctx - Host context carrying Session queries and identity projections.
* @param historyController - passive reading and listener construction.
* @param promote - background activation requested before an opening snapshot is delivered.
*/
constructor(
private readonly ctx: Context,
private readonly historyController: HistoryController,
private readonly promote: (observation: SessionObservation) => void,
) {}
/**
* Open a history snapshot and follow uncovered history and assistant notifications.
* @param request - Session identity, page budget, assistant output and optional activation (default false).
* @param signal - cancellation for opening and streaming.
* @returns an opening snapshot followed by uncovered queued increments.
*/
async *follow(request: SessionFollowRequest, signal: AbortSignal): AsyncIterable<SessionFollowFrame> {
const sessionId = addressId(request.address)
const spec = resolveHistoryRequest(request)
using queue = new ListenerBuffer<SessionFollowFrame>(this.ctx, signal)
using listener = new EventListener(this.ctx, sessionId, signal, queue)
const streams = this.historyController.streamsListener
using assistantListener = request.assistantStream === true
? streams.bind(sessionId, signal, queue)
: undefined
using observation = await this.historyController.usingObservation({ sessionId, projection: true }, signal)
.catch((error: unknown) => {
if (remoteErrorOf(error)?.code === 'session/not-found') rejectNotFound(request.address)
throw error
})
validateObservationAddress(request.address, observation)
let opening: SessionFollowFrame = {
type: 'snapshot',
header: { ...observation.header },
projections: projections(observation),
...history(observation, spec),
}
listener.setCursor(opening.cursor)
if (assistantListener !== undefined) {
const { baseline: assistantStream, ordinal } = assistantListener.snapshot()
opening = { ...opening, assistantStream }
assistantListener.setOrdinalCut(ordinal)
}
// Increments come from the queue, so the opening ends this lease in every branch;
// activation continues on its own retained lease. Promotion starts before the opening is
// delivered, so a consumer that stops after the snapshot still gets the activation it asked for.
if (request.activate === true && !signal.aborted
&& observation.header.origin !== 'subagent' && observation.source === 'prepared') {
const promotion = observation.retain()
observation[Symbol.dispose]()
try {
this.promote(promotion)
} catch (error: unknown) {
promotion[Symbol.dispose]()
throw error
}
} else {
observation[Symbol.dispose]()
}
yield opening
yield* queue.items()
}
}
packages/api/session-controller/tsconfig.host.json
+12
View File
@@ -16,12 +16,24 @@
"src/control.ts",
"src/file-references.ts",
"src/history.ts",
"src/history/controller.ts",
"src/history/listener.ts",
"src/history/assistant-stream.ts",
"src/state/controller.ts",
"src/state/listener.ts",
"src/web/controller.ts",
"src/list.ts",
"src/media-references.ts",
"src/archived-session-gate.ts",
"src/skill-catalog.ts"
],
"references": [
{
"path": "../gateway/tsconfig.host.json"
},
{
"path": "../../util/timeout"
},
{
"path": "../../../vendor/cordis"
},
pnpm-lock.yaml
Generated
+3
View File
@@ -1574,6 +1574,9 @@ importers:
'@deepseek-ai/dsh-deque':
specifier: workspace:*
version: link:../../util/deque
'@deepseek-ai/dsh-timeout':
specifier: workspace:*
version: link:../../util/timeout
'@deepseek-ai/schemastery':
specifier: link:../../../vendor/schemastery
version: link:../../../vendor/schemastery