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

refactor(client): isolate trajectory history reads

Browse Source
This commit is contained in:
_Kerman
2026-07-28 13:50:46 +08:00
parent 5fc9a041ba
commit 028ac5c2f6
18 changed files with 909 additions and 670 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
packages/client/runtime/src/client/index.ts
View File

@@ -28,16 +28,21 @@ export type {
EngineStoreHandle, EngineStoreInstance, ObservableSnapshot, SnapshotStore,
} from './contract/store.ts'
export type {
AssistantBlock, AssistantMessageNode, AssistantTiming, CodeSubCall, ComposerPhase, ContextMessageNode,
ConversationNode,
AssistantBlock, AssistantMessageNode, AssistantProvenanceView, AssistantRequestConfig,
AssistantTiming, CodeSubCall, ComposerPhase, ContextMessageNode, ConversationNode,
ConversationSnapshot, QueuedMessage, RunningToolCall,
SteeringMessageNode, TodoItem, ToolResultNode, UnknownSurfaceNode, UserMessageNode,
} from './sessions/conversation.ts'
export type {
AssistantProvenanceView, AssistantRequestConfig, CompactionRequestView,
ConversationContext, ConversationContextOriginKind, ConversationPromptChange,
ConversationPromptSnapshot, ModelRequestView,
} from './sessions/inspection.ts'
ConversationContext, ConversationContextOriginKind,
} from './sessions/conversation-context.ts'
export type {
ConversationPromptSnapshot, RequestInspectionSnapshot, RequestPromptChange, RequestView,
} from './sessions/request-inspection.ts'
export { inspectRequests } from './sessions/request-inspection.ts'
export { projectConversationHistory } from './sessions/fold-adapter.ts'
export type { ConversationHistoryProjection } from './sessions/fold-adapter.ts'
export type { SessionHistory, SessionHistorySnapshot } from './sessions/history.ts'
export { PendingWait } from './sessions/pending.ts'
export type { PendingInteraction, PendingKind, PendingPayloads } from './sessions/pending.ts'
export type { SessionId } from '@deepseek-ai/dsh-client-connection/client'

23
packages/client/runtime/src/client/sessions/conversation-context.ts Normal file
View File

@@ -0,0 +1,23 @@
import type { ConversationNode } from './conversation.ts'
import type { ConversationPromptSnapshot } from './request-inspection.ts'
/** Operation that started a new append-only model context. */
export type ConversationContextOriginKind = 'compaction' | 'rewind' | 'rewrite'
/** One immutable model-context generation reconstructed from surface replacements. */
export interface ConversationContext {
/** Zero-based generation within the session; stable across later appends. */
id: number
/** Previous generation in this session; absent for the initial context. */
parentId?: number
/** Why this generation exists; absent for the initial context. */
origin?: ConversationContextOriginKind
/** Event seq of the replacement that created this generation. */
originSeq?: number
/** Unix epoch ms of the replacement that created this generation. */
createdAt?: number
/** Latest request header observed in this generation, inherited until a later header replaces it. */
prompt?: ConversationPromptSnapshot
/** Final frozen nodes for historical generations, or current folded nodes for the tail. */
nodes: readonly ConversationNode[]
}

32
packages/client/runtime/src/client/sessions/conversation.ts
View File

@@ -9,12 +9,26 @@ import type {
RpcError, SessionId, ToolCallView, ToolResultView,
} from '@deepseek-ai/dsh-client-connection/client'
import type { PendingInteraction } from './pending.ts'
import type {
AssistantProvenanceView, AssistantRequestConfig, ConversationContext, SessionInspectionSnapshot,
} from './inspection.ts'
export type { TodoItem }
/** Request configuration recorded for one provider call. */
export interface AssistantRequestConfig {
provider: string
model: string
purpose?: string
thinking?: string
reasoningEffort?: string
temperature?: number
maxTokens?: number
stop?: readonly string[]
}
/** Stable provider/model identity reported for one completed request. */
export interface AssistantProvenanceView {
provider: string
model: string
}
/** Assistant content blocks sorted by what the UI cares about
* (text body / collapsible reasoning / tool-call card head / other fallback). */
export type AssistantBlock =
@@ -224,14 +238,6 @@ export interface ConversationSnapshot {
sessionId: SessionId
/** Surface fold product (finalized conversation nodes in surface order). */
nodes: readonly ConversationNode[]
/** Append-only context generations split at every model-surface replacement. */
contexts?: readonly ConversationContext[]
/** Auxiliary compaction requests, including those without an assistant/message surface node. */
compactionRequests?: SessionInspectionSnapshot['compactionRequests']
/** Ordinary provider requests, including failed attempts that produced no assistant message. */
requestAttempts?: SessionInspectionSnapshot['requestAttempts']
/** System-prompt/tool-catalog changes in request order. */
promptChanges?: SessionInspectionSnapshot['promptChanges']
/** Fold degradation flag (cross-window replace defense): when true, nodes come from the lenient linear scan. */
foldDegraded: boolean
partial: PartialAssistant | null
@@ -243,8 +249,6 @@ export interface ConversationSnapshot {
* unrelated snapshot swaps (memo premise, same regime as `nodes`).
*/
codeDispatches: ReadonlyMap<string, readonly CodeSubCall[]>
/** Model-visible tool schema captured for each recorded call id. */
callSchemas?: SessionInspectionSnapshot['callSchemas']
pending: readonly PendingInteraction[]
/** Read-only inbox mirror (session/queued frames + mux-open baseline; cleared by the leave-running flip). */
queue: readonly QueuedMessage[]

81
packages/client/runtime/src/client/sessions/fold-adapter.ts
View File

@@ -10,13 +10,45 @@ import type { SessionEvent } from '@deepseek-ai/dsh-session/types'
import {
SurfaceManager, isSurfaceEligibleType, isSurfaceEvent,
} from '@deepseek-ai/dsh-session/surface'
import type { ToolCallView, ToolEventView, ToolResultView } from '@deepseek-ai/dsh-client-connection/client'
import type { AssistantTiming, ConversationNode } from './conversation.ts'
import type {
HistoryEntry, ToolCallView, ToolEventView, ToolResultView,
} from '@deepseek-ai/dsh-client-connection/client'
import type {
AssistantRequestConfig, AssistantTiming, ConversationNode,
} from './conversation.ts'
import { toAssistantBlocks } from './conversation.ts'
import type {
AssistantRequestConfig, ConversationContext, ConversationContextOriginKind,
ConversationPromptSnapshot,
} from './inspection.ts'
ConversationContext, ConversationContextOriginKind,
} from './conversation-context.ts'
import type { ConversationPromptSnapshot } from './request-inspection.ts'
/** Lazy event-order and context-generation projection for history consumers. */
export interface ConversationHistoryProjection {
eventNodes: readonly ConversationNode[]
contexts: readonly ConversationContext[]
}
/**
* Project an immutable raw history window into event-order nodes and context
* generations. Session's chat snapshot never computes this projection.
* @param entries - Contiguous history entries in sequence order.
* @returns Inspection-oriented conversation projections.
*/
export function projectConversationHistory(
entries: readonly HistoryEntry[],
): ConversationHistoryProjection {
const adapter = new FoldAdapter(true)
const events = entries.map(entry => entry.event)
adapter.reset(
events,
events[0]?.seq ?? 0,
entries.map(entry => entry.view),
)
return {
eventNodes: adapter.eventNodes(),
contexts: adapter.contexts(),
}
}
/** In-window tool/call index entry (result-card backfill + runningCalls material). */
export interface CallIndexEntry {
@@ -155,6 +187,7 @@ export class FoldAdapter {
* reference-stability contract (§A.9.4) starts here. */
private rev = 0
private nodesResult: { rev: number; value: { nodes: ConversationNode[]; degraded: boolean } } | null = null
private eventNodesResult: { rev: number; value: readonly ConversationNode[] } | null = null
/** Revision of context structure or its request header; unrelated log-only events do not rebuild contexts. */
private contextRev = 0
private contextsResult: { rev: number; value: readonly ConversationContext[] } | null = null
@@ -162,6 +195,12 @@ export class FoldAdapter {
private activePrompt: ConversationPromptSnapshot | undefined
private promptsByContext = new Map<number, ConversationPromptSnapshot>()
/**
* @param projectContexts - Whether to maintain context-generation indexes
* for a later history projection. The live chat fold leaves this disabled.
*/
constructor(private readonly projectContexts = false) {}
/** In-window tool/call index (Session uses it for runningCalls and result-card backfill). */
get callIndex(): ReadonlyMap<string, CallIndexEntry> {
return this.callIdx
@@ -176,7 +215,7 @@ export class FoldAdapter {
*/
reset(events: readonly SessionEvent[], baseSeq: number, views?: readonly (ToolEventView | undefined)[]): void {
this.rev++
this.contextRev++
if (this.projectContexts) this.contextRev++
this.baseSeq = baseSeq
this.padded = []
for (let i = 0; i < baseSeq; i++) this.padded.push(paddingEvent(i))
@@ -194,7 +233,7 @@ export class FoldAdapter {
/* v8 ignore next -- dense-array guard: i stays within events.length, so the undefined arm needs a sparse array no caller builds. */
if (event !== undefined) {
this.indexCall(event, views?.[i])
this.indexContextPrompt(event)
if (this.projectContexts) this.indexContextPrompt(event)
}
}
}
@@ -207,10 +246,12 @@ export class FoldAdapter {
*/
append(event: SessionEvent, view?: ToolEventView): void {
this.rev++
if (isSurfaceEvent(event) || event.type === 'request/header') this.contextRev++
if (this.projectContexts && (isSurfaceEvent(event) || event.type === 'request/header')) {
this.contextRev++
}
this.padded.push(event)
this.indexCall(event, view)
this.indexContextPrompt(event)
if (this.projectContexts) this.indexContextPrompt(event)
}
/**
@@ -243,11 +284,33 @@ export class FoldAdapter {
return value
}
/**
* Every in-window message-producing event in original sequence order, without surface replacement folding.
* @returns append-only event projection for history inspection.
*/
eventNodes(): readonly ConversationNode[] {
if (this.eventNodesResult !== null && this.eventNodesResult.rev === this.rev) {
return this.eventNodesResult.value
}
const nodes: ConversationNode[] = []
for (let seq = this.baseSeq; seq < this.padded.length; seq++) {
const event = this.padded[seq]
if (event === undefined || !isSurfaceEligibleType(event.type)) continue
const node = this.materialize(seq)
if (node !== undefined) nodes.push(node)
}
this.eventNodesResult = { rev: this.rev, value: nodes }
return nodes
}
/**
* Append-only context generations reconstructed from canonical surface replacements.
* @returns Frozen historical contexts followed by the current context.
*/
contexts(): readonly ConversationContext[] {
if (!this.projectContexts) {
throw new Error('FoldAdapter context projection was not enabled')
}
if (this.contextsResult !== null && this.contextsResult.rev === this.contextRev) {
return this.contextsResult.value
}

33
packages/client/runtime/src/client/sessions/history.ts Normal file
View File

@@ -0,0 +1,33 @@
import type {
HistoryEntry, RpcError, SessionId,
} from '@deepseek-ai/dsh-client-connection/client'
import type { ObservableSnapshot } from '../contract/store.ts'
import type { OpenState } from './conversation.ts'
/**
* Immutable read window over one session's durable event log.
*
* The conversation snapshot is a chat projection. Consumers that need event
* order or request lifecycle data read this source instead of widening that
* projection with inspection-only fields.
*/
export interface SessionHistorySnapshot {
sessionId: SessionId
/** Contiguous raw log entries in ascending sequence order. */
entries: readonly HistoryEntry[]
/** Sequence of the first entry, or zero while the window is empty. */
baseSeq: number
openState: OpenState
openError: RpcError | null
hasMore: boolean
loadingOlder: boolean
}
/** Read-only observable history plus explicit full-ledger paging. */
export interface SessionHistory extends ObservableSnapshot<SessionHistorySnapshot> {
/**
* Load every earlier page currently available.
* @returns When paging is exhausted or cannot advance.
*/
loadAll(): Promise<void>
}

493
packages/client/runtime/src/client/sessions/inspection.ts
View File

@@ -1,493 +0,0 @@
// Session inspection read models. These projections preserve durable request
// and prompt semantics for diagnostic UIs without making the conversation fold
// or the core SessionSurface own inspection-only history.
import type { ContentBlock, ToolSchema } from '@deepseek-ai/dsh-llm/types'
import type { SessionEvent } from '@deepseek-ai/dsh-session/types'
import type { ConversationNode } from './conversation.ts'
/** Request configuration recorded in the effective header for one assistant response. */
export interface AssistantRequestConfig {
provider: string
model: string
purpose?: string
thinking?: string
reasoningEffort?: string
temperature?: number
maxTokens?: number
stop?: readonly string[]
}
/** Stable provider/model identity attached to one assistant response. */
export interface AssistantProvenanceView {
provider: string
model: string
}
/** Operation that started a new append-only model context. */
export type ConversationContextOriginKind = 'compaction' | 'rewind' | 'rewrite'
/** Latest complete model request header in force within one context generation. */
export interface ConversationPromptSnapshot {
/** Provider/model and sampling configuration from the latest effective request header. */
config?: AssistantRequestConfig
/** Rendered system prompt text; empty when the request had no system prompt. */
system: string
/** Complete tool catalog sent with the request, including tools that were never called. */
tools: readonly ToolSchema[]
}
/** One system-prompt/tool-catalog state that became effective in the request timeline. */
export interface ConversationPromptChange {
/** Sequence of the request/header event that introduced this state. */
seq: number
/** Unix epoch ms from the request/header event. */
time: number
/** How the model-visible system configuration differs from the prior recorded state. */
kind: 'initial' | 'system' | 'tools' | 'system-and-tools'
/** Complete state effective from this event onward. */
prompt: ConversationPromptSnapshot
/** State immediately before this change; absent for the initial header. */
previous?: ConversationPromptSnapshot
}
/** One immutable model-context generation reconstructed from surface replacements. */
export interface ConversationContext {
/** Zero-based generation within the session; stable across later appends. */
id: number
/** Previous generation in this session; absent for the initial context. */
parentId?: number
/** Why this generation exists; absent for the initial context. */
origin?: ConversationContextOriginKind
/** Event seq of the replacement that created this generation. */
originSeq?: number
/** Unix epoch ms of the replacement that created this generation. */
createdAt?: number
/** Latest request header observed in this generation, inherited until a later header replaces it. */
prompt?: ConversationPromptSnapshot
/** Final frozen nodes for historical generations, or current folded nodes for the tail. */
nodes: readonly ConversationNode[]
}
/** One auxiliary compaction model request reconstructed from its durable lifecycle events. */
export interface CompactionRequestView {
startSeq: number
turn: number
startedAt: number
completedAt: number | null
status: 'running' | 'complete' | 'error'
error?: string
summarySeq?: number
replacementSeq?: number
summary?: readonly ContentBlock[]
rawOutput?: readonly ContentBlock[]
provenance?: AssistantProvenanceView
requestConfig?: AssistantRequestConfig
usage?: unknown
}
/** One ordinary provider-request attempt reconstructed from a durable step boundary. */
export interface ModelRequestView {
/** Sequence of the step/start event that opened this attempt. */
startSeq: number
turn: number
step: number
/** Unix epoch ms from step/start. */
startedAt: number
/** Assistant completion time, or the failed step/end time when no response completed. */
completedAt: number | null
status: 'running' | 'complete' | 'error'
error?: string
/** Assistant/message sequence when this attempt completed successfully. */
resultSeq?: number
provenance?: AssistantProvenanceView
requestConfig?: AssistantRequestConfig
usage?: unknown
/** Retry ordinal scheduled after this failed attempt. */
retry?: number
maxRetries?: number
retryDelayMs?: number
}
/** Stable inspection substructures attached to a conversation snapshot. */
export interface SessionInspectionSnapshot {
compactionRequests: readonly CompactionRequestView[]
requestAttempts: readonly ModelRequestView[]
promptChanges: readonly ConversationPromptChange[]
callSchemas: ReadonlyMap<string, ToolSchema>
}
interface RetryEvent {
type: 'llm/retry'
seq: number
time: number
data: {
turn: number
step: number
retry: number
maxRetries: number
delayMs: number
failure: { message: string }
}
}
interface CompactionStartEvent {
type: 'compact/start'
seq: number
time: number
data: { turn: number }
}
interface CompactionSummaryEvent {
type: 'compact/summary'
seq: number
time: number
data: {
summary: readonly ContentBlock[]
rawOutput?: readonly ContentBlock[]
provider: string
model: string
maxTokens?: number
usage?: unknown
}
}
interface CompactionEndEvent {
type: 'compact/end'
seq: number
time: number
data: { turn: number; error?: string }
}
/**
* Own incremental invalidation and call-time schema capture for inspection
* read models. Conversation state delegates events here but owns no request
* reconstruction details.
*/
export class SessionInspection {
private activeToolSchemas = new Map<string, ToolSchema>()
private callSchemas = new Map<string, ToolSchema>()
private callSchemasRev = 0
private callSchemasCache: { rev: number; value: ReadonlyMap<string, ToolSchema> } | null = null
private modelRequestsRev = 0
private modelRequestsCache: {
rev: number
value: readonly ModelRequestView[]
} | null = null
private compactionRequestsRev = 0
private compactionRequestsCache: {
rev: number
value: readonly CompactionRequestView[]
} | null = null
private promptChangesRev = 0
private promptChangesCache: {
rev: number
value: readonly ConversationPromptChange[]
} | null = null
/**
* Invalidate projections before replaying a rebuilt history window.
* @returns Nothing.
*/
reset(): void {
this.activeToolSchemas = new Map()
this.callSchemas = new Map()
this.callSchemasRev++
this.modelRequestsRev++
this.compactionRequestsRev++
this.promptChangesRev++
}
/**
* Apply inspection-specific incremental state for one durable event.
* @param event - Event entering the current history window.
* @returns Nothing.
*/
applyEvent(event: SessionEvent): void {
if (affectsModelRequests(event)) this.modelRequestsRev++
if (affectsCompactionRequests(event)) this.compactionRequestsRev++
if (event.type === 'request/header') {
this.promptChangesRev++
this.activeToolSchemas = new Map(
(event.data.header.tools ?? []).map(schema => [schema.name, schema]),
)
return
}
if (event.type === 'tool/call') {
this.captureCallSchema(String(event.data.callId), event.data.name)
}
}
/**
* Preserve the schema active when one native or nested call starts.
* @param callId - Durable or synthetic call identifier.
* @param name - Tool name used to resolve the active catalog entry.
* @returns Nothing.
*/
captureCallSchema(callId: string, name: string): void {
if (this.callSchemas.has(callId)) return
const schema = this.activeToolSchemas.get(name)
if (schema === undefined) return
this.callSchemas.set(callId, schema)
this.callSchemasRev++
}
/**
* Materialize reference-stable inspection projections for the current log.
* @param events - Current contiguous client history window.
* @returns Inspection projections with stable unchanged substructure references.
*/
snapshot(events: readonly SessionEvent[]): SessionInspectionSnapshot {
if (this.callSchemasCache === null || this.callSchemasCache.rev !== this.callSchemasRev) {
this.callSchemasCache = { rev: this.callSchemasRev, value: new Map(this.callSchemas) }
}
if (
this.modelRequestsCache === null
|| this.modelRequestsCache.rev !== this.modelRequestsRev
) {
this.modelRequestsCache = {
rev: this.modelRequestsRev,
value: deriveModelRequests(events),
}
}
if (
this.compactionRequestsCache === null
|| this.compactionRequestsCache.rev !== this.compactionRequestsRev
) {
this.compactionRequestsCache = {
rev: this.compactionRequestsRev,
value: deriveCompactionRequests(events),
}
}
if (
this.promptChangesCache === null
|| this.promptChangesCache.rev !== this.promptChangesRev
) {
this.promptChangesCache = {
rev: this.promptChangesRev,
value: derivePromptChanges(events),
}
}
return {
callSchemas: this.callSchemasCache.value,
requestAttempts: this.modelRequestsCache.value,
compactionRequests: this.compactionRequestsCache.value,
promptChanges: this.promptChangesCache.value,
}
}
}
function modelRequestKey(turn: number, step: number): string {
return `${turn}\u0000${step}`
}
function affectsModelRequests(event: SessionEvent): boolean {
switch (event.type) {
case 'request/header':
case 'step/start':
case 'assistant/message':
case 'step/end':
return true
case 'turn/end':
return event.data.reason.kind === 'error'
default:
return (event.type as string) === 'llm/retry'
}
}
function affectsCompactionRequests(event: SessionEvent): boolean {
const type = event.type as string
return type === 'compact/start'
|| type === 'compact/summary'
|| type === 'compact/end'
|| (event.type === 'user/message' && isCompactionSource(event.data.source))
}
/** Project every durable step into one provider request, retaining failed retry attempts. */
function deriveModelRequests(events: readonly SessionEvent[]): readonly ModelRequestView[] {
const requests: ModelRequestView[] = []
const byStep = new Map<string, number>()
let activeStep: string | undefined
let activeConfig: ConversationPromptSnapshot['config']
const update = (key: string, change: Partial<ModelRequestView>): void => {
const index = byStep.get(key)
if (index === undefined) return
const request = requests[index]
if (request !== undefined) requests[index] = { ...request, ...change }
}
for (const sourceEvent of events) {
if (sourceEvent.type === 'request/header') {
activeConfig = sourceEvent.data.header.config
if (activeStep !== undefined) update(activeStep, { requestConfig: activeConfig })
continue
}
if (sourceEvent.type === 'step/start') {
const { turn, step } = sourceEvent.data
const key = modelRequestKey(turn, step)
byStep.set(key, requests.length)
requests.push({
startSeq: sourceEvent.seq,
turn,
step,
startedAt: sourceEvent.time,
completedAt: null,
status: 'running',
...(activeConfig === undefined ? {} : { requestConfig: activeConfig }),
})
activeStep = key
continue
}
if (sourceEvent.type === 'assistant/message') {
const key = modelRequestKey(sourceEvent.data.turn, sourceEvent.data.step)
update(key, {
completedAt: sourceEvent.time,
status: 'complete',
resultSeq: sourceEvent.seq,
provenance: {
provider: sourceEvent.data.provenance.provider,
model: sourceEvent.data.provenance.model,
},
...(sourceEvent.data.usage === undefined ? {} : { usage: sourceEvent.data.usage }),
})
continue
}
if (sourceEvent.type === 'step/end') {
const key = modelRequestKey(sourceEvent.data.turn, sourceEvent.data.step)
const index = byStep.get(key)
const request = index === undefined ? undefined : requests[index]
if (index !== undefined && request !== undefined && request.status === 'running') {
requests[index] = {
...request,
completedAt: sourceEvent.time,
status: 'error',
}
}
if (activeStep === key) activeStep = undefined
continue
}
if ((sourceEvent.type as string) === 'llm/retry') {
const event = sourceEvent as unknown as RetryEvent
update(modelRequestKey(event.data.turn, event.data.step), {
status: 'error',
error: event.data.failure.message,
retry: event.data.retry,
maxRetries: event.data.maxRetries,
retryDelayMs: event.data.delayMs,
})
continue
}
if (sourceEvent.type !== 'turn/end' || sourceEvent.data.reason.kind !== 'error') continue
const reason = sourceEvent.data.reason
update(modelRequestKey(sourceEvent.data.turn, reason.step), {
status: 'error',
error: 'failure' in reason ? reason.failure.message : reason.message,
})
}
return requests
}
/** Project log-only compaction request brackets without coupling the client runtime to one backend package. */
function deriveCompactionRequests(events: readonly SessionEvent[]): readonly CompactionRequestView[] {
const requests: CompactionRequestView[] = []
let active: CompactionRequestView | undefined
for (const sourceEvent of events) {
const type = sourceEvent.type as string
if (type === 'compact/start') {
const event = sourceEvent as unknown as CompactionStartEvent
active = {
startSeq: event.seq,
turn: event.data.turn,
startedAt: event.time,
completedAt: null,
status: 'running',
}
continue
}
if (type === 'compact/summary' && active !== undefined) {
const event = sourceEvent as unknown as CompactionSummaryEvent
active = {
...active,
summarySeq: event.seq,
summary: event.data.summary,
...(event.data.rawOutput === undefined ? {} : { rawOutput: event.data.rawOutput }),
provenance: {
provider: event.data.provider,
model: event.data.model,
},
requestConfig: {
provider: event.data.provider,
model: event.data.model,
purpose: 'compaction',
...(event.data.maxTokens === undefined ? {} : { maxTokens: event.data.maxTokens }),
},
...(event.data.usage === undefined ? {} : { usage: event.data.usage }),
}
continue
}
if (
sourceEvent.type === 'user/message'
&& active?.summarySeq !== undefined
&& isCompactionSource(sourceEvent.data.source)
) {
active = { ...active, replacementSeq: sourceEvent.seq }
continue
}
if (type !== 'compact/end' || active === undefined) continue
const event = sourceEvent as unknown as CompactionEndEvent
active = {
...active,
completedAt: event.time,
status: event.data.error === undefined ? 'complete' : 'error',
...(event.data.error === undefined ? {} : { error: event.data.error }),
}
requests.push(active)
active = undefined
}
if (active !== undefined) requests.push(active)
return requests
}
/** Project request headers into model-visible system/tool changes only. */
function derivePromptChanges(events: readonly SessionEvent[]): readonly ConversationPromptChange[] {
const changes: ConversationPromptChange[] = []
let previous: ConversationPromptSnapshot | undefined
for (const event of events) {
if (event.type !== 'request/header') continue
const prompt: ConversationPromptSnapshot = {
config: event.data.header.config,
system: event.data.header.system ?? '',
tools: event.data.header.tools ?? [],
}
const systemChanged = previous !== undefined && previous.system !== prompt.system
const toolsChanged = previous !== undefined
&& JSON.stringify(previous.tools) !== JSON.stringify(prompt.tools)
if (previous === undefined || systemChanged || toolsChanged) {
changes.push({
seq: event.seq,
time: event.time,
kind: previous === undefined
? 'initial'
: systemChanged && toolsChanged
? 'system-and-tools'
: systemChanged
? 'system'
: 'tools',
prompt,
...(previous === undefined ? {} : { previous }),
})
}
previous = prompt
}
return changes
}
function isCompactionSource(source: unknown): boolean {
return typeof source === 'object'
&& source !== null
&& 'kind' in source
&& source.kind === 'plugin'
&& 'plugin' in source
&& source.plugin === 'compact'
}

352
packages/client/runtime/src/client/sessions/request-inspection.ts Normal file
View File

@@ -0,0 +1,352 @@
// Request-centric inspection read model. Ordinary generation and compaction
// calls share one chronological projection; presentation-specific grouping
// remains in the trajectory consumer.
import type { ContentBlock, ToolSchema } from '@deepseek-ai/dsh-llm/types'
import type { HistoryEntry } from '@deepseek-ai/dsh-client-connection/client'
import type { SessionEvent } from '@deepseek-ai/dsh-session/types'
import type {
AssistantProvenanceView, AssistantRequestConfig,
} from './conversation.ts'
export type {
AssistantProvenanceView, AssistantRequestConfig,
} from './conversation.ts'
/** Complete model-visible request header in force for an ordinary generation. */
export interface ConversationPromptSnapshot {
/** Provider/model and sampling configuration from the effective request header. */
config: AssistantRequestConfig
/** Rendered system prompt text; empty when the request had no system prompt. */
system: string
/** Complete tool catalog sent with the request, including tools that were never called. */
tools: readonly ToolSchema[]
}
/** System/tool change introduced while preparing one ordinary request. */
export interface RequestPromptChange {
/** Sequence of the request/header event that introduced this state. */
seq: number
/** Unix epoch ms from the request/header event. */
time: number
/** How the model-visible prompt differs from the previous recorded state. */
kind: 'initial' | 'system' | 'tools' | 'system-and-tools'
/** State immediately before this change; absent for the initial header. */
previous?: ConversationPromptSnapshot
}
/** One provider request reconstructed from durable request lifecycle events. */
export interface RequestView {
/** Request category; compaction is a purpose, not a separate projection. */
purpose: 'assistant' | 'compaction'
/** Sequence that opened the operation represented by this request. */
startSeq: number
turn: number
/** Agent-loop step, or zero for a direct compaction request. */
step: number
startedAt: number
completedAt: number | null
status: 'running' | 'complete' | 'error'
error?: string
/** Effective ordinary request input, inherited until a later header changes it. */
prompt?: ConversationPromptSnapshot
/** Prompt change logged while preparing this request. */
promptChange?: RequestPromptChange
provenance?: AssistantProvenanceView
requestConfig?: AssistantRequestConfig
usage?: unknown
/** Assistant message or compaction summary sequence produced by this request. */
resultSeq?: number
/** Compaction replacement message sequence, when one was committed. */
replacementSeq?: number
/** Safe compaction summary projection. */
summary?: readonly ContentBlock[]
/** Complete compaction provider output before the safe projection. */
rawOutput?: readonly ContentBlock[]
/** Retry ordinal scheduled after a failed ordinary request. */
retry?: number
maxRetries?: number
retryDelayMs?: number
}
/** Immutable request-centric projection derived from one history window. */
export interface RequestInspectionSnapshot {
requests: readonly RequestView[]
callSchemas: ReadonlyMap<string, ToolSchema>
}
/**
* Derive the request-centric read model from one immutable history window.
* Compaction participates as a request purpose rather than a parallel
* top-level collection.
* @param entries - Contiguous raw session history.
* @returns Requests and call-time schemas derived from that history.
*/
export function inspectRequests(
entries: readonly HistoryEntry[],
): RequestInspectionSnapshot {
const events = entries.map(entry => entry.event)
return {
requests: deriveRequests(events),
callSchemas: deriveCallSchemas(events),
}
}
interface RetryEvent {
type: 'llm/retry'
seq: number
time: number
data: {
turn: number
step: number
retry: number
maxRetries: number
delayMs: number
failure: { message: string }
}
}
interface CompactionStartEvent {
type: 'compact/start'
seq: number
time: number
data: { turn: number }
}
interface CompactionSummaryEvent {
type: 'compact/summary'
seq: number
time: number
data: {
summary: readonly ContentBlock[]
rawOutput?: readonly ContentBlock[]
provider: string
model: string
maxTokens?: number
usage?: unknown
}
}
interface CompactionEndEvent {
type: 'compact/end'
seq: number
time: number
data: { turn: number; error?: string }
}
function requestKey(turn: number, step: number): string {
return `${turn}\u0000${step}`
}
function deriveCallSchemas(
events: readonly SessionEvent[],
): ReadonlyMap<string, ToolSchema> {
let active = new Map<string, ToolSchema>()
const calls = new Map<string, ToolSchema>()
const capture = (callId: string, name: string): void => {
if (calls.has(callId)) return
const schema = active.get(name)
if (schema !== undefined) calls.set(callId, schema)
}
for (const event of events) {
if (event.type === 'request/header') {
active = new Map(
(event.data.header.tools ?? []).map(schema => [schema.name, schema]),
)
continue
}
if (event.type === 'tool/call') {
capture(String(event.data.callId), event.data.name)
continue
}
const type = event.type as string
if (type === 'tool/code-dispatch-start' || type === 'tool/code-dispatch') {
const data = event.data as unknown as { subCallId: string; name: string }
capture(data.subCallId, data.name)
}
}
return calls
}
function promptChange(
previous: ConversationPromptSnapshot | undefined,
prompt: ConversationPromptSnapshot,
event: SessionEvent<'request/header'>,
): RequestPromptChange | undefined {
const systemChanged = previous !== undefined && previous.system !== prompt.system
const toolsChanged = previous !== undefined
&& JSON.stringify(previous.tools) !== JSON.stringify(prompt.tools)
if (previous !== undefined && !systemChanged && !toolsChanged) return
return {
seq: event.seq,
time: event.time,
kind: previous === undefined
? 'initial'
: systemChanged && toolsChanged
? 'system-and-tools'
: systemChanged
? 'system'
: 'tools',
...(previous === undefined ? {} : { previous }),
}
}
/** Project ordinary and compaction provider calls into one chronological request stream. */
function deriveRequests(events: readonly SessionEvent[]): readonly RequestView[] {
const requests: RequestView[] = []
const ordinaryByStep = new Map<string, number>()
let activeStep: string | undefined
let activePrompt: ConversationPromptSnapshot | undefined
let activeCompaction: number | undefined
const update = (index: number | undefined, change: Partial<RequestView>): void => {
if (index === undefined) return
const request = requests[index]
if (request !== undefined) requests[index] = { ...request, ...change }
}
for (const sourceEvent of events) {
if (sourceEvent.type === 'step/start') {
const { turn, step } = sourceEvent.data
const key = requestKey(turn, step)
ordinaryByStep.set(key, requests.length)
requests.push({
purpose: 'assistant',
startSeq: sourceEvent.seq,
turn,
step,
startedAt: sourceEvent.time,
completedAt: null,
status: 'running',
...(activePrompt === undefined
? {}
: { prompt: activePrompt, requestConfig: activePrompt.config }),
})
activeStep = key
continue
}
if (sourceEvent.type === 'request/header') {
const prompt: ConversationPromptSnapshot = {
config: sourceEvent.data.header.config,
system: sourceEvent.data.header.system ?? '',
tools: sourceEvent.data.header.tools ?? [],
}
const change = promptChange(activePrompt, prompt, sourceEvent)
activePrompt = prompt
update(activeStep === undefined ? undefined : ordinaryByStep.get(activeStep), {
prompt,
requestConfig: prompt.config,
...(change === undefined ? {} : { promptChange: change }),
})
continue
}
if (sourceEvent.type === 'assistant/message') {
update(ordinaryByStep.get(requestKey(sourceEvent.data.turn, sourceEvent.data.step)), {
completedAt: sourceEvent.time,
status: 'complete',
resultSeq: sourceEvent.seq,
provenance: {
provider: sourceEvent.data.provenance.provider,
model: sourceEvent.data.provenance.model,
},
...(sourceEvent.data.usage === undefined ? {} : { usage: sourceEvent.data.usage }),
})
continue
}
if (sourceEvent.type === 'step/end') {
const key = requestKey(sourceEvent.data.turn, sourceEvent.data.step)
const index = ordinaryByStep.get(key)
const request = index === undefined ? undefined : requests[index]
if (request?.status === 'running') {
update(index, {
completedAt: sourceEvent.time,
status: 'error',
})
}
if (activeStep === key) activeStep = undefined
continue
}
if ((sourceEvent.type as string) === 'llm/retry') {
const event = sourceEvent as unknown as RetryEvent
update(ordinaryByStep.get(requestKey(event.data.turn, event.data.step)), {
status: 'error',
error: event.data.failure.message,
retry: event.data.retry,
maxRetries: event.data.maxRetries,
retryDelayMs: event.data.delayMs,
})
continue
}
if (sourceEvent.type === 'turn/end' && sourceEvent.data.reason.kind === 'error') {
const reason = sourceEvent.data.reason
update(ordinaryByStep.get(requestKey(sourceEvent.data.turn, reason.step)), {
status: 'error',
error: 'failure' in reason ? reason.failure.message : reason.message,
})
continue
}
const type = sourceEvent.type as string
if (type === 'compact/start') {
const event = sourceEvent as unknown as CompactionStartEvent
activeCompaction = requests.length
requests.push({
purpose: 'compaction',
startSeq: event.seq,
turn: event.data.turn,
step: 0,
startedAt: event.time,
completedAt: null,
status: 'running',
})
continue
}
if (type === 'compact/summary' && activeCompaction !== undefined) {
const event = sourceEvent as unknown as CompactionSummaryEvent
update(activeCompaction, {
resultSeq: event.seq,
summary: event.data.summary,
...(event.data.rawOutput === undefined ? {} : { rawOutput: event.data.rawOutput }),
provenance: {
provider: event.data.provider,
model: event.data.model,
},
requestConfig: {
provider: event.data.provider,
model: event.data.model,
purpose: 'compaction',
...(event.data.maxTokens === undefined ? {} : { maxTokens: event.data.maxTokens }),
},
...(event.data.usage === undefined ? {} : { usage: event.data.usage }),
})
continue
}
if (
sourceEvent.type === 'user/message'
&& activeCompaction !== undefined
&& isCompactionSource(sourceEvent.data.source)
) {
update(activeCompaction, { replacementSeq: sourceEvent.seq })
continue
}
if (type !== 'compact/end' || activeCompaction === undefined) continue
const event = sourceEvent as unknown as CompactionEndEvent
update(activeCompaction, {
completedAt: event.time,
status: event.data.error === undefined ? 'complete' : 'error',
...(event.data.error === undefined ? {} : { error: event.data.error }),
})
activeCompaction = undefined
}
return requests.sort((left, right) => left.startSeq - right.startSeq)
}
function isCompactionSource(source: unknown): boolean {
return typeof source === 'object'
&& source !== null
&& 'kind' in source
&& source.kind === 'plugin'
&& 'plugin' in source
&& source.plugin === 'compact'
}

92
packages/client/runtime/src/client/sessions/session.ts
View File

@@ -15,7 +15,7 @@ import type {
CodeSubCall, ComposerPhase, ConversationNode, ConversationSnapshot,
OpenState, PromptError, QueuedMessage, RunningToolCall,
} from './conversation.ts'
import { SessionInspection } from './inspection.ts'
import type { SessionHistory, SessionHistorySnapshot } from './history.ts'
import type { PendingInteraction } from './pending.ts'
import { PendingWait } from './pending.ts'
import { FoldAdapter } from './fold-adapter.ts'
@@ -59,8 +59,9 @@ function queuePreviewOf(content: readonly ContentBlock[]): string {
}
/**
* Owns a session's event window, derived conversation state, and observable
* snapshot. React bindings remain outside this data layer.
* Owns a session's event window and exposes two observable read surfaces:
* the folded chat conversation and the raw history window. React bindings
* remain outside this data layer.
*/
export class Session implements ObservableSnapshot<ConversationSnapshot> {
// ---- Window and derived state (all private; the snapshot is the only read surface) ----
@@ -108,8 +109,9 @@ export class Session implements ObservableSnapshot<ConversationSnapshot> {
private codeDispatches = new Map<string, readonly CodeSubCall[]>()
private dispatchesRev = 0
private dispatchesCache: { rev: number; value: ReadonlyMap<string, readonly CodeSubCall[]> } | null = null
/** Diagnostic projections isolated from the ordinary conversation state machine. */
private readonly inspection = new SessionInspection()
/** Raw history revision; published entries are copied so later live appends never mutate a prior snapshot. */
private historyRev = 0
private historyEntriesCache: { rev: number; value: readonly HistoryEntry[] } | null = null
private running = false
/**
* Sticky send marker, private input of the composerPhase derivation: set
@@ -130,9 +132,13 @@ export class Session implements ObservableSnapshot<ConversationSnapshot> {
private subscribedLastSeq: number | null = null
private snapshotCache: ConversationSnapshot
private historySnapshotCache: SessionHistorySnapshot | undefined
private readonly notifier = new Notifier(() => {
this.snapshotCache = this.buildSnapshot()
this.historySnapshotCache = this.buildHistorySnapshot()
})
/** Raw log read surface; trajectory-like consumers project their own model from it. */
readonly history: SessionHistory
/**
* Agent-scoped cordis context, bound once by SessionsService when it
* mints the scope (the client mirror of the host Agent's loopCtx). The
@@ -153,6 +159,19 @@ export class Session implements ObservableSnapshot<ConversationSnapshot> {
private readonly options: SessionOptions = {},
) {
this.snapshotCache = this.buildSnapshot()
this.historySnapshotCache = this.buildHistorySnapshot()
this.history = {
getSnapshot: () => {
this.notifier.ensureFresh()
/* v8 ignore next -- constructor initializes the cache before history is published. */
if (this.historySnapshotCache === undefined) {
throw new Error(`session ${this.sessionId} history cache is uninitialized`)
}
return this.historySnapshotCache
},
subscribe: listener => this.notifier.subscribe(listener),
loadAll: () => this.loadAllHistory(),
}
}
/**
@@ -268,6 +287,7 @@ export class Session implements ObservableSnapshot<ConversationSnapshot> {
}
this.events = [...older.map(e => e.event), ...this.events]
this.views = [...older.map(e => e.view), ...this.views]
this.historyRev++
/* v8 ignore next -- the ?? arm needs older[0] undefined, but the empty-page branch above already returned. */
this.baseSeq = older[0]?.event.seq ?? this.baseSeq
this.hasMore = result.value.hasMore
@@ -281,6 +301,20 @@ export class Session implements ObservableSnapshot<ConversationSnapshot> {
}
}
/**
* Exhaust history paging for inspection surfaces that require a complete
* session ledger. Stops after a failed or non-advancing page so a transient
* backend failure cannot become an automatic retry loop.
* @returns When the available history has been exhausted or paging stops making progress.
*/
private async loadAllHistory(): Promise<void> {
while (this.openState === 'open' && this.hasMore && !this.loadingOlder) {
const previousBaseSeq = this.baseSeq
await this.loadOlder()
if (this.baseSeq === previousBaseSeq) return
}
}
/** Reconnect rebuild (manager calls this on onConnected for instances that were opened):
* reset the window and rerun open; pending waits for the baseline replay. Invalidates any
* in-flight open first — its history request rode the dead connection and must not settle
@@ -298,6 +332,7 @@ export class Session implements ObservableSnapshot<ConversationSnapshot> {
this.openError = null
this.events = []
this.views = []
this.historyRev++
this.baseSeq = 0
// Superseded, not settled: the baseline replay re-sends still-pending requested frames verbatim
// (same rpcId), re-minting fresh waits; a stale reference's respond() still reaches the host.
@@ -512,6 +547,7 @@ export class Session implements ObservableSnapshot<ConversationSnapshot> {
private installWindow(entries: HistoryEntry[], hasMore: boolean, todos: readonly TodoItem[] | undefined): void {
this.events = entries.map(e => e.event)
this.views = entries.map(e => e.view)
this.historyRev++
this.baseSeq = this.events[0]?.seq ?? 0
this.hasMore = hasMore
// Session-level projection from the tail page (full-log latest todo/write,
@@ -536,6 +572,7 @@ export class Session implements ObservableSnapshot<ConversationSnapshot> {
if (tailSeq !== null && event.seq <= tailSeq) return // replay overlap, drop
this.events.push(event)
this.views.push(view)
this.historyRev++
this.foldAdapter.append(event, view)
this.applyEventSideEffects(event, view)
}
@@ -605,7 +642,6 @@ export class Session implements ObservableSnapshot<ConversationSnapshot> {
/** Per-event side effects (right column of the §A.9 dispatch table):
* chunk accumulation / partial clear on finalize / openCalls add-remove. */
private applyEventSideEffects(event: SessionEvent, view?: ToolEventView): void {
this.inspection.applyEvent(event)
// The `tool/code-dispatch-start`/`tool/code-dispatch` pair is declared by
// the host-side dsh-tools plugin whose types cannot enter the client
// program (its host Context merges collide with the client's), so this
@@ -626,7 +662,6 @@ export class Session implements ObservableSnapshot<ConversationSnapshot> {
argsRaw: JSON.stringify(data.arguments),
turn: 0, step: 0, time: event.time, callView: null,
}
this.inspection.captureCallSchema(data.subCallId, data.name)
const siblings = this.codeDispatches.get(data.parentCallId) ?? []
this.codeDispatches.set(data.parentCallId, [...siblings, running])
this.dispatchesRev++
@@ -646,7 +681,6 @@ export class Session implements ObservableSnapshot<ConversationSnapshot> {
content: ContentBlock[]
}
const siblings = this.codeDispatches.get(data.parentCallId) ?? []
this.inspection.captureCallSchema(data.subCallId, data.name)
const at = siblings.findIndex(sub => sub.callId === data.subCallId)
const started = at === -1 ? undefined : siblings[at]
const settled: CodeSubCall = {
@@ -755,7 +789,6 @@ export class Session implements ObservableSnapshot<ConversationSnapshot> {
this.frozenRev++
this.codeDispatches = new Map()
this.dispatchesRev++
this.inspection.reset()
for (let i = 0; i < this.events.length; i++) {
const event = this.events[i]
/* v8 ignore next -- dense-array guard: i stays within events.length, so the undefined arm needs a sparse array no caller builds. */
@@ -770,7 +803,6 @@ export class Session implements ObservableSnapshot<ConversationSnapshot> {
private buildSnapshot(): ConversationSnapshot {
const { nodes: folded, degraded } = this.foldAdapter.nodes()
const contexts = this.foldAdapter.contexts()
// Frozen interrupted nodes ride fractional seqs: a stable merge keeps them in flow order.
// The merged array is cached on (folded reference, frozenRev) so an unchanged flow keeps its
// reference across snapshot swaps (§A.9.4).
@@ -795,21 +827,15 @@ export class Session implements ObservableSnapshot<ConversationSnapshot> {
if (this.queueCache === null || this.queueCache.rev !== this.queueRev) {
this.queueCache = { rev: this.queueRev, value: this.queued.map(entry => entry.row) }
}
const inspection = this.inspection.snapshot(this.events)
const partial = this.partial?.toPartial() ?? null
return {
sessionId: this.sessionId,
nodes,
contexts,
compactionRequests: inspection.compactionRequests,
requestAttempts: inspection.requestAttempts,
promptChanges: inspection.promptChanges,
foldDegraded: degraded,
partial,
runningCalls: this.callsCache.value,
pending: this.pendingCache.value,
codeDispatches: this.dispatchesCache.value,
callSchemas: inspection.callSchemas,
queue: this.queueCache.value,
running: this.running,
composerPhase: derivePhase(
@@ -827,6 +853,40 @@ export class Session implements ObservableSnapshot<ConversationSnapshot> {
todos: this.todos,
}
}
/** Build the raw history read surface without leaking the mutable window arrays. */
private buildHistorySnapshot(): SessionHistorySnapshot {
if (this.historyEntriesCache === null || this.historyEntriesCache.rev !== this.historyRev) {
this.historyEntriesCache = {
rev: this.historyRev,
value: this.events.map((event, index) => {
const view = this.views[index]
return view === undefined ? { event } : { event, view }
}),
}
}
const previous = this.historySnapshotCache
if (
previous !== undefined
&& previous.entries === this.historyEntriesCache.value
&& previous.baseSeq === this.baseSeq
&& previous.openState === this.openState
&& previous.openError === this.openError
&& previous.hasMore === this.hasMore
&& previous.loadingOlder === this.loadingOlder
) {
return previous
}
return {
sessionId: this.sessionId,
entries: this.historyEntriesCache.value,
baseSeq: this.baseSeq,
openState: this.openState,
openError: this.openError,
hasMore: this.hasMore,
loadingOlder: this.loadingOlder,
}
}
}
/**

11
packages/client/runtime/tests/fold-adapter.spec.ts
View File

@@ -6,7 +6,9 @@
import { describe, expect, it, vi } from 'vitest'
import type { SessionEvent } from '@deepseek-ai/dsh-session/types'
import { FoldAdapter } from '../src/client/sessions/fold-adapter.ts'
import {
FoldAdapter, projectConversationHistory,
} from '../src/client/sessions/fold-adapter.ts'
import { ev, plainTurn } from './event-script.ts'
const at = (seq: number, e: Record<string, unknown>): SessionEvent =>
@@ -35,8 +37,7 @@ describe('FoldAdapter', () => {
})
it('projects frozen surface generations without widening the core live surface', () => {
const adapter = new FoldAdapter()
adapter.reset([
const events = [
ev.user(0, 'a'),
ev.user(1, 'b'),
at(2, {
@@ -61,9 +62,9 @@ describe('FoldAdapter', () => {
provenance: { provider: 'fake', model: 'fake' },
},
}),
], 0)
]
expect(adapter.contexts().map(context => ({
expect(projectConversationHistory(events.map(event => ({ event }))).contexts.map(context => ({
id: context.id,
parentId: context.parentId,
originSeq: context.originSeq,

108
packages/client/runtime/tests/request-inspection.spec.ts Normal file
View File

@@ -0,0 +1,108 @@
import { describe, expect, it } from 'vitest'
import type { HistoryEntry } from '@deepseek-ai/dsh-client-connection/client'
import type { SessionEvent } from '@deepseek-ai/dsh-session/types'
import { inspectRequests } from '../src/client/sessions/request-inspection.ts'
const at = (seq: number, type: string, data: unknown): SessionEvent =>
({ seq, time: 1_700_000_000_000 + seq, type, data }) as SessionEvent
const entriesOf = (events: readonly SessionEvent[]): HistoryEntry[] =>
events.map(event => ({ event }))
describe('inspectRequests', () => {
it('projects ordinary and compaction calls into one chronological request stream', () => {
const events = [
at(0, 'step/start', { turn: 1, step: 1 }),
at(1, 'request/header', {
reason: 'initial',
header: {
config: { provider: 'fake', model: 'model' },
system: 'system',
tools: [{
name: 'read',
description: 'Read a file.',
parameters: { type: 'object' },
}],
},
}),
at(2, 'tool/call', {
turn: 1,
step: 1,
callId: 'call-1',
name: 'read',
arguments: '{}',
}),
at(3, 'assistant/message', {
turn: 1,
step: 1,
content: [{ type: 'text', text: 'done' }],
provenance: { provider: 'fake', model: 'model' },
usage: { inputTokens: 5, outputTokens: 2 },
}),
at(4, 'step/end', { turn: 1, step: 1 }),
at(5, 'compact/start', { turn: 1 }),
at(6, 'compact/summary', {
summary: [{ type: 'text', text: 'summary' }],
rawOutput: [
{ type: 'reasoning', text: 'thought' },
{ type: 'text', text: 'summary' },
],
provider: 'fake',
model: 'compact-model',
usage: { inputTokens: 8, outputTokens: 3 },
}),
at(7, 'user/message', {
content: [{ type: 'text', text: 'checkpoint' }],
source: { kind: 'plugin', plugin: 'compact' },
}),
at(8, 'compact/end', { turn: 1 }),
]
const snapshot = inspectRequests(entriesOf(events))
expect(snapshot.requests).toMatchObject([
{
purpose: 'assistant',
startSeq: 0,
resultSeq: 3,
status: 'complete',
prompt: {
config: { provider: 'fake', model: 'model' },
system: 'system',
},
promptChange: { seq: 1, kind: 'initial' },
},
{
purpose: 'compaction',
startSeq: 5,
resultSeq: 6,
replacementSeq: 7,
status: 'complete',
summary: [{ type: 'text', text: 'summary' }],
},
])
expect(snapshot.callSchemas.get('call-1')?.name).toBe('read')
})
it('captures schemas for nested tool dispatches from the active request header', () => {
const snapshot = inspectRequests(entriesOf([
at(0, 'request/header', {
reason: 'initial',
header: {
config: { provider: 'fake', model: 'model' },
tools: [{
name: 'read',
description: 'Read a file.',
parameters: { type: 'object' },
}],
},
}),
at(1, 'tool/code-dispatch-start', {
parentCallId: 'parent',
subCallId: 'nested',
name: 'read',
arguments: {},
}),
]))
expect(snapshot.callSchemas.get('nested')?.name).toBe('read')
})
})

42
packages/client/runtime/tests/session.spec.ts
View File

@@ -256,6 +256,42 @@ describe('paging', () => {
expect(snapshot.nodes.map(n => n.seq)).toEqual([1, 3, 7, 9])
})
it('loads every older page for complete-history inspection', async () => {
const pages = [
plainTurn(0, 0, '最早问', '最早答'),
plainTurn(6, 1, '中间问', '中间答'),
plainTurn(12, 2, '最新问', '最新答'),
]
const { api, session } = makeSession()
api.onHistory = (payload) => {
if (payload.beforeSeq === undefined) return histResponse(pages[2]!, true)
if (payload.beforeSeq === 12) return histResponse(pages[1]!, true)
return histResponse(pages[0]!, false)
}
await session.open()
await session.history.loadAll()
expect(api.callsOf('session.history')).toHaveLength(3)
expect(session.history.getSnapshot().hasMore).toBe(false)
expect(session.history.getSnapshot().entries.map(entry => entry.event.seq))
.toEqual([...Array(18).keys()])
expect(session.getSnapshot().nodes.map(node => node.seq)).toEqual([1, 3, 7, 9, 13, 15])
})
it('stops complete-history loading when a page makes no progress', async () => {
const { api, session } = makeSession()
api.onHistory = payload => payload.beforeSeq === undefined
? histResponse(plainTurn(6, 1, '新问', '新答'), true)
: Promise.resolve(err({ code: 'internal', message: 'page unavailable', details: {} }))
await session.open()
await session.history.loadAll()
expect(api.callsOf('session.history')).toHaveLength(2)
expect(session.history.getSnapshot().hasMore).toBe(true)
})
it('drops a discontinuous older page fail-soft (window unchanged, hasMore cleared)', async () => {
const { api, session } = makeSession()
api.onHistory = payload => payload.beforeSeq === undefined
@@ -855,16 +891,12 @@ describe('reference stability (the memo contract)', () => {
expect(after).not.toBe(before)
expect(after.runningCalls).toBe(before.runningCalls)
expect(after.pending).toBe(before.pending)
expect(after.requestAttempts).toBe(before.requestAttempts)
expect(after.compactionRequests).toBe(before.compactionRequests)
// And a mutation on the tracked domain swaps that array.
feed(ev.toolResult(11, 1, 'c1', 'ECHO'))
const resolved = session.getSnapshot()
expect(resolved.runningCalls).not.toBe(after.runningCalls)
expect(resolved.pending).toBe(after.pending)
feed(ev.assistant(12, 1, '完成'))
const completed = session.getSnapshot()
expect(completed.requestAttempts).not.toBe(resolved.requestAttempts)
expect(completed.compactionRequests).toBe(resolved.compactionRequests)
expect(session.getSnapshot()).not.toBe(resolved)
})
})