fix(session): bound trajectory history projections
This commit is contained in:
@@ -114,6 +114,16 @@ export class Session implements ObservableSnapshot<ConversationSnapshot> {
|
|||||||
private callSchemas = new Map<string, ToolSchema>()
|
private callSchemas = new Map<string, ToolSchema>()
|
||||||
private callSchemasRev = 0
|
private callSchemasRev = 0
|
||||||
private callSchemasCache: { rev: number; value: ReadonlyMap<string, ToolSchema> } | null = null
|
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 promptChangesRev = 0
|
||||||
private promptChangesCache: {
|
private promptChangesCache: {
|
||||||
rev: number
|
rev: number
|
||||||
@@ -614,6 +624,8 @@ export class Session implements ObservableSnapshot<ConversationSnapshot> {
|
|||||||
/** Per-event side effects (right column of the §A.9 dispatch table):
|
/** Per-event side effects (right column of the §A.9 dispatch table):
|
||||||
* chunk accumulation / partial clear on finalize / openCalls add-remove. */
|
* chunk accumulation / partial clear on finalize / openCalls add-remove. */
|
||||||
private applyEventSideEffects(event: SessionEvent, view?: ToolEventView): void {
|
private applyEventSideEffects(event: SessionEvent, view?: ToolEventView): void {
|
||||||
|
if (affectsModelRequests(event)) this.modelRequestsRev++
|
||||||
|
if (affectsCompactionRequests(event)) this.compactionRequestsRev++
|
||||||
// The `tool/code-dispatch-start`/`tool/code-dispatch` pair is declared by
|
// The `tool/code-dispatch-start`/`tool/code-dispatch` pair is declared by
|
||||||
// the host-side dsh-tools plugin whose types cannot enter the client
|
// the host-side dsh-tools plugin whose types cannot enter the client
|
||||||
// program (its host Context merges collide with the client's), so this
|
// program (its host Context merges collide with the client's), so this
|
||||||
@@ -783,6 +795,8 @@ export class Session implements ObservableSnapshot<ConversationSnapshot> {
|
|||||||
this.activeToolSchemas = new Map()
|
this.activeToolSchemas = new Map()
|
||||||
this.callSchemas = new Map()
|
this.callSchemas = new Map()
|
||||||
this.callSchemasRev++
|
this.callSchemasRev++
|
||||||
|
this.modelRequestsRev++
|
||||||
|
this.compactionRequestsRev++
|
||||||
this.promptChangesRev++
|
this.promptChangesRev++
|
||||||
for (let i = 0; i < this.events.length; i++) {
|
for (let i = 0; i < this.events.length; i++) {
|
||||||
const event = this.events[i]
|
const event = this.events[i]
|
||||||
@@ -823,6 +837,24 @@ export class Session implements ObservableSnapshot<ConversationSnapshot> {
|
|||||||
if (this.callSchemasCache === null || this.callSchemasCache.rev !== this.callSchemasRev) {
|
if (this.callSchemasCache === null || this.callSchemasCache.rev !== this.callSchemasRev) {
|
||||||
this.callSchemasCache = { rev: this.callSchemasRev, value: new Map(this.callSchemas) }
|
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(this.events),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if (
|
||||||
|
this.compactionRequestsCache === null
|
||||||
|
|| this.compactionRequestsCache.rev !== this.compactionRequestsRev
|
||||||
|
) {
|
||||||
|
this.compactionRequestsCache = {
|
||||||
|
rev: this.compactionRequestsRev,
|
||||||
|
value: deriveCompactionRequests(this.events),
|
||||||
|
}
|
||||||
|
}
|
||||||
if (this.queueCache === null || this.queueCache.rev !== this.queueRev) {
|
if (this.queueCache === null || this.queueCache.rev !== this.queueRev) {
|
||||||
this.queueCache = { rev: this.queueRev, value: this.queued.map(entry => entry.row) }
|
this.queueCache = { rev: this.queueRev, value: this.queued.map(entry => entry.row) }
|
||||||
}
|
}
|
||||||
@@ -840,8 +872,8 @@ export class Session implements ObservableSnapshot<ConversationSnapshot> {
|
|||||||
sessionId: this.sessionId,
|
sessionId: this.sessionId,
|
||||||
nodes,
|
nodes,
|
||||||
contexts,
|
contexts,
|
||||||
compactionRequests: deriveCompactionRequests(this.events),
|
compactionRequests: this.compactionRequestsCache.value,
|
||||||
requestAttempts: deriveModelRequests(this.events),
|
requestAttempts: this.modelRequestsCache.value,
|
||||||
promptChanges: this.promptChangesCache.value,
|
promptChanges: this.promptChangesCache.value,
|
||||||
foldDegraded: degraded,
|
foldDegraded: degraded,
|
||||||
partial,
|
partial,
|
||||||
@@ -901,6 +933,28 @@ function modelRequestKey(turn: number, step: number): string {
|
|||||||
return `${turn}\u0000${step}`
|
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. */
|
/** Project every durable step into one provider request, retaining failed retry attempts. */
|
||||||
function deriveModelRequests(events: readonly SessionEvent[]): readonly ModelRequestView[] {
|
function deriveModelRequests(events: readonly SessionEvent[]): readonly ModelRequestView[] {
|
||||||
const requests: ModelRequestView[] = []
|
const requests: ModelRequestView[] = []
|
||||||
|
|||||||
@@ -844,20 +844,27 @@ describe('reference stability (the memo contract)', () => {
|
|||||||
await session.open()
|
await session.open()
|
||||||
const feed = (event: SessionEvent) => { session.handleMuxEnvelope('r' as never, { type: 'session/event', sessionId: SID, event }) }
|
const feed = (event: SessionEvent) => { session.handleMuxEnvelope('r' as never, { type: 'session/event', sessionId: SID, event }) }
|
||||||
feed(ev.turnStart(6, 1))
|
feed(ev.turnStart(6, 1))
|
||||||
feed(ev.toolCall(7, 1, 'c1', 'echo', '{}'))
|
feed(ev.stepStart(7, 1))
|
||||||
|
feed(ev.toolCall(8, 1, 'c1', 'echo', '{}'))
|
||||||
session.handleMuxEnvelope('ra' as never, { type: 'approval/requested', sessionId: SID, approvalId: 'ap1' as never, toolName: 'rm' })
|
session.handleMuxEnvelope('ra' as never, { type: 'approval/requested', sessionId: SID, approvalId: 'ap1' as never, toolName: 'rm' })
|
||||||
const before = session.getSnapshot()
|
const before = session.getSnapshot()
|
||||||
// A chunk storm touches partial/nodes only: runningCalls and pending must keep identity.
|
// A chunk storm touches partial/nodes only: unrelated projections keep identity.
|
||||||
feed(ev.chunkStart(8, 1))
|
feed(ev.chunkStart(9, 1))
|
||||||
feed(ev.chunkText(9, 1, '与工具无关的流式'))
|
feed(ev.chunkText(10, 1, '与工具无关的流式'))
|
||||||
const after = session.getSnapshot()
|
const after = session.getSnapshot()
|
||||||
expect(after).not.toBe(before)
|
expect(after).not.toBe(before)
|
||||||
expect(after.runningCalls).toBe(before.runningCalls)
|
expect(after.runningCalls).toBe(before.runningCalls)
|
||||||
expect(after.pending).toBe(before.pending)
|
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.
|
// And a mutation on the tracked domain swaps that array.
|
||||||
feed(ev.toolResult(10, 1, 'c1', 'ECHO'))
|
feed(ev.toolResult(11, 1, 'c1', 'ECHO'))
|
||||||
const resolved = session.getSnapshot()
|
const resolved = session.getSnapshot()
|
||||||
expect(resolved.runningCalls).not.toBe(after.runningCalls)
|
expect(resolved.runningCalls).not.toBe(after.runningCalls)
|
||||||
expect(resolved.pending).toBe(after.pending)
|
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)
|
||||||
})
|
})
|
||||||
})
|
})
|
||||||
|
|||||||
@@ -270,14 +270,11 @@ function planSurfaceEvent(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/** Apply one event and return replacement metadata only when one occurred. */
|
/** Commit one validated transition and return replacement metadata when one occurred. */
|
||||||
function applySurfaceEvent(
|
function applySurfacePlan(
|
||||||
state: SurfaceFoldState,
|
state: SurfaceFoldState,
|
||||||
event: SessionEvent,
|
plan: SurfacePlan | undefined,
|
||||||
expectedSeq: number,
|
|
||||||
events: readonly SessionEvent[],
|
|
||||||
): SurfaceFoldReplacement | undefined {
|
): SurfaceFoldReplacement | undefined {
|
||||||
const plan = planSurfaceEvent(state, event, expectedSeq, events)
|
|
||||||
if (plan?.kind === 'append') {
|
if (plan?.kind === 'append') {
|
||||||
state.nodes.push(plan.seq)
|
state.nodes.push(plan.seq)
|
||||||
} else if (plan?.kind === 'replace') {
|
} else if (plan?.kind === 'replace') {
|
||||||
@@ -293,6 +290,16 @@ function applySurfaceEvent(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/** Apply one event and return replacement metadata only when one occurred. */
|
||||||
|
function applySurfaceEvent(
|
||||||
|
state: SurfaceFoldState,
|
||||||
|
event: SessionEvent,
|
||||||
|
expectedSeq: number,
|
||||||
|
events: readonly SessionEvent[],
|
||||||
|
): SurfaceFoldReplacement | undefined {
|
||||||
|
return applySurfacePlan(state, planSurfaceEvent(state, event, expectedSeq, events))
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Replay a complete session log through the canonical surface fold.
|
* Replay a complete session log through the canonical surface fold.
|
||||||
* @param events - session events in contiguous seq order.
|
* @param events - session events in contiguous seq order.
|
||||||
@@ -309,14 +316,35 @@ export function foldSurface(events: readonly SessionEvent[]): SurfaceFoldResult
|
|||||||
return { nodes: [...state.nodes], replacements }
|
return { nodes: [...state.nodes], replacements }
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/** Reconstruct every surface generation only for consumers that request history. */
|
||||||
|
function foldSurfaceContexts(events: readonly SessionEvent[]): SurfaceFoldContext[] {
|
||||||
|
const state = createFoldState()
|
||||||
|
const contexts: SurfaceFoldContext[] = []
|
||||||
|
let origin: SurfaceFoldReplacement | undefined
|
||||||
|
for (const [index, event] of events.entries()) {
|
||||||
|
const plan = planSurfaceEvent(state, event, index, events)
|
||||||
|
const priorNodes = plan?.kind === 'replace' ? [...state.nodes] : undefined
|
||||||
|
const replacement = applySurfacePlan(state, plan)
|
||||||
|
if (replacement === undefined || priorNodes === undefined) continue
|
||||||
|
contexts.push({
|
||||||
|
generation: state.replaceGeneration - 1,
|
||||||
|
nodes: priorNodes,
|
||||||
|
...(origin === undefined ? {} : { origin }),
|
||||||
|
})
|
||||||
|
origin = replacement
|
||||||
|
}
|
||||||
|
contexts.push({
|
||||||
|
generation: state.replaceGeneration,
|
||||||
|
nodes: [...state.nodes],
|
||||||
|
...(origin === undefined ? {} : { origin }),
|
||||||
|
})
|
||||||
|
return contexts
|
||||||
|
}
|
||||||
|
|
||||||
/** Incremental ordered surface view and append-boundary validator. */
|
/** Incremental ordered surface view and append-boundary validator. */
|
||||||
export class SurfaceManager implements SessionSurface {
|
export class SurfaceManager implements SessionSurface {
|
||||||
/** Shared transition state for the live surface. */
|
/** Shared transition state for the live surface. */
|
||||||
private _state = createFoldState()
|
private _state = createFoldState()
|
||||||
/** Frozen generations completed by replacements. */
|
|
||||||
private _contexts: SurfaceFoldContext[] = []
|
|
||||||
/** Replacement that created the live generation. */
|
|
||||||
private _contextOrigin: SurfaceFoldReplacement | undefined
|
|
||||||
/** Last processed seq; -1 folds a seeded log on first access. */
|
/** Last processed seq; -1 folds a seeded log on first access. */
|
||||||
private _lastProcessedSeq = -1
|
private _lastProcessedSeq = -1
|
||||||
|
|
||||||
@@ -343,17 +371,10 @@ export class SurfaceManager implements SessionSurface {
|
|||||||
return this._state.nodes
|
return this._state.nodes
|
||||||
}
|
}
|
||||||
|
|
||||||
/** Frozen generations followed by a detached snapshot of the live generation. */
|
/** Surface generations reconstructed on demand without burdening ordinary live sessions. */
|
||||||
get contexts(): readonly SurfaceFoldContext[] {
|
get contexts(): readonly SurfaceFoldContext[] {
|
||||||
if (this._lastProcessedSeq < this.log.length - 1) this._processDelta()
|
if (this._lastProcessedSeq < this.log.length - 1) this._processDelta()
|
||||||
return [
|
return foldSurfaceContexts(this.log)
|
||||||
...this._contexts,
|
|
||||||
{
|
|
||||||
generation: this._state.replaceGeneration,
|
|
||||||
nodes: [...this._state.nodes],
|
|
||||||
...(this._contextOrigin === undefined ? {} : { origin: this._contextOrigin }),
|
|
||||||
},
|
|
||||||
]
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/** Fold events appended since the previous access. */
|
/** Fold events appended since the previous access. */
|
||||||
@@ -361,17 +382,7 @@ export class SurfaceManager implements SessionSurface {
|
|||||||
for (let i = this._lastProcessedSeq + 1; i < this.log.length; i++) {
|
for (let i = this._lastProcessedSeq + 1; i < this.log.length; i++) {
|
||||||
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion -- bounded by the loop condition
|
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion -- bounded by the loop condition
|
||||||
const event = this.log[i]!
|
const event = this.log[i]!
|
||||||
const op = surfaceOpOf(event)
|
applySurfaceEvent(this._state, event, i, this.log)
|
||||||
const priorNodes = typeof op === 'object' ? [...this._state.nodes] : undefined
|
|
||||||
const replacement = applySurfaceEvent(this._state, event, i, this.log)
|
|
||||||
if (replacement !== undefined && priorNodes !== undefined) {
|
|
||||||
this._contexts.push({
|
|
||||||
generation: this._state.replaceGeneration - 1,
|
|
||||||
nodes: priorNodes,
|
|
||||||
...(this._contextOrigin === undefined ? {} : { origin: this._contextOrigin }),
|
|
||||||
})
|
|
||||||
this._contextOrigin = replacement
|
|
||||||
}
|
|
||||||
this._lastProcessedSeq = i
|
this._lastProcessedSeq = i
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -218,6 +218,7 @@ describe('SurfaceManager', () => {
|
|||||||
expect(s.surface.nodes).toEqual([1])
|
expect(s.surface.nodes).toEqual([1])
|
||||||
const manager = s.surface as unknown as { _state: object }
|
const manager = s.surface as unknown as { _state: object }
|
||||||
expect(Object.hasOwn(manager._state, 'replacements')).toBe(false)
|
expect(Object.hasOwn(manager._state, 'replacements')).toBe(false)
|
||||||
|
expect(Object.hasOwn(manager, '_contexts')).toBe(false)
|
||||||
expect(foldSurface(s.events).replacements).toEqual([
|
expect(foldSurface(s.events).replacements).toEqual([
|
||||||
{ seq: 1, start: 0, end: 0, shadowedSeqs: [0] },
|
{ seq: 1, start: 0, end: 0, shadowedSeqs: [0] },
|
||||||
])
|
])
|
||||||
|
|||||||
Reference in New Issue
Block a user