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

Unify session surface validation

Browse Source
This commit is contained in:
Hypatia May
2026-07-14 13:49:36 +08:00
parent d9714fb30b
commit dbe65e1d13
17 changed files with 327 additions and 405 deletions
Download Patch File Download Diff File Expand all files Collapse all files

57
packages/core/session/src/index.ts
View File

@@ -15,7 +15,7 @@ import type { ContentBlock, Message, MessageSource } from '@deepseek-ai/dsh-llm'
import { SESSION_FORMAT_VERSION, SessionId } from './types.ts'
import type { CreateSessionOptions, EpochHeader, SessionEvent, SessionEventMap, SessionEventType, SessionHeader, SurfaceIntent, SurfaceEventType } from './types.ts'
import { snapshotJsonValue } from './json.ts'
import { SurfaceManager, validateSurfaceMetadata } from './surface.ts'
import { SurfaceManager } from './surface.ts'
import { foldRequestHeader } from './request-header.ts'
export * from './types.ts'
@@ -23,7 +23,7 @@ export { isJsonValue, snapshotJsonValue } from './json.ts'
export type { JsonValue } from './json.ts'
export { interruptedTurnClosers } from './repair.ts'
export type { SurfaceFoldReplacement, SurfaceFoldResult, SurfaceNode } from './surface.ts'
export { foldSurface, isSurfaceEvent, isSurfaceEligibleType, validateSurfaceMetadata } from './surface.ts'
export { foldSurface, isSurfaceEvent, isSurfaceEligibleType } from './surface.ts'
export { isToolPairingBalanced } from './tool-pairing.ts'
export { applyHeaderDelta, canonicalHeader, diffHeader, foldRequestHeader, headerEquals } from './request-header.ts'
@@ -223,13 +223,15 @@ const attachments = new WeakMap<Session, SessionEntry>()
*/
export class Session {
private log: SessionEvent[] = []
/** Incremental acceptance state, kept separate from the public lazy view. */
private readonly surfaceValidator = new SurfaceManager(this.log)
/**
* Derived surface — a cached linked list of message-producing events.
* Lazily rebuilt from `surfaceOp` markers in the log; processes only new
* events (delta) on each access — the log is append-only, so prior events
* never change.
* `append`. Undefined until first accessed (including after fork/seed).
* Undefined until first accessed (including after fork/seed).
*/
private _surface: SurfaceManager | undefined
@@ -258,7 +260,7 @@ export class Session {
// `seq = log.length` contract the whole system relies on). Without this,
// a bad seed would surface only later as a backend rejection or a silent
// divergence between the live log and disk.
this.log = Array.from(seed, (source, index) => {
for (const [index, source] of seed.entries()) {
// The seed is a persistence/replay boundary: validate and detach the
// complete event in one lossless-JSON pass.
const snapshot = snapshotJsonValue(source)
@@ -269,23 +271,16 @@ export class Session {
if (snapshot.seq !== index) {
throw new Error(`seed event at index ${index} has seq ${snapshot.seq} (expected ${index}); seed must be contiguous from 0`)
}
// Surface-eligible events MUST carry a surfaceOp marker — the surface is
// the sole source of derived history, so a marker-less message event
// would load fine yet vanish from deriveMessages(). `append` enforces
// this at compile time via its typed overload; a seed arrives as raw
// SessionEvent[] (replay/fork/load), bypassing that, so re-check at
// runtime here rather than silently resuming with empty history.
let violation: ReturnType<typeof validateSurfaceMetadata>
// A seed is accepted incrementally through the same transition as a
// live append and a full-log fold. The candidate is planned before it
// enters `log`, so a failure cannot partially mutate the surface.
try {
violation = validateSurfaceMetadata(snapshot)
this.surfaceValidator.validateNext(snapshot)
} catch (error: unknown) {
throw new Error(`invalid seed event at index ${index}: ${error instanceof Error ? error.message : 'invalid surface metadata'}`)
}
if (violation !== undefined) {
throw new Error(`invalid seed event at index ${index}: ${violation.message}`)
}
return deepFreeze(snapshot)
})
this.log.push(deepFreeze(snapshot))
}
}
this.header = snapshotSessionHeader(id, header)
}
@@ -332,7 +327,10 @@ export class Session {
* @throws if `data` or surface metadata is not losslessly JSON-serializable
* (BigInt, function, symbol, undefined, negative zero, non-finite number,
* circular reference, sparse array, or an exotic object such as
* Map/Set/Date/class instance). One recursive pass reads, validates, and
* Map/Set/Date/class instance), or when the candidate violates the
* canonical surface contract (marker shape and eligibility, unique known
* earlier provenance, positional replacement validity, and complete
* shadowed-node coverage). One recursive pass reads, validates, and
* copies each nested value once, so a stateful getter cannot supply one value
* to validation and another to storage. The event log is the durable source
* of truth, so a bad event fails at the append site rather than later during
@@ -358,26 +356,21 @@ export class Session {
if (surfaceMetadataSnapshot === undefined) {
throw new Error(`session event "${type}" carries non-JSON-serializable surface metadata`)
}
const surfaceViolation = validateSurfaceMetadata({
type,
seq: this.log.length,
...(surfaceMetadataSnapshot as { surfaceOp?: unknown; sourceEventSeqs?: unknown }),
})
if (surfaceViolation !== undefined) throw new Error(surfaceViolation.message)
const entry = attachments.get(this)
if (entry?.appending) {
throw new Error('session append cannot reenter while another append is being published')
}
const event = deepFreeze({
type,
seq: this.log.length,
time: Date.now(),
data: dataSnapshot,
...(surfaceMetadataSnapshot as { surfaceOp?: unknown; sourceEventSeqs?: unknown }),
} as unknown as SessionEvent<T>)
this.surfaceValidator.validateNext(event as SessionEvent)
if (entry !== undefined) entry.appending = true
try {
const event = deepFreeze({
type,
seq: this.log.length,
time: Date.now(),
data: dataSnapshot,
...surfaceMetadataSnapshot,
} as unknown as SessionEvent<T>)
let callbacks: SessionCallback[] | undefined
const callbackArgs: unknown[] = [this, event]
if (entry !== undefined) {

312
packages/core/session/src/surface.ts
View File

@@ -81,165 +81,118 @@ export interface SurfaceFoldResult {
replacements: SurfaceFoldReplacement[]
}
/**
* Validate one event's surface metadata through the canonical structural and
* provenance contract. Structural validation always runs; when `knownSeqs` is
* supplied, provenance must additionally name unique known earlier events and
* cover every shadowed surface node. The tagged result lets callers retain
* their own surface-versus-provenance error taxonomy.
* @param event - event whose `surfaceOp` and `sourceEventSeqs` are being checked.
* @param knownSeqs - seqs preceding `event`, or `undefined` for local shape validation only.
* @param shadowedSeqs - surface nodes directly removed by this event.
* @returns the first tagged contract violation, or `undefined` when valid.
*/
export function validateSurfaceMetadata(
event: Pick<SessionEvent, 'type' | 'seq'> & {
surfaceOp?: unknown
sourceEventSeqs?: unknown
},
knownSeqs?: ReadonlySet<number>,
shadowedSeqs: readonly number[] = [],
): { kind: 'surface' | 'provenance'; message: string } | undefined {
const eligible = isSurfaceEligibleType(event.type)
const surfaceOp = event.surfaceOp
const sources = event.sourceEventSeqs
if (!eligible && surfaceOp !== undefined) {
return {
kind: 'surface',
message: `session event "${event.type}" is not surface-eligible and cannot carry surfaceOp`,
}
}
if (eligible && surfaceOp === undefined) {
return {
kind: 'surface',
message: `session event "${event.type}" is surface-eligible and requires a surfaceOp marker`,
}
}
if (surfaceOp !== undefined && surfaceOp !== 'append') {
if (surfaceOp === null || typeof surfaceOp !== 'object' || Array.isArray(surfaceOp)) {
return {
kind: 'surface',
message: `session event "${event.type}" carries an invalid surfaceOp`,
}
}
const op = surfaceOp as Record<string, unknown>
const keys = Object.keys(op)
if (keys.length !== 3 || !Object.hasOwn(op, 'op') || !Object.hasOwn(op, 'start') || !Object.hasOwn(op, 'end')
|| op['op'] !== 'replace'
|| typeof op['start'] !== 'number' || !Number.isSafeInteger(op['start']) || op['start'] < 0
|| typeof op['end'] !== 'number' || !Number.isSafeInteger(op['end']) || op['end'] < 0) {
return {
kind: 'surface',
message: `session event "${event.type}" carries an invalid replace surfaceOp`,
}
}
}
if (sources !== undefined && !eligible) {
return {
kind: 'provenance',
message: `${event.type} cannot carry sourceEventSeqs (non-surface event)`,
}
}
if (sources !== undefined && !Array.isArray(sources)) {
return {
kind: 'provenance',
message: `sourceEventSeqs on event at seq ${event.seq} must be an array when present`,
}
}
if (Array.isArray(sources)
&& sources.some(source => typeof source !== 'number' || !Number.isSafeInteger(source) || source < 0)) {
return {
kind: 'provenance',
message: `session event "${event.type}" sourceEventSeqs must contain non-negative safe integers`,
}
}
if (knownSeqs === undefined) return
const sourceSeqs = sources as number[] | undefined
if (sourceSeqs !== undefined && sourceSeqs.length === 0) {
return { kind: 'provenance', message: 'sourceEventSeqs must not be empty when present' }
}
const unique = new Set<number>()
for (const source of sourceSeqs ?? []) {
if (unique.has(source)) {
return { kind: 'provenance', message: 'sourceEventSeqs must not contain duplicates' }
}
unique.add(source)
if (source >= event.seq) {
return {
kind: 'provenance',
message: `sourceEventSeqs must reference earlier events: ${source} >= current seq ${event.seq}`,
}
}
if (!knownSeqs.has(source)) {
return { kind: 'provenance', message: `sourceEventSeqs references unknown seq ${source}` }
}
}
const sourceSet = new Set(sourceSeqs ?? [])
const missing = shadowedSeqs.filter(seq => !sourceSet.has(seq))
if (missing.length > 0) {
return {
kind: 'provenance',
message: `surface replace: sourceEventSeqs must include every shadowed surface node; missing ${missing.join(', ')}`,
}
}
return undefined
}
/** Mutable state shared by the incremental manager and the full-log fold. */
interface SurfaceFoldState {
nodes: SurfaceNode[]
nodeBySeq: Map<number, SurfaceNode>
knownSeqs: Set<number>
replaceGeneration: number
}
/** A validated replacement transition that has not mutated fold state yet. */
interface SurfaceReplacePlan extends SurfaceFoldReplacement {
kind: 'replace'
startIdx: number
endIdx: number
}
/** One validated surface transition that has not mutated fold state yet. */
type SurfacePlan =
| { kind: 'append'; seq: number }
| SurfaceReplacePlan
/** Create an empty surface fold state. */
function createFoldState(replaceGeneration = 0): SurfaceFoldState {
return {
nodes: [],
nodeBySeq: new Map(),
knownSeqs: new Set(),
replaceGeneration,
}
}
/** Apply one event and return replacement metadata only when one occurred. */
function applySurfaceEvent(
state: SurfaceFoldState,
event: SessionEvent,
): SurfaceFoldReplacement | undefined {
const violation = validateSurfaceMetadata(event)
if (violation?.kind === 'surface') throw new Error(violation.message)
if (!isSurfaceEligibleType(event.type)) return
// The canonical metadata validation above proves this runtime shape.
const surfaceEvent = event as SurfaceEvent
/** Whether a runtime value is a non-negative safe event sequence. */
function isEventSeq(value: unknown): value is number {
return typeof value === 'number' && Number.isSafeInteger(value) && value >= 0
}
if (surfaceEvent.surfaceOp === 'append') {
const tail = state.nodes.length > 0 ? state.nodes[state.nodes.length - 1] : undefined
const node: SurfaceNode = { seq: surfaceEvent.seq, prev: tail?.seq ?? null, next: null }
if (tail) tail.next = surfaceEvent.seq
state.nodes.push(node)
state.nodeBySeq.set(surfaceEvent.seq, node)
/** Whether a runtime value is the exact positional-replacement shape. */
function isReplaceOp(value: object): value is Extract<SurfaceOp, { op: 'replace' }> {
const op = value as Record<string, unknown>
return Object.keys(op).length === 3
&& Object.hasOwn(op, 'op')
&& Object.hasOwn(op, 'start')
&& Object.hasOwn(op, 'end')
&& op['op'] === 'replace'
&& isEventSeq(op['start'])
&& isEventSeq(op['end'])
}
/** Validate event-local metadata and narrow a surface-eligible event. */
function surfaceEventOf(event: SessionEvent): SurfaceEvent | undefined {
const raw = event as SessionEvent & { surfaceOp?: unknown; sourceEventSeqs?: unknown }
if (!isSurfaceEligibleType(event.type)) {
if (raw.surfaceOp !== undefined) {
throw new Error(`session event "${event.type}" is not surface-eligible and cannot carry surfaceOp`)
}
if (raw.sourceEventSeqs !== undefined) {
throw new Error(`session event "${event.type}" is not surface-eligible and cannot carry sourceEventSeqs`)
}
return
}
if (raw.surfaceOp === undefined) {
throw new Error(`session event "${event.type}" is surface-eligible and requires a surfaceOp marker`)
}
if (raw.surfaceOp !== 'append') {
if (raw.surfaceOp === null || typeof raw.surfaceOp !== 'object' || Array.isArray(raw.surfaceOp)) {
throw new Error(`session event "${event.type}" carries an invalid surfaceOp`)
}
if (!isReplaceOp(raw.surfaceOp)) {
throw new Error(`session event "${event.type}" carries an invalid replace surfaceOp`)
}
}
if (raw.sourceEventSeqs !== undefined && !Array.isArray(raw.sourceEventSeqs)) {
throw new Error(`sourceEventSeqs on event at seq ${event.seq} must be an array when present`)
}
if (Array.isArray(raw.sourceEventSeqs) && !raw.sourceEventSeqs.every(isEventSeq)) {
throw new Error(`session event "${event.type}" sourceEventSeqs must contain non-negative safe integers`)
}
return event as SurfaceEvent
}
return {
seq: surfaceEvent.seq,
start: surfaceEvent.surfaceOp.start,
end: surfaceEvent.surfaceOp.end,
shadowedSeqs: replaceSurface(state, surfaceEvent.seq, surfaceEvent.surfaceOp),
/** Validate provenance against prior log entries and the replacement range. */
function assertProvenance(
event: SurfaceEvent,
knownSeqs: ReadonlySet<number>,
shadowedSeqs: readonly number[],
): void {
const sources = event.sourceEventSeqs
if (sources !== undefined && sources.length === 0) {
throw new Error('sourceEventSeqs must not be empty when present')
}
const sourceSet = new Set(sources ?? [])
if (sources !== undefined && sourceSet.size !== sources.length) {
throw new Error('sourceEventSeqs must not contain duplicates')
}
for (const source of sources ?? []) {
if (source >= event.seq) {
throw new Error(`sourceEventSeqs must reference earlier events: ${source} >= current seq ${event.seq}`)
}
if (!knownSeqs.has(source)) {
throw new Error(`sourceEventSeqs references unknown seq ${source}`)
}
}
const missing = shadowedSeqs.filter(seq => !sourceSet.has(seq))
if (missing.length > 0) {
throw new Error(`surface replace: sourceEventSeqs must include every shadowed surface node; missing ${missing.join(', ')}`)
}
}
/** Apply one positional replacement and return the nodes it removed. */
function replaceSurface(
/** Locate one replacement range without mutating the current fold state. */
function replacementRange(
state: SurfaceFoldState,
newSeq: number,
op: Extract<SurfaceOp, { op: 'replace' }>,
): number[] {
): Pick<SurfaceReplacePlan, 'startIdx' | 'endIdx' | 'shadowedSeqs'> {
const startNode = state.nodeBySeq.get(op.start)
if (!startNode) {
throw new Error(`surface replace: start seq ${op.start} not found in surface`)
@@ -253,6 +206,35 @@ function replaceSurface(
if (startIdx > endIdx) {
throw new Error(`surface replace: start seq ${op.start} (index ${startIdx}) is after end seq ${op.end} (index ${endIdx})`)
}
return {
startIdx,
endIdx,
shadowedSeqs: state.nodes.slice(startIdx, endIdx + 1).map(node => node.seq),
}
}
/** Validate one event and prepare its atomic fold transition. */
function planSurfaceEvent(state: SurfaceFoldState, event: SessionEvent): SurfacePlan | undefined {
const surfaceEvent = surfaceEventOf(event)
if (surfaceEvent === undefined) return
if (surfaceEvent.surfaceOp === 'append') {
assertProvenance(surfaceEvent, state.knownSeqs, [])
return { kind: 'append', seq: event.seq }
}
const range = replacementRange(state, surfaceEvent.surfaceOp)
assertProvenance(surfaceEvent, state.knownSeqs, range.shadowedSeqs)
return {
kind: 'replace',
seq: event.seq,
start: surfaceEvent.surfaceOp.start,
end: surfaceEvent.surfaceOp.end,
...range,
}
}
/** Apply one already-validated positional replacement. */
function replaceSurface(state: SurfaceFoldState, plan: SurfaceReplacePlan): void {
const { startIdx, endIdx } = plan
const removed = state.nodes.splice(startIdx, endIdx - startIdx + 1)
for (const node of removed) state.nodeBySeq.delete(node.seq)
@@ -260,16 +242,40 @@ function replaceSurface(
const prevNode = startIdx > 0 ? state.nodes[startIdx - 1] : undefined
const nextNode = startIdx < state.nodes.length ? state.nodes[startIdx] : undefined
const newNode: SurfaceNode = {
seq: newSeq,
seq: plan.seq,
prev: prevNode?.seq ?? null,
next: nextNode?.seq ?? null,
}
if (prevNode) prevNode.next = newSeq
if (nextNode) nextNode.prev = newSeq
if (prevNode) prevNode.next = plan.seq
if (nextNode) nextNode.prev = plan.seq
state.nodes.splice(startIdx, 0, newNode)
state.nodeBySeq.set(newSeq, newNode)
state.nodeBySeq.set(plan.seq, newNode)
state.replaceGeneration += 1
return removed.map(node => node.seq)
}
/** Apply one event and return replacement metadata only when one occurred. */
function applySurfaceEvent(
state: SurfaceFoldState,
event: SessionEvent,
): SurfaceFoldReplacement | undefined {
const plan = planSurfaceEvent(state, event)
if (plan?.kind === 'append') {
const tail = state.nodes.at(-1)
const node: SurfaceNode = { seq: plan.seq, prev: tail?.seq ?? null, next: null }
if (tail) tail.next = plan.seq
state.nodes.push(node)
state.nodeBySeq.set(plan.seq, node)
} else if (plan?.kind === 'replace') {
replaceSurface(state, plan)
}
state.knownSeqs.add(event.seq)
if (plan?.kind !== 'replace') return
return {
seq: plan.seq,
start: plan.start,
end: plan.end,
shadowedSeqs: plan.shadowedSeqs,
}
}
/**
@@ -280,8 +286,9 @@ function replaceSurface(
* models cannot disagree with `deriveMessages()` about replacement ranges.
* @param events - session events in contiguous seq order.
* @returns the current surface and every positional replacement.
* @throws when an event violates the `surfaceOp` type/marker contract, or a
* replacement names nodes that are absent or reversed on the current surface.
* @throws when any event violates the unified surface contract: metadata must
* be well shaped and type-eligible, provenance must name unique known earlier
* events, and a positional replacement must name and cite its complete range.
*/
export function foldSurface(events: readonly SessionEvent[]): SurfaceFoldResult {
const state = createFoldState()
@@ -297,11 +304,10 @@ export function foldSurface(events: readonly SessionEvent[]): SurfaceFoldResult
}
/**
* Maintains a cached linked list of surface nodes, rebuilt lazily from
* `surfaceOp` markers in the event log. Because the log is append-only, it
* processes only the delta since the last rebuild — new events are folded
* into the existing surface in O(new events) rather than rescanning the
* whole log.
* Maintains a cached linked list of surface nodes and validates each candidate
* before it enters the event log. Because the log is append-only, it processes
* only committed deltas and plans the candidate without mutation rather than
* rescanning the whole log.
*/
export class SurfaceManager {
/** Incremental state shared with the complete surface fold. */
@@ -311,6 +317,18 @@ export class SurfaceManager {
constructor(private log: readonly SessionEvent[]) {}
/**
* Validate one candidate as the next log event without applying it. The
* committed log is folded first, then the candidate's complete surface and
* provenance transition is planned atomically; a failure leaves the current
* surface unchanged.
* @param event - candidate event that has not entered `log` yet.
*/
validateNext(event: SessionEvent): void {
if (this._lastProcessedSeq < this.log.length - 1) this._processDelta()
planSurfaceEvent(this._state, event)
}
/**
* Reset to unprocessed state. Call after the log has been replaced
* wholesale (e.g. after Session seed). Not needed for normal appends —
@@ -353,7 +371,7 @@ export class SurfaceManager {
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion
const event = this.log[i]!
applySurfaceEvent(this._state, event)
this._lastProcessedSeq = i
}
this._lastProcessedSeq = this.log.length - 1
}
}