Merge remote-tracking branch 'origin/master' into xtr/durable-inbox-recovery

# Conflicts:
#	docs/event-producer-consumer.i18n.yaml
#	docs/event-producer-consumer.md
#	docs/event-producer-consumer.zh.md
#	docs/module-graph.i18n.yaml
#	docs/module-graph.md
#	docs/module-graph.zh.md
#	packages/api/session-controller/tests/session-projections.host.spec.ts
This commit is contained in:
_Kerman committed 2026-08-25 13:20:42 +08:00
commit e46a2499df
2188 files changed
+36021 -15266

No files matched your search

+158 -45
View File
@@ -7,12 +7,13 @@ import type {
Agent, AgentOptions, AgentSetup, ModelSelection as AgentModelSelection, ModelSelectionRef,
} from '@deepseek-ai/dsh-agent'
import type {} from '@deepseek-ai/dsh-agent-default-model'
import { resolveSessionPreset } from '@deepseek-ai/dsh-agent-presets'
import type {} from '@deepseek-ai/dsh-agent-presets'
import { ReasoningEffortId } from '@deepseek-ai/dsh-llm'
import type { Session, SessionEvent, SessionHeader, SessionId } from '@deepseek-ai/dsh-session'
import type {} from '@deepseek-ai/dsh-session-persistence'
import { SessionQueryError, type SessionObservation } from '@deepseek-ai/dsh-session-query'
import { TypertLookupFailure } from '@deepseek-ai/dsh-typert-protocol'
import type {} from '@deepseek-ai/dsh-typert-registry'
import type { SessionError } from './types.ts'
import type { ModelSelection, SessionError } from './types.ts'
/** Cold Session identity absent from persistence. */
export class ApiSessionNotFound extends Error {}
@@ -66,7 +67,10 @@ export type ApiSessionAgentResult =
| { readonly agent: Agent }
| { readonly error: ApiSessionAgentError }
type InstalledSelection = ModelSelectionRef & { current: AgentModelSelection }
type InstalledSelection = ModelSelectionRef & {
current: AgentModelSelection
consume(provider: string, model: string, reasoningEffort: string | undefined): boolean
}
/**
* Test whether generic Session routing must leave an identity to subagent routing.
@@ -112,19 +116,22 @@ export async function inspectApiSession(
sessionId: SessionId,
signal?: AbortSignal,
): Promise<{ meta: SessionHeader; events: SessionEvent[] }> {
const persistence = ctx.get('sessionPersistence')
if (persistence === undefined) {
throw new Error('session persistence is not configured (load a dsh-session-persistence backend)')
try {
using observation = await ctx.sessionQuery.observeSession(sessionId, {
...(signal === undefined ? {} : { signal }),
projectionMode: 'none',
})
if (observation.header.cwd === undefined) {
throw new ApiSessionNotFound(`session "${sessionId}" not found`)
}
return { meta: observation.header, events: [...observation.events] }
} catch (error: unknown) {
if (error instanceof SessionQueryError
&& error.code === 'SESSION_QUERY_SESSION_NOT_FOUND') {
throw new ApiSessionNotFound(`session "${sessionId}" not found`)
}
throw error
}
const meta = (await persistence.list(signal)).find(candidate => candidate.id === sessionId)
if (meta === undefined || meta.cwd === undefined) {
throw new ApiSessionNotFound(`session "${sessionId}" not found`)
}
const inspected = await persistence.inspect(sessionId, signal)
if (inspected.meta.cwd === undefined) {
throw new ApiSessionNotFound(`session "${sessionId}" not found`)
}
return { meta: inspected.meta, events: [...inspected.events] }
}
/** Owns every operation that may create, resume, or configure a Web Agent. */
@@ -159,6 +166,22 @@ export class ApiSessionAgentController {
* @returns the live Agent or a stable Session-domain failure.
*/
async resolveAgent(sessionId: SessionId): Promise<ApiSessionAgentResult> {
return this.resolve(sessionId)
}
/**
* Resolve one ordinary Session from an already-retained exact observation.
* @param observation - Host-owned observation whose preparation stays pinned through setup.
* @returns the live Agent or a stable Session-domain failure.
*/
async resolveObservedAgent(observation: SessionObservation): Promise<ApiSessionAgentResult> {
return this.resolve(observation.header.id, observation)
}
private async resolve(
sessionId: SessionId,
observation?: SessionObservation,
): Promise<ApiSessionAgentResult> {
const live = this.liveAgent(sessionId)
if (live !== undefined) return live
const attached = this.ctx.sessions.get(sessionId)
@@ -168,7 +191,7 @@ export class ApiSessionAgentController {
let resume = this.resumes.get(sessionId)
if (resume === undefined) {
resume = this.resume(sessionId).finally(() => { this.resumes.delete(sessionId) })
resume = this.resume(sessionId, observation).finally(() => { this.resumes.delete(sessionId) })
this.resumes.set(sessionId, resume)
}
try {
@@ -240,7 +263,9 @@ export class ApiSessionAgentController {
if (hasApiSessionSubagentOwner(this.ctx, agent.session, agent)) {
throw new ApiSessionSubagentOwnership(sessionId)
}
this.assertPresetUnchanged(sessionId, presetId, resolveSessionPreset(agent.session))
if (presetId !== undefined) {
this.assertPresetUnchanged(sessionId, presetId, this.presetForSession(agent.session))
}
if (agent.session.header.cwd !== cwd) {
throw new ApiSessionCwdConflict(sessionId, cwd, agent.session.header.cwd)
}
@@ -255,7 +280,13 @@ export class ApiSessionAgentController {
selectionFor(agent: Agent): InstalledSelection {
const installed = this.selections.get(agent)
if (installed !== undefined) return installed
let picked: AgentModelSelection | undefined
const projectionState = this.ctx.sessionProjections.stateOf(agent.session, 'modelSelection')
if (projectionState === undefined) {
throw new Error('api-session: required modelSelection projection is not registered')
}
let picked = projectionState.pending === null
? undefined
: agentModelSelection(projectionState.pending)
const defaultModel = this.ctx.agentDefaultModel
const selection: InstalledSelection = {
get current(): AgentModelSelection {
@@ -271,6 +302,13 @@ export class ApiSessionAgentController {
set current(next: AgentModelSelection) {
picked = next
},
consume(provider: string, model: string, reasoningEffort: string | undefined): boolean {
if (picked?.provider !== provider
|| picked.model !== model
|| picked.reasoningEffort !== reasoningEffort) return false
picked = undefined
return true
},
assembled: undefined,
}
installModelSelection(agent.ctx, selection)
@@ -278,6 +316,42 @@ export class ApiSessionAgentController {
return selection
}
/**
* Commit and cache one validated selection for the next prompt assembly.
* @param agent - live Agent that owns the selection.
* @param selection - validated selection to record and apply.
*/
selectForNextRequest(agent: Agent, selection: AgentModelSelection): void {
agent.session.append('model/selection', selection)
this.selectionFor(agent).current = selection
}
/**
* Let a matching durable request header retire the execution cache.
* @param agent - live Agent whose request was recorded.
* @param provider - provider route used by the request.
* @param model - provider-owned model used by the request.
* @param reasoningEffort - adapter-owned effort used by the request.
* @returns whether the pending selection was consumed.
*/
consumeSelection(
agent: Agent,
provider: string,
model: string,
reasoningEffort: string | undefined,
): boolean {
return this.selections.get(agent)?.consume(provider, model, reasoningEffort) ?? false
}
/**
* Read the current Agent preset from the Session projection.
* @param session - live Session whose projection state is available.
* @returns the current preset, or undefined when the capability is absent.
*/
presetForSession(session: Session): string | undefined {
return this.ctx.sessionProjections.stateOf(session, 'agentPreset') ?? undefined
}
/**
* Serialize image admission and model selection for one Agent.
* @param agent - live Agent that owns the serialization chain.
@@ -319,15 +393,31 @@ export class ApiSessionAgentController {
: { agent }
}
private async resume(sessionId: SessionId): Promise<Agent> {
const inspected = await inspectApiSession(this.ctx, sessionId)
if (hasApiSessionSubagentOwner(this.ctx, { header: inspected.meta }, undefined)) {
private async resume(sessionId: SessionId, supplied?: SessionObservation): Promise<Agent> {
if (supplied !== undefined) return this.resumeObserved(sessionId, supplied)
try {
using observation = await this.ctx.sessionQuery.observeSession(sessionId)
return await this.resumeObserved(sessionId, observation)
} catch (error: unknown) {
if (error instanceof SessionQueryError
&& error.code === 'SESSION_QUERY_SESSION_NOT_FOUND') {
throw new ApiSessionNotFound(`session "${sessionId}" not found`)
}
throw error
}
}
private async resumeObserved(
sessionId: SessionId,
observation: SessionObservation,
): Promise<Agent> {
if (observation.header.id !== sessionId || observation.header.cwd === undefined) {
throw new ApiSessionNotFound(`session "${sessionId}" not found`)
}
if (hasApiSessionSubagentOwner(this.ctx, { header: observation.header }, undefined)) {
throw new ApiSessionSubagentOwnership(sessionId)
}
const composition = await this.composeAgent(resolveSessionPreset({
header: inspected.meta,
events: inspected.events,
}))
const composition = await this.composeAgent(this.presetForObservation(observation))
const published = this.ctx.sessions.get(sessionId)
const live = this.ctx.agents.get(sessionId)
if (published !== undefined && hasApiSessionSubagentOwner(this.ctx, published, live)) {
@@ -353,26 +443,27 @@ export class ApiSessionAgentController {
}
if (live !== undefined) return live
const persistence = checkPersistedIdentity ? this.ctx.get('sessionPersistence') : undefined
const stored = persistence === undefined
? undefined
: (await persistence.list()).find(header => header.id === sessionId)
if (persistence !== undefined && stored !== undefined) {
const inspected = await persistence.inspect(sessionId)
if (hasApiSessionSubagentOwner(this.ctx, { header: inspected.meta }, undefined)) {
throw new ApiSessionSubagentOwnership(sessionId)
if (checkPersistedIdentity) {
try {
using observation = await this.ctx.sessionQuery.observeSession(sessionId)
if (hasApiSessionSubagentOwner(this.ctx, { header: observation.header }, undefined)) {
throw new ApiSessionSubagentOwnership(sessionId)
}
if (observation.header.cwd !== cwd) {
throw new ApiSessionCwdConflict(sessionId, cwd, observation.header.cwd)
}
const storedPreset = this.presetForObservation(observation)
this.assertPresetUnchanged(sessionId, presetId, storedPreset)
const composition = await this.composeAgent(storedPreset)
return (await this.ctx.agents.resume({
resumeSessionId: sessionId,
agentOptions: this.agentOptions(),
setup: composition.setup,
})).agent
} catch (error: unknown) {
if (!(error instanceof SessionQueryError)
|| error.code !== 'SESSION_QUERY_SESSION_NOT_FOUND') throw error
}
if (inspected.meta.cwd !== cwd) {
throw new ApiSessionCwdConflict(sessionId, cwd, inspected.meta.cwd)
}
const storedPreset = resolveSessionPreset({ header: inspected.meta, events: inspected.events })
this.assertPresetUnchanged(sessionId, presetId, storedPreset)
const composition = await this.composeAgent(storedPreset)
return (await this.ctx.agents.resume({
resumeSessionId: sessionId,
agentOptions: this.agentOptions(),
setup: composition.setup,
})).agent
}
try {
@@ -403,6 +494,18 @@ export class ApiSessionAgentController {
this.selectionFor(agent)
}
/**
* Read the current Agent preset from an all-projections observation.
* @param observation - exact Session observation carrying its projection snapshot.
* @returns the current preset, or undefined when the capability is absent.
*/
presetForObservation(observation: SessionObservation): string | undefined {
if (observation.projections === undefined) {
throw new Error('api-session: Agent activation requires a projected Session observation')
}
return observation.projections.values.agentPreset ?? undefined
}
private assertPresetUnchanged(
sessionId: SessionId,
requested: string | undefined,
@@ -412,3 +515,13 @@ export class ApiSessionAgentController {
throw new ApiSessionPresetConflict(sessionId, requested, existing)
}
}
function agentModelSelection(selection: ModelSelection): AgentModelSelection {
return {
provider: selection.provider,
model: selection.model,
...(selection.reasoningEffort === undefined
? {}
: { reasoningEffort: ReasoningEffortId(selection.reasoningEffort) }),
}
}
+11 -7
View File
@@ -2,21 +2,23 @@
import type { Context } from '@deepseek-ai/cordis'
import type {
ModelCatalogFailure,
ModelProviderGroup,
ModelCatalog,
ModelReasoning,
ModelSelection,
} from './types.ts'
/**
* Build the browser model catalog without requiring a Session.
* @param ctx - Host context carrying the live LLM registry.
* @param defaultSelection - deployment default used before a Session selects a model.
* @returns successful non-empty provider groups and isolated provider failures.
*/
export async function buildModelCatalog(ctx: Context): Promise<{
readonly groups: ModelProviderGroup[]
readonly failures: ModelCatalogFailure[]
}> {
const catalog = await Promise.all(ctx.llm.listProviders().map(async (provider) => {
export async function buildModelCatalog(
ctx: Context,
defaultSelection: ModelSelection = ctx.agentDefaultModel.currentSelection(),
): Promise<ModelCatalog> {
const providers = ctx.llm.listProviders()
const catalog = await Promise.all(providers.map(async (provider) => {
try {
const models = await ctx.llm.listModels(provider.id)
const entries = await Promise.all(models.map(async (model) => {
@@ -56,6 +58,8 @@ export async function buildModelCatalog(ctx: Context): Promise<{
}
}))
return {
default: { ...defaultSelection },
routableProviders: providers.map(provider => provider.id),
groups: catalog.flatMap(item => item.kind === 'group' ? [item.group] : [])
.filter(group => group.models.length > 0),
failures: catalog.flatMap(item => item.kind === 'failure' ? [item.failure] : []),
@@ -66,14 +66,6 @@ export interface ISessions {
*/
refreshSubagents(parentSessionId: SessionId): Promise<void>
/**
* Record the composition one session now runs. The agent-preset seat calls
* this after a successful blank-session switch, so the header label moves
* with the composition instead of waiting for the next full list refresh.
* @param sessionId - the switched session.
* @param agentPreset - the preset id the host confirmed.
*/
noteAgentPreset(sessionId: SessionId, agentPreset: string): void
/** Clear the current selection into the no-session view state. */
clear(): void
/**
@@ -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
@@ -25,8 +25,6 @@ export interface SessionListEntry {
/** Coarse durable origin for navigation filtering; not a continuation capability. */
origin?: 'subagent'
cwd?: string
/** Agent preset the session's agent was composed from (summary passthrough). */
agentPreset?: string
/** Current host-computed projection values for list consumers. */
projectionValues?: Readonly<Partial<SessionProjectionMap>>
/** Finished running while not selected and not yet opened — the sidebar's green "done" reminder (clears on select or the next run). */
@@ -61,11 +61,19 @@ export interface SessionListSnapshot {
}
/** One parent-addressed durable catalog projected through the sessions snapshot. */
export interface SubagentCatalogSnapshot extends SubagentCatalog {
export type SubagentCatalogSnapshot = Omit<SubagentCatalog, 'parentAvailable'> & {
/** Absent until the first successful catalog read. */
readonly parentAvailable?: boolean
state: 'loading' | 'ready' | 'error'
error: ClientFailure | null
}
function catalogAvailability(parentAvailable: boolean | undefined): {
readonly parentAvailable?: boolean
} {
return parentAvailable === undefined ? {} : { parentAvailable }
}
interface CatalogInflight {
readonly promise: Promise<void>
readonly expandableRows: Set<SessionId>
@@ -166,8 +174,8 @@ export class SessionManager {
this.sessions.get(sessionId)?.configureSubagent(
address,
address === undefined
? false
: this.catalogs.get(address.parentSessionId)?.parentAvailable ?? false,
? undefined
: this.catalogs.get(address.parentSessionId)?.parentAvailable,
)
this.selected = sessionId
// Looking at the session consumes its completion reminder (dot clears).
@@ -187,7 +195,7 @@ export class SessionManager {
throw new Error(`sessions.selectSubagent: ${address.childSessionId} is not a healthy catalog child`)
}
this.addresses.set(address.childSessionId, address)
this.sessions.get(address.childSessionId)?.configureSubagent(address, catalog?.parentAvailable ?? false)
this.sessions.get(address.childSessionId)?.configureSubagent(address, catalog?.parentAvailable)
this.selected = address.childSessionId
this.completedNotifications.delete(address.childSessionId)
void this.refreshSubagents(address.childSessionId)
@@ -310,10 +318,13 @@ export class SessionManager {
private createSession(sessionId: SessionId): Session {
const address = this.addresses.get(sessionId)
const parentAvailable = address === undefined
? undefined
: this.catalogs.get(address.parentSessionId)?.parentAvailable
return new Session(sessionId, this.api, this.remote, {
...(address === undefined ? {} : {
address,
parentAvailable: this.catalogs.get(address.parentSessionId)?.parentAvailable ?? false,
...catalogAvailability(parentAvailable),
}),
// The sender's local first-send flip mirrors into the list row so the
// session surfaces (lists filter on blank) before any host frame lands.
@@ -349,7 +360,9 @@ export class SessionManager {
const activityRows = new Map<SessionId, 'running' | 'inactive'>()
this.catalogs.set(parentSessionId, {
entries: previous?.entries ?? [],
parentAvailable: previous?.parentAvailable ?? false,
...(previous?.parentAvailable === undefined
? {}
: { parentAvailable: previous.parentAvailable }),
state: 'loading',
error: null,
})
@@ -376,8 +389,10 @@ export class SessionManager {
entries: this.withCatalogMutations(
previous?.entries ?? [], expandableRows, activityRows,
),
parentAvailable: this.catalogInflight.get(parentSessionId)?.parentAvailableOverride
?? previous?.parentAvailable ?? false,
...catalogAvailability(
this.catalogInflight.get(parentSessionId)?.parentAvailableOverride
?? previous?.parentAvailable,
),
state: 'error',
error: result.error,
})
@@ -388,8 +403,10 @@ export class SessionManager {
entries: this.withCatalogMutations(
previous?.entries ?? [], expandableRows, activityRows,
),
parentAvailable: this.catalogInflight.get(parentSessionId)?.parentAvailableOverride
?? previous?.parentAvailable ?? false,
...catalogAvailability(
this.catalogInflight.get(parentSessionId)?.parentAvailableOverride
?? previous?.parentAvailable,
),
state: 'error',
error: folded.ok ? null : folded.error,
})
@@ -555,7 +572,6 @@ export class SessionManager {
this.recordMutation({ kind: 'upsert', summary: {
sessionId: result.value.sessionId, updatedAt: Date.now(), running: false, blank: true,
...(opts.cwd !== undefined ? { cwd: opts.cwd } : {}),
...(result.value.agentPreset !== undefined ? { agentPreset: result.value.agentPreset } : {}),
} })
} else {
const publishedSessionId = workspaceAttachSessionId(result.error)
@@ -621,17 +637,6 @@ export class SessionManager {
this.recordMutation({ kind: 'upsert', summary })
}
/**
* Record a host-confirmed composition switch (see ISessions.noteAgentPreset).
* @param sessionId - the switched session.
* @param agentPreset - the preset id the host confirmed.
*/
noteAgentPreset(sessionId: SessionId, agentPreset: string): void {
this.recordMutation({ kind: 'upsert', summary: {
sessionId, updatedAt: Date.now(), running: false, blank: true, agentPreset,
} })
}
/** Apply immediately and retain for replay when a list response is in flight. */
private recordMutation(mutation: SessionListMutation): void {
this.listMutations?.push(mutation)
@@ -932,7 +937,7 @@ export class SessionManager {
const prev = this.entryCache.get(entry.sessionId)
if (
prev !== undefined && prev.updatedAt === entry.updatedAt && prev.running === entry.running
&& prev.blank === entry.blank && prev.agentPreset === entry.agentPreset
&& prev.blank === entry.blank
&& prev.parentSessionId === entry.parentSessionId && prev.cwd === entry.cwd
&& prev.origin === entry.origin && prev.title === entry.title && prev.depth === entry.depth
&& prev.projectionValues === entry.projectionValues
@@ -980,15 +985,10 @@ function applyMutation(summaries: readonly SessionSummary[], mutation: SessionLi
? { parentSessionId: mutation.summary.parentSessionId } : {}),
...(existing.origin === undefined && mutation.summary.origin !== undefined
? { origin: mutation.summary.origin } : {}),
// Newest wins, not fill-only: a blank-session preset switch replaces
// the creation-time value, and every producer of this field (the
// create echo, the select echo, a list row) reports the CURRENT one.
...(mutation.summary.agentPreset !== undefined
? { agentPreset: mutation.summary.agentPreset } : {}),
}
if (filled.cwd === existing.cwd && filled.parentSessionId === existing.parentSessionId
&& filled.origin === existing.origin && filled.blank === existing.blank
&& filled.agentPreset === existing.agentPreset) return [...summaries]
) return [...summaries]
return summaries.map(summary => summary.sessionId === mutation.summary.sessionId ? filled : summary)
}
case 'remove':
@@ -2,8 +2,8 @@
* Generic per-session projection value store (push model; see the
* session-projection subsystem page, docs/subsystems/session-projection.md):
* the host is the only computation site; the client holds finished
* whole values per key — `key → { value, seq }` — seeded by a Session page's
* projections block and updated by Session Controller `projection` frames,
* whole values per key — `key → { value, seq }` — seeded by a follow opening
* baseline and updated by Session Controller `projection` frames,
* under the single rule **higher seq wins**. No client-side domain folding
* exists: a domain ships projection support with zero client code. Per-key
* bare observable faces feed `useProjection` (ui-renderer binds them).
@@ -40,8 +40,7 @@ export type UseProjection = {
}
/**
* Tail-page projections baseline — structurally identical to Session
* Controller's `SessionProjectionsBlock`, restated here so the
* Follow-opening projection baseline, restated here so the
* React-free store depends only on the type table, not the wire package's
* response vocabulary.
*/
@@ -45,12 +45,6 @@ export interface SessionSummary {
/** Human-facing label: durable title, project basename, then session id. */
displayTitle: string
cwd?: string
/**
* Agent preset this session's agent was composed from; absent when the
* deployment composes no presets. The session header labels what the
* session actually runs rather than the deployment's current default.
*/
agentPreset?: string
parentId?: SessionId
/** Coarse durable origin for navigation filtering; not a continuation capability. */
origin?: 'subagent'
@@ -317,10 +311,6 @@ export class ClientSessions implements ISessions {
return this.manager.refreshSubagents(parentSessionId)
}
noteAgentPreset(sessionId: SessionId, agentPreset: string): void {
this.manager.noteAgentPreset(sessionId, agentPreset)
}
/**
* Clear the current selection so the layout shows the no-session empty
* state (new-session affordance and the workspace preselection flow).
@@ -611,7 +601,6 @@ export class ClientSessions implements ISessions {
...(entry.cwd !== undefined ? { cwd: entry.cwd } : {}),
...(entry.parentSessionId !== undefined ? { parentId: entry.parentSessionId } : {}),
...(entry.origin !== undefined ? { origin: entry.origin } : {}),
...(entry.agentPreset !== undefined ? { agentPreset: entry.agentPreset } : {}),
}
}
if (current !== undefined && currentAddress !== undefined) {
@@ -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,
+18 -35
View File
@@ -3,9 +3,7 @@
import { randomUUID } from 'node:crypto'
import type { Context } from '@deepseek-ai/cordis'
import type { Agent, ModelSelection as AgentModelSelection } from '@deepseek-ai/dsh-agent'
import {
PresetMountError, UnknownPresetError, resolveSessionPreset,
} from '@deepseek-ai/dsh-agent-presets'
import { PresetMountError, UnknownPresetError } from '@deepseek-ai/dsh-agent-presets'
import { AttachmentError, admitEncodedImages } from '@deepseek-ai/dsh-attachment'
import type { ImageAttachmentRef } from '@deepseek-ai/dsh-attachment'
import {
@@ -13,7 +11,8 @@ import {
} from '@deepseek-ai/dsh-llm'
import type { ContentBlock, MessageSource } from '@deepseek-ai/dsh-llm'
import { SessionId } from '@deepseek-ai/dsh-session'
import type { Session, SessionEvent, SessionHeader, UserMessage } from '@deepseek-ai/dsh-session'
import type { SessionEvent, SessionHeader, UserMessage } from '@deepseek-ai/dsh-session'
import { SessionQueryError, type SessionObservation } from '@deepseek-ai/dsh-session-query'
import { SessionTitleInvalidError } from '@deepseek-ai/dsh-session-title'
import { TypertRemoteFailure } from '@deepseek-ai/dsh-typert-protocol'
import type { Workspace } from '@deepseek-ai/dsh-workspace'
@@ -27,7 +26,6 @@ import {
hasApiSessionSubagentOwner,
inspectApiSession,
} from './agent.ts'
import { buildModelCatalog } from './catalog.ts'
import type {
SessionAttachmentRequest,
SessionAttachmentValue,
@@ -37,8 +35,6 @@ import type {
SessionCreateValue,
SessionForkRequest,
SessionForkValue,
SessionModels,
SessionModelsRequest,
SessionPromptRequest,
SessionPromptValue,
SessionRenameRequest,
@@ -110,27 +106,10 @@ export class SessionCommandController {
)
}
}
const agentPreset = resolveSessionPreset(adopted.session)
const agentPreset = this.agents.presetForSession(adopted.session)
return { sessionId, ...(agentPreset === undefined ? {} : { agentPreset }) }
}
/**
* Read the current selection and advisory model catalog, explicitly resuming the Session.
* @param request - Session whose model state is requested.
* @returns the current selection and available model groups.
*/
async models(request: SessionModelsRequest): Promise<SessionModels> {
const agent = await this.resolveAgent(request.sessionId)
const current = this.agents.selectionFor(agent).current
const { groups, failures } = await buildModelCatalog(this.ctx)
return {
current: { ...current },
routable: routeServed(this.ctx, current.provider),
groups,
failures,
}
}
/**
* Validate and install one Session-local model selection.
* @param request - Session identity and requested model selection.
@@ -154,7 +133,7 @@ export class SessionCommandController {
? {}
: { reasoningEffort: resolved.reasoningEffort }),
}
this.agents.selectionFor(agent).current = selected
this.agents.selectForNextRequest(agent, selected)
try {
await this.ctx.agentDefaultModel.saveSelection(selected)
} catch (error) {
@@ -210,12 +189,15 @@ export class SessionCommandController {
&& (!Number.isInteger(request.atSeq) || request.atSeq < 0)) {
reject('bad-request', 'atSeq must be a non-negative integer', {})
}
let source: SessionReadState
let observed: SessionObservation
try {
source = await this.readSessionState(request.sessionId)
observed = await this.ctx.sessionQuery.observeSession(request.sessionId)
} catch (error) {
if (error instanceof ApiSessionNotFound) {
reject('session-not-found', error.message, { sessionId: request.sessionId })
if (error instanceof SessionQueryError
&& error.code === 'SESSION_QUERY_SESSION_NOT_FOUND') {
reject('session-not-found', `session "${request.sessionId}" not found`, {
sessionId: request.sessionId,
})
}
reject(
'internal',
@@ -223,6 +205,7 @@ export class SessionCommandController {
{},
)
}
using source = observed
const lastSeq = source.events.at(-1)?.seq ?? -1
const atSeq = request.atSeq
const anchoredBoundary = atSeq === undefined
@@ -245,7 +228,7 @@ export class SessionCommandController {
while (cut < source.events.length && source.events[cut]?.type !== 'turn/start') cut++
let workspace: Workspace | undefined
try {
workspace = await this.forkWorkspace(source)
workspace = await this.forkWorkspace(source.header)
} catch (error) {
reject(
'internal',
@@ -254,7 +237,7 @@ export class SessionCommandController {
)
}
const childId = SessionId(`session-${randomUUID()}`)
const composition = await this.agents.composeAgent(resolveSessionPreset(source))
const composition = await this.agents.composeAgent(this.agents.presetForObservation(source))
try {
const { provider, model } = this.ctx.agentDefaultModel.currentSelection()
await this.ctx.agents.create({
@@ -262,7 +245,7 @@ export class SessionCommandController {
seed: source.events.slice(0, cut),
meta: {
...(source.header.cwd === undefined ? {} : { cwd: source.header.cwd }),
parentSession: source.id,
parentSession: source.header.id,
seedLength: cut,
...(composition.agentPreset === undefined
? {}
@@ -507,10 +490,10 @@ export class SessionCommandController {
return { id: inspected.meta.id, header: inspected.meta, events: inspected.events }
}
private async forkWorkspace(source: Pick<Session, 'id' | 'header'>): Promise<Workspace | undefined> {
private async forkWorkspace(source: SessionHeader): Promise<Workspace | undefined> {
const workspaces = this.ctx.workspaceRegistry.list()
const direct = workspaces.find(workspace => workspace.sessionIds.includes(source.id))
if (direct !== undefined || source.header.origin !== 'subagent') return direct
if (direct !== undefined || source.origin !== 'subagent') return direct
const lineage = await this.ctx.sessionQuery.traceSession(source.id)
for (const ancestor of lineage.ancestors) {
const workspace = workspaces.find(candidate => candidate.sessionIds.includes(ancestor.header.id))
@@ -10,7 +10,7 @@ import type {
SessionControlBaseline,
SessionControlFrame,
SessionJob,
SessionProjectionsBlock,
SessionProjectionBaseline,
SessionProjectionValues,
SessionQueuedItem,
} from './types.ts'
@@ -22,10 +22,6 @@ export class SessionControlController {
/** @param ctx - Host context carrying live Agent, projection, and jobs services. */
constructor(private readonly ctx: Context) {
ctx.on('session/event', (session, event) => { this.onSessionEvent(session, event) })
ctx.on('session/created', (session) => {
const jobs = this.jobsFor(this.ctx.agents.get(session.id))
if (jobs.length > 0) this.broadcast({ type: 'jobs', sessionId: session.id, jobs })
})
ctx.inject(['sessionProjections'], (projectionCtx) => {
projectionCtx.sessionProjections.onChanged((session, key, value, seq) => {
this.broadcast({
@@ -40,6 +36,10 @@ export class SessionControlController {
ctx.inject(['jobs'], (jobsCtx) => {
jobsCtx.jobs.onJobsChanged((owner) => { this.onJobsChanged(owner) })
})
ctx.on('session/created', (session) => {
const jobs = this.jobsFor(this.ctx.agents.get(session.id))
if (jobs.length > 0) this.broadcast({ type: 'jobs', sessionId: session.id, jobs })
})
ctx.effect(() => () => {
for (const stream of this.streams) stream.end()
this.streams.clear()
@@ -82,9 +82,9 @@ export class SessionControlController {
private projectionBaseline(
sessions: readonly Session[],
): Readonly<Record<SessionId, SessionProjectionsBlock>> {
): Readonly<Record<SessionId, SessionProjectionBaseline>> {
const registry = this.ctx.get('sessionProjections')
const blocks = Object.create(null) as Record<SessionId, SessionProjectionsBlock>
const blocks = Object.create(null) as Record<SessionId, SessionProjectionBaseline>
for (const session of sessions) {
const snapshot = registry?.snapshot(session)
blocks[session.id] = snapshot === undefined
+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 {
+37 -17
View File
@@ -4,6 +4,7 @@ import { Context } from '@deepseek-ai/cordis'
import z from '@deepseek-ai/schemastery'
import { errorChain } from '@deepseek-ai/dsh-llm'
import type { SessionEvent, SessionHeader, SessionId } from '@deepseek-ai/dsh-session'
import type { SessionObservation } from '@deepseek-ai/dsh-session-query'
import { Remote, TypertRemoteService } from '@deepseek-ai/dsh-typert-protocol'
import {
ApiSessionAgentController,
@@ -14,6 +15,7 @@ import { SessionCommandController } from './commands.ts'
import { SessionControlController } from './control.ts'
import { SessionHistoryController } from './history.ts'
import { ApiSessionList, DEFAULT_COLD_BLANK_PROBE_MAX_BYTES } from './list.ts'
import { installModelSelectionProjection } from './model-selection-projection.ts'
import type {
SessionAttachmentRequest,
SessionAttachmentValue,
@@ -28,8 +30,6 @@ import type {
SessionForkValue,
SessionListRequest,
SessionListValue,
SessionModels,
SessionModelsRequest,
SessionPage,
SessionPageRequest,
SessionPromptRequest,
@@ -56,7 +56,7 @@ declare module '@deepseek-ai/cordis' {
/** Session Controller deployment policy. */
export interface Config {
/** Maximum cold Session artifact size read to determine blankness. */
/** Maximum cold Session artifact size eligible for one full projection observation. */
readonly coldBlankProbeMaxBytes?: number
}
@@ -68,6 +68,7 @@ export class SessionController extends TypertRemoteService {
'attachments',
'llm',
'sessions',
'sessionProjections',
'sessionQuery',
'typert',
'workspaceRegistry',
@@ -82,17 +83,24 @@ export class SessionController extends TypertRemoteService {
private readonly controlState: SessionControlController
private readonly history: SessionHistoryController
private readonly listState: ApiSessionList
private readonly promotions = new Set<Promise<void>>()
/**
* @param ctx - Host context containing the Session capability assembly.
* @param config - cold-list read policy.
* @param config - cold-list observation policy.
*/
constructor(ctx: Context, config: Config) {
super(ctx, 'sessionController', { namespace: 'session' })
installModelSelectionProjection(ctx)
this.agents = new ApiSessionAgentController(ctx)
this.commands = new SessionCommandController(ctx, this.agents, process.cwd())
this.controlState = new SessionControlController(ctx)
this.history = new SessionHistoryController(ctx)
// Registered before history so reverse-order teardown closes every
// follower before waiting for already-admitted promotions.
ctx.effect(() => async () => {
await Promise.allSettled([...this.promotions])
}, 'session-controller.promotions')
this.history = new SessionHistoryController(ctx, (observation) => { this.promote(observation) })
this.listState = new ApiSessionList(
ctx,
config.coldBlankProbeMaxBytes ?? DEFAULT_COLD_BLANK_PROBE_MAX_BYTES,
@@ -111,11 +119,33 @@ export class SessionController extends TypertRemoteService {
ctx.emit('api-session/error', agent.id, errorChain(error))
})
ctx.on('session/event', (session, event) => {
if (event.type === 'request/header') {
const agent = ctx.agents.get(session.id)
if (agent?.session === session) this.agents.consumeSelection(
agent,
event.data.header.config.provider,
event.data.header.config.model,
event.data.header.config.reasoningEffort,
)
}
if (event.type !== 'user/message' || event.data.source.kind !== 'user') return
ctx.emit('api-session/activity', session.id, event.time)
})
}
private promote(observation: SessionObservation): void {
const sessionId = observation.header.id
const task = (async () => {
using ownedObservation = observation
const result = await this.agents.resolveObservedAgent(ownedObservation)
if ('error' in result) this.ctx.emit('api-session/error', sessionId, result.error.message)
})().catch((error: unknown) => {
this.ctx.logger.error(`session-controller: background activation for "${sessionId}" failed: ${errorChain(error)}`)
})
this.promotions.add(task)
void task.finally(() => { this.promotions.delete(task) })
}
/**
* Resolve or resume one ordinary Session for another Host API domain.
* @param sessionId - Session identity whose Agent owns the operation.
@@ -174,16 +204,6 @@ export class SessionController extends TypertRemoteService {
return this.commands.create(request)
}
/**
* Read model choices after explicitly resuming the addressed Session.
* @param request - Session whose model state is requested.
* @returns the current selection and available model groups.
*/
@Remote('models')
models(request: SessionModelsRequest): Promise<SessionModels> {
return this.commands.models(request)
}
/**
* Select one Session-local model after explicitly resuming the Session.
* @param request - Session identity and requested model selection.
@@ -260,7 +280,7 @@ export class SessionController extends TypertRemoteService {
* Read one cold-safe, message-aligned Session history page.
* @param request - durable address, backward cursor, and page budget.
* @param signal - cancellation for persistence reads.
* @returns one chronological page and optional latest projections.
* @returns one chronological page.
*/
@Remote('page')
page(request: SessionPageRequest, signal: AbortSignal): Promise<SessionPage> {
@@ -271,7 +291,7 @@ export class SessionController extends TypertRemoteService {
* Follow one Session log from its opening or resume cursor.
* @param request - durable address and last committed sequence already held by the caller.
* @param signal - cancellation owned by the Remote stream carrier.
* @returns an opened cursor followed by gap-free event frames.
* @returns a complete opening snapshot followed by gap-free event frames.
*/
@Remote({ mode: 'stream' })
follow(request: SessionFollowRequest, signal: AbortSignal): AsyncIterable<SessionFollowFrame> {
+93 -112
View File
@@ -2,10 +2,9 @@
import { stat } from 'node:fs/promises'
import type { Context } from '@deepseek-ai/cordis'
import { resolveSessionPreset } from '@deepseek-ai/dsh-agent-presets'
import type {} from '@deepseek-ai/dsh-agent-presets'
import type { ImageAttachmentLimits } from '@deepseek-ai/dsh-attachment'
import type { Session, SessionEvent, SessionHeader, SessionId } from '@deepseek-ai/dsh-session'
import type { SessionPersistence } from '@deepseek-ai/dsh-session-persistence'
import type {} from '@deepseek-ai/dsh-session-projection'
import type {} from '@deepseek-ai/dsh-session-projection-cache'
import { SessionQueryError, type SessionSearchCursor } from '@deepseek-ai/dsh-session-query'
@@ -16,11 +15,11 @@ import {
SESSION_SEARCH_SNIPPET_MAX_CODE_POINTS,
} from './types.ts'
import type {
SessionListMetadata, SessionProjectionsBlock, SessionProjectionValues, SessionSearchItem,
SessionListMetadata, SessionProjectionHints, SessionProjectionValues, SessionSearchItem,
SessionSearchValue, SessionSummary,
} from './types.ts'
/** Default maximum artifact size eligible for one cold blankness read. */
/** Default maximum artifact size eligible for one cold projection observation. */
export const DEFAULT_COLD_BLANK_PROBE_MAX_BYTES = 1024
const COLD_SUMMARY_BATCH_SIZE = 16
@@ -61,17 +60,6 @@ export function applySessionListMetadata(
: { blank, lastPromptAt }
}
/**
* Fold exact list metadata for an attached Session.
* @param events - complete attached Session event log.
* @returns metadata derived from the event prefix.
*/
export function sessionListMetadata(events: readonly SessionEvent[]): SessionListMetadata {
let state: SessionListMetadata = { blank: true, lastPromptAt: null }
for (const event of events) state = applySessionListMetadata(state, event)
return state
}
/**
* Return the longest prefix containing at most `maximum` Unicode code points.
* @param value - source text.
@@ -89,11 +77,11 @@ export function truncateUnicodeCodePoints(value: string, maximum: number): strin
return value
}
/** Owns list projection registration, cold summaries, and authorized search. */
/** Owns list projection registration, bounded cold summaries, and authorized search. */
export class ApiSessionList {
/**
* @param ctx - Host context carrying Session, persistence, and projection services.
* @param coldBlankProbeMaxBytes - maximum physical artifact size read to verify cold blankness.
* @param ctx - Host context carrying Session, query, persistence, and projection services.
* @param coldBlankProbeMaxBytes - maximum physical artifact size eligible for a full observation.
*/
constructor(
private readonly ctx: Context,
@@ -130,14 +118,14 @@ export class ApiSessionList {
* @returns current list metadata and available projections.
*/
summaryFor(session: Session): SessionSummary {
const metadata = sessionListMetadata(session.events)
const projections = this.projectionsFor(session.header, session)
const metadata = projections?.values.sessionListMetadata
return {
sessionId: session.id,
updatedAt: updatedAt(session.header, metadata),
running: this.ctx.agents.get(session.id)?.status === 'running',
blank: metadata.blank,
...listFields(session.header, session.events),
blank: metadata?.blank ?? session.seq === 0,
...listFields(session.header),
...(projections === undefined ? {} : { projections }),
}
}
@@ -149,41 +137,86 @@ export class ApiSessionList {
*/
async list(signal?: AbortSignal): Promise<SessionSummary[]> {
signal?.throwIfAborted()
const items = this.ctx.sessions.list().map(session => this.summaryFor(session))
const attached = new Set(items.map(item => item.sessionId))
const persistence = this.ctx.get('sessionPersistence')
if (persistence !== undefined) {
const cold = (await persistence.list(signal))
.filter(meta => !attached.has(meta.id) && meta.cwd !== undefined)
signal?.throwIfAborted()
for (let offset = 0; offset < cold.length; offset += COLD_SUMMARY_BATCH_SIZE) {
const settled = await Promise.allSettled(cold.slice(offset, offset + COLD_SUMMARY_BATCH_SIZE)
.map(async (meta) => {
const projections = this.projectionsFor(meta, undefined)
const summary = await summarizeCold(
this.ctx,
persistence,
meta,
projections?.values.sessionListMetadata,
this.coldBlankProbeMaxBytes,
signal,
)
const raced = this.ctx.sessions.get(meta.id)
if (raced !== undefined) return this.summaryFor(raced)
return { ...summary, ...(projections === undefined ? {} : { projections }) }
}))
const summaries = settled.map((result) => {
if (result.status === 'rejected') throw result.reason
return result.value
})
signal?.throwIfAborted()
items.push(...summaries)
const records = await this.ctx.sessionQuery.listSessions(signal)
signal?.throwIfAborted()
const items: SessionSummary[] = []
const cold: SessionHeader[] = []
for (const record of records) {
const live = this.ctx.sessions.get(record.header.id)
if (live !== undefined) {
items.push(this.summaryFor(live))
continue
}
if (record.header.cwd === undefined) continue
cold.push(record.header)
}
for (let offset = 0; offset < cold.length; offset += COLD_SUMMARY_BATCH_SIZE) {
const settled = await Promise.allSettled(cold.slice(offset, offset + COLD_SUMMARY_BATCH_SIZE)
.map(header => this.summarizeCold(header, signal)))
for (const result of settled) {
if (result.status === 'rejected') throw result.reason
items.push(result.value)
}
}
items.sort((left, right) => right.updatedAt - left.updatedAt)
return items
}
private async summarizeCold(
header: SessionHeader,
signal: AbortSignal | undefined,
): Promise<SessionSummary> {
const cached = this.projectionsFor(header, undefined)
const projections = cached?.values.sessionListMetadata?.blank === false
? cached
: await this.probeSmallCold(header, signal) ?? cached
const raced = this.ctx.sessions.get(header.id)
if (raced !== undefined) return this.summaryFor(raced)
const metadata = projections?.values.sessionListMetadata
return {
sessionId: header.id,
updatedAt: updatedAt(header, metadata),
running: false,
// A large or inaccessible cache miss remains unknown and visible.
blank: metadata?.blank ?? false,
...listFields(header),
...(projections === undefined ? {} : { projections }),
}
}
private async probeSmallCold(
header: SessionHeader,
signal: AbortSignal | undefined,
): Promise<SessionProjectionHints | undefined> {
if (this.coldBlankProbeMaxBytes === 0) return undefined
const persistence = this.ctx.get('sessionPersistence')
const location = persistence?.locate(header)
if (location === undefined) return undefined
signal?.throwIfAborted()
try {
if ((await stat(location.path)).size > this.coldBlankProbeMaxBytes) return undefined
} catch {
signal?.throwIfAborted()
return undefined
}
try {
using observation = await this.ctx.sessionQuery.observeSession(header.id, {
...(signal === undefined ? {} : { signal }),
projectionMode: 'all',
})
const block = observation.projections
return block === undefined
? undefined
: { asOfSeq: block.asOfSeq, values: block.values as SessionProjectionValues }
} catch (error: unknown) {
signal?.throwIfAborted()
this.ctx.logger.warn(
`api-session.list: small cold observation for "${header.id}" failed; serving it as visible: ${String(error)}`,
)
return undefined
}
}
/**
* Search current visible message content without activating any matching Session.
* @param query - literal message-content query.
@@ -202,10 +235,12 @@ export class ApiSessionList {
)
}
try {
const visible = await this.list(signal)
const visible = await provider.listSessions(signal)
signal.throwIfAborted()
if (visible.length === 0) return { items: [], hasMore: false }
const visibleIds = new Set(visible.map(item => item.sessionId))
const visibleIds = new Set(visible
.filter(record => record.header.cwd !== undefined)
.map(record => record.header.id))
if (visibleIds.size === 0) return { items: [], hasMore: false }
const authorized: SessionSearchItem[] = []
const acceptedIds = new Set<SessionId>()
const seenCursors = new Set<SessionSearchCursor>()
@@ -293,15 +328,16 @@ export class ApiSessionList {
private projectionsFor(
header: SessionHeader,
session: Session | undefined,
): SessionProjectionsBlock | undefined {
): SessionProjectionHints | undefined {
try {
const block = session === undefined
? this.ctx.get('sessionProjectionCache')?.cachedSnapshot(header)
: this.ctx.get('sessionProjections')?.snapshot(session)
: this.ctx.get('sessionProjections')?.cachedSnapshot(session)
return block !== undefined && Object.keys(block.values).length > 0
? {
asOfSeq: block.asOfSeq,
// Projection definitions validate whole JSON values before snapshot publication.
// Listing hints contain every currently cached wire value but remain
// partial: missing cells and cache rows are never materialized here.
values: block.values as SessionProjectionValues,
}
: undefined
@@ -340,69 +376,14 @@ function updatedAt(header: SessionHeader, metadata: SessionListMetadata | undefi
return Math.max(header.createdAt, metadata?.lastPromptAt ?? 0)
}
function listFields(header: SessionHeader, events: readonly SessionEvent[] = []): {
function listFields(header: SessionHeader): {
readonly parentSessionId?: SessionId
readonly origin?: 'subagent'
readonly cwd?: string
readonly agentPreset?: string
} {
const agentPreset = resolveSessionPreset({ header, events })
return {
...(header.parentSession === undefined ? {} : { parentSessionId: header.parentSession }),
...(header.origin === undefined ? {} : { origin: header.origin }),
...(header.cwd === undefined ? {} : { cwd: header.cwd }),
...(agentPreset === undefined ? {} : { agentPreset }),
}
}
async function summarizeCold(
ctx: Context,
persistence: SessionPersistence,
header: SessionHeader,
metadata: SessionListMetadata | undefined,
blankProbeMaxBytes: number,
signal?: AbortSignal,
): Promise<SessionSummary> {
const probed = metadata?.blank === false
? undefined
: await probeColdMetadata(ctx, persistence, header, blankProbeMaxBytes, signal)
return {
sessionId: header.id,
updatedAt: updatedAt(header, probed ?? metadata),
running: false,
blank: metadata?.blank === false ? false : probed?.blank ?? false,
...listFields(header),
}
}
async function probeColdMetadata(
ctx: Context,
persistence: SessionPersistence,
header: SessionHeader,
maxBytes: number,
signal?: AbortSignal,
): Promise<SessionListMetadata | undefined> {
if (maxBytes === 0) return undefined
signal?.throwIfAborted()
const location = persistence.locate(header)
if (location === undefined) return undefined
let size: number
try {
size = (await stat(location.path)).size
} catch {
signal?.throwIfAborted()
return undefined
}
if (size > maxBytes) return undefined
try {
const { events } = await persistence.readFrom(header.id, 0, signal)
signal?.throwIfAborted()
return sessionListMetadata(events)
} catch (error) {
signal?.throwIfAborted()
ctx.logger.warn(
`api-session.list: blank probe for "${header.id}" failed; serving it as visible: ${String(error)}`,
)
return undefined
}
}
@@ -0,0 +1,83 @@
/** Durable model-selection intent and request-use projection. */
import type { Context } from '@deepseek-ai/cordis'
import type { SessionEvent } from '@deepseek-ai/dsh-session'
import type { ProjectionDefinition } from '@deepseek-ai/dsh-session-projection'
import { z } from 'zod'
import type {
ModelSelection,
ModelSelectionProjection,
ModelSelectionProjectionState,
} from './types.ts'
const modelSelectionSchema = z.object({
provider: z.string().min(1),
model: z.string().min(1),
reasoningEffort: z.string().min(1).optional(),
}) as unknown as z.ZodType<ModelSelection>
const modelSelectionProjectionStateSchema = z.object({
lastUsed: modelSelectionSchema.nullable(),
pending: modelSelectionSchema.nullable(),
}) as unknown as z.ZodType<ModelSelectionProjectionState>
const modelSelectionProjectionSchema = z.object({
lastUsed: modelSelectionSchema.nullable(),
next: modelSelectionSchema.nullable(),
}) as unknown as z.ZodType<ModelSelectionProjection>
/**
* Advance durable model-selection state by one Session event.
* @param state - selection state before the event.
* @param event - next committed Session event.
* @returns the original or advanced selection state.
*/
function applyModelSelectionProjection(
state: ModelSelectionProjectionState,
event: SessionEvent,
): ModelSelectionProjectionState {
if (event.type === 'model/selection') {
return sameSelection(state.pending, event.data)
? state
: { lastUsed: state.lastUsed, pending: event.data }
}
if (event.type !== 'request/header') return state
const lastUsed: ModelSelection = {
provider: event.data.header.config.provider,
model: event.data.header.config.model,
...(event.data.header.config.reasoningEffort === undefined
? {}
: { reasoningEffort: String(event.data.header.config.reasoningEffort) }),
}
const pending = sameSelection(state.pending, lastUsed) ? null : state.pending
return sameSelection(state.lastUsed, lastUsed) && pending === state.pending
? state
: { lastUsed, pending }
}
const modelSelectionProjection = {
key: 'modelSelection',
stateSchema: modelSelectionProjectionStateSchema,
init: () => ({ lastUsed: null, pending: null }),
apply: applyModelSelectionProjection,
wire: {
viewSchema: modelSelectionProjectionSchema,
view: state => ({ lastUsed: state.lastUsed, next: state.pending ?? state.lastUsed }),
},
stateVersion: 2,
} satisfies ProjectionDefinition<'modelSelection', ModelSelectionProjectionState>
function sameSelection(left: ModelSelection | null, right: ModelSelection | null): boolean {
return left === right || (left !== null && right !== null
&& left.provider === right.provider
&& left.model === right.model
&& left.reasoningEffort === right.reasoningEffort)
}
/**
* Register the durable model-selection projection when the registry is present.
* @param ctx - Session Controller context.
*/
export function installModelSelectionProjection(ctx: Context): void {
ctx.sessionProjections.register(modelSelectionProjection)
}
+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. */