From dbe65e1d13fb8c4d482177b60fab7cd552c1facc Mon Sep 17 00:00:00 2001 From: Hypatia May Date: Tue, 14 Jul 2026 13:49:36 +0800 Subject: [PATCH] Unify session surface validation --- docs/cordis-catalog/services.md | 2 +- docs/core-data-structures/session-query.md | 1 - .../2026-07-13-session-query-tracing.md | 2 +- packages/core/session/README.md | 7 +- packages/core/session/src/index.ts | 57 ++-- packages/core/session/src/surface.ts | 312 +++++++++--------- packages/core/session/tests/session.spec.ts | 27 +- packages/core/session/tests/surface.spec.ts | 125 ++++--- .../core/session/tests/tool-pairing.spec.ts | 3 +- .../session-query/session-query/README.md | 4 +- .../session-query/session-query/src/config.ts | 1 - .../session-query/src/tracing.ts | 34 +- .../session-query/tests/session-query.spec.ts | 14 +- .../session-query/tests/tracing.spec.ts | 12 +- packages/support/invariants/README.md | 3 +- packages/support/invariants/src/index.ts | 91 +---- .../invariants/tests/invariants.spec.ts | 37 +-- 17 files changed, 327 insertions(+), 405 deletions(-) diff --git a/docs/cordis-catalog/services.md b/docs/cordis-catalog/services.md index e62a96a7bb..b948ea898f 100644 --- a/docs/cordis-catalog/services.md +++ b/docs/cordis-catalog/services.md @@ -247,7 +247,7 @@ list(): Session[] fork(source: SessionForkSource, boundary?: number, childSessionId?: SessionId): Session ``` -Source: [`packages/core/session/src/index.ts:557`](../../packages/core/session/src/index.ts) +Source: [`packages/core/session/src/index.ts:550`](../../packages/core/session/src/index.ts) ## `ctx.skills` — `SkillService` diff --git a/docs/core-data-structures/session-query.md b/docs/core-data-structures/session-query.md index 71fb956dcb..86d2259f7f 100644 --- a/docs/core-data-structures/session-query.md +++ b/docs/core-data-structures/session-query.md @@ -112,7 +112,6 @@ export type SessionQueryErrorCode = | 'SESSION_QUERY_EVENT_NOT_FOUND' | 'SESSION_QUERY_INVALID_CONFIG' | 'SESSION_QUERY_INVALID_LINEAGE' - | 'SESSION_QUERY_INVALID_PROVENANCE' | 'SESSION_QUERY_INVALID_SURFACE' | 'SESSION_QUERY_INVALID_WINDOW' | 'SESSION_QUERY_PERSISTENCE_FAILED' diff --git a/docs/rfc/implemented/feature/2026-07-13-session-query-tracing.md b/docs/rfc/implemented/feature/2026-07-13-session-query-tracing.md index ff2e9b7c23..9ebb8ee961 100644 --- a/docs/rfc/implemented/feature/2026-07-13-session-query-tracing.md +++ b/docs/rfc/implemented/feature/2026-07-13-session-query-tracing.md @@ -16,7 +16,7 @@ Session relationships are encoded across immutable headers, positional surface o ## Validation boundary -Event tracing checks target existence before surface analysis. Before returning a trace it validates the whole loaded log through `dsh-session`'s shared surface-metadata checker: surface markers obey event-type eligibility, provenance belongs only to surface event types, present arrays are nonempty and duplicate-free, every source is a known earlier seq, and every positional replacement names all surface nodes it removed. Surface-marker and positional-fold failures use `SESSION_QUERY_INVALID_SURFACE`; provenance failures use `SESSION_QUERY_INVALID_PROVENANCE`. `listEvents()` remains a surface-classification operation and does not acquire trace-specific provenance rejection. +Event tracing checks target existence before surface analysis. Both event listing and tracing then use `dsh-session`'s one-pass surface fold, which accepts or rejects the loaded log as a whole: surface markers obey event-type eligibility, provenance belongs only to surface event types, present arrays are nonempty and duplicate-free, every source is a known earlier seq, and every positional replacement names and cites all surface nodes it removes. Every contract failure uses `SESSION_QUERY_INVALID_SURFACE`; there is no weaker classification-only surface standard. All returned records and arrays are detached. A known live event trace never consults persistence; persisted event traces preserve the exact-read list/load consistency check. Session lineage is necessarily a cross-corpus operation and therefore preserves cross-corpus persistence failure semantics. diff --git a/packages/core/session/README.md b/packages/core/session/README.md index fc7e6d4300..7632212621 100644 --- a/packages/core/session/README.md +++ b/packages/core/session/README.md @@ -32,10 +32,10 @@ The store pairs announced creation with disposal, publishes each append, and pro Plain class (not a Cordis Service). Create via `ctx.sessions.create()`. -- `session.append(type, data, opts?): SessionEvent` — synchronous, never blocks on I/O. At this durable boundary, data and surface metadata are lossless-JSON snapshotted and deep-frozen. For an attached session, a reentrant append during dispatch/observer publication rejects, and detach waits for that publication to unwind. Callbacks resolve before the log push; the push is the commit point, after which each observer failure is contained independently. Runtime surface validation covers widened unions and raw seed/load logs. +- `session.append(type, data, opts?): SessionEvent` — synchronous, never blocks on I/O. At this durable boundary, data and surface metadata are lossless-JSON snapshotted and deep-frozen, then the same atomic surface transition used by replay validates marker shape, provenance, and complete replacement coverage before the log changes. For an attached session, a reentrant append during dispatch/observer publication rejects, and detach waits for that publication to unwind. Callbacks resolve before the log push; the push is the commit point, after which each observer failure is contained independently. - `session.deriveMessages(): Message[]` — the LLM message history, CACHED: each surface node is projected exactly once, when first seen (O(new nodes) per call; a surface rewrite rebuilds via `surface.replaceGeneration`). Returns a fresh array per call over shared, deep-frozen `Message` objects. Each projection reuses the already deep-frozen content in its durable log event, so no second deep clone is needed and a consumer still cannot mutate logged data. The surface is the single source of derived history — there is no raw-log fallback. - `session.deriveEventMessage(event): Message | null` — the per-event projection `deriveMessages()` folds: a fresh message wrapper that reuses the event's already frozen content, or `null` when the event produces none (a non-surface event, or an empty-content `assistant/message` hosting only usage). External reconstructors and the dev invariant fold the same function over a log prefix's surface, so no two paths can disagree about what a request's messages were (the reconstructability RFC). -- `session.surface: SurfaceManager` — the derived surface, 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. `surface.replaceGeneration` is the rewrite signal: bumped by every folded `replace` and by `invalidate()`, never reset, so an incremental consumer comparing generations cannot be fooled. +- `session.surface: SurfaceManager` — the derived surface, lazily rebuilt from `surfaceOp` markers in the log. It processes only new events (delta) on each access; event acceptance uses a separate manager with the same transition so validation does not eagerly mutate this public view. `surface.replaceGeneration` is the rewrite signal: bumped by every folded `replace` and by `invalidate()`, never reset, so an incremental consumer comparing generations cannot be fooled. - `session.events` — a cached, frozen array snapshot over deep-frozen events. Repeated reads without an append return the same array; an append invalidates the cache and the next read returns a new snapshot, while earlier snapshots stay unchanged. Neither a cast nor a retained reference can push into the live log or rewrite an accepted event. - `session.seq`, `session.id` — current sequence and readonly typed identity. - `session.header: SessionHeader` — detached, deep-frozen creation metadata (`version`, `id`, `createdAt`, optional `cwd`/`parentSession`/`seedLength`). Construction validates the durable record and requires its id to match `session.id`. @@ -49,8 +49,7 @@ Durable values need one accepted representation, not a check followed by a secon - `SurfaceOp` — how a surface node entered the linked list: `'append'` (normal tail append) or `{ op: 'replace', start, end }` (replace nodes from `start` through `end` inclusive — both must be valid surface node seqs; `start === end` replaces a single node). Used by compaction to shadow old nodes without deleting them. - `SurfaceIntent` — `{ surfaceOp: SurfaceOp; sourceEventSeqs?: number[] }`, the required third parameter to `session.append()` for surface-eligible types. - `SurfaceNode` — `{ seq: number; prev: number | null; next: number | null }`, one node in the surface linked list. -- `foldSurface(events)` — replay the canonical surface transitions into detached current nodes and actual replacement ranges, rejecting misplaced, missing, malformed, or positionally invalid `surfaceOp` metadata while leaving provenance validation to callers. `SurfaceManager` shares the same transitions while retaining its incremental cache. -- `validateSurfaceMetadata(event, knownSeqs?, shadowedSeqs?)` — canonical structural and provenance check shared by session acceptance, surface folding, incremental invariants, and exact readers. It tags violations as `surface` or `provenance` so callers retain their error taxonomy; omit `knownSeqs` for local shape validation only. +- `foldSurface(events)` — replay the one canonical surface contract into detached current nodes and actual replacement ranges. The same pass rejects misplaced or malformed metadata, empty or duplicate provenance, unknown or non-earlier sources, invalid positional ranges, and replacements that fail to cite every shadowed surface node; `SurfaceManager` shares the atomic transition while retaining its incremental cache. - `isSurfaceEvent(event)` / `isSurfaceEligibleType(type)` — the first narrows a `SessionEvent` to a fully-formed surface node (type is surface-eligible AND `surfaceOp` present); the second is the type-only check (is this one of the five `SurfaceEventType` values?), used to detect a surface-eligible event MISSING its marker — e.g. when validating a seed/load log. ### Request-header reconstruction (`request-header.ts`) diff --git a/packages/core/session/src/index.ts b/packages/core/session/src/index.ts index 88b3c15aff..fc3a7fcd29 100644 --- a/packages/core/session/src/index.ts +++ b/packages/core/session/src/index.ts @@ -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() */ 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 + // 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) + 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) let callbacks: SessionCallback[] | undefined const callbackArgs: unknown[] = [this, event] if (entry !== undefined) { diff --git a/packages/core/session/src/surface.ts b/packages/core/session/src/surface.ts index c8642d18fc..7e1da76832 100644 --- a/packages/core/session/src/surface.ts +++ b/packages/core/session/src/surface.ts @@ -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 & { - surfaceOp?: unknown - sourceEventSeqs?: unknown - }, - knownSeqs?: ReadonlySet, - 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 - 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() - 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 + knownSeqs: Set 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 { + const op = value as Record + 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, + 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, -): number[] { +): Pick { 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 } } diff --git a/packages/core/session/tests/session.spec.ts b/packages/core/session/tests/session.spec.ts index fe44de871f..e3e5cc76c2 100644 --- a/packages/core/session/tests/session.spec.ts +++ b/packages/core/session/tests/session.spec.ts @@ -301,12 +301,19 @@ describe('Session', () => { type: 'user/message', seq: 0, time: 1, + data: { content: [{ type: 'text', text: 'source' }], source: { kind: 'user' } }, + surfaceOp: 'append', + }, { + type: 'user/message', + seq: 1, + time: 2, data: { content: [{ type: 'text', text: 'hello' }], source: { kind: 'user' } }, surfaceOp, + sourceEventSeqs: [0], }] as unknown as SessionEvent[] const session = new Session(SessionId('seed-unstable-metadata'), seed) - const event = session.events[0]! + const event = session.events[1]! if (event.type !== 'user/message') throw new Error('test fixture must remain a user/message') expect(reads).toBe(1) @@ -326,13 +333,20 @@ describe('Session', () => { type: 'user/message', seq: 0, time: 1, + data: { content: [{ type: 'text', text: 'source' }], source: { kind: 'user' } }, + surfaceOp: 'append', + }, { + type: 'user/message', + seq: 1, + time: 2, data: { content: [{ type: 'text', text: 'hello' }], source: { kind: 'user' } }, surfaceOp: { op: 'replace', start: 0, end: 0 }, + sourceEventSeqs: [0], }] as unknown as SessionEvent[] try { expect(() => new Session(SessionId('seed-non-error-metadata-failure'), seed)) - .toThrow(`invalid seed event at index 0: ${expected}`) + .toThrow(`invalid seed event at index 1: ${expected}`) } finally { hasOwn.mockRestore() } @@ -418,6 +432,11 @@ describe('Session', () => { it('reads a nested append-metadata getter once and stores its first JSON value', () => { const session = new Session(SessionId('append-unstable-metadata')) + const source = session.append( + 'user/message', + { content: [{ type: 'text', text: 'source' }], source: { kind: 'user' } }, + { surfaceOp: 'append' }, + ) let reads = 0 const surfaceOp = Object.defineProperty({ op: 'replace', end: 0 }, 'start', { enumerable: true, @@ -430,12 +449,12 @@ describe('Session', () => { const event = session.append( 'user/message', { content: [{ type: 'text', text: 'hello' }], source: { kind: 'user' } }, - { surfaceOp } as never, + { surfaceOp, sourceEventSeqs: [0] } as never, ) expect(reads).toBe(1) expect(event.surfaceOp).toEqual({ op: 'replace', start: 0, end: 0 }) - expect(session.events).toEqual([event]) + expect(session.events).toEqual([source, event]) }) it('rejects invalid plain surface metadata shapes at append', () => { diff --git a/packages/core/session/tests/surface.spec.ts b/packages/core/session/tests/surface.spec.ts index fa7b1ff3ce..e1f948f2b1 100644 --- a/packages/core/session/tests/surface.spec.ts +++ b/packages/core/session/tests/surface.spec.ts @@ -6,7 +6,6 @@ import { foldSurface, isSurfaceEligibleType, isSurfaceEvent, - validateSurfaceMetadata, } from '@deepseek-ai/dsh-session' import { CallId } from '@deepseek-ai/dsh-llm' @@ -27,53 +26,52 @@ function provenanceEvent(seq: number, sourceEventSeqs: unknown): SessionEvent { time: seq, data: { content: [], source: { kind: 'user' } }, surfaceOp: 'append', - sourceEventSeqs, + ...sourceEventSeqs === undefined ? {} : { sourceEventSeqs }, } as unknown as SessionEvent } -describe('validateSurfaceMetadata', () => { +describe('foldSurface provenance', () => { it('accepts absent or valid provenance and complete replacement coverage', () => { - expect(validateSurfaceMetadata(provenanceEvent(0, undefined), new Set())) - .toBeUndefined() - expect(validateSurfaceMetadata(provenanceEvent(2, [0, 1]), new Set([0, 1]), [1])) - .toBeUndefined() + const events = [ + provenanceEvent(0, undefined), + provenanceEvent(1, undefined), + { + ...provenanceEvent(2, [0, 1]), + surfaceOp: { op: 'replace', start: 0, end: 1 }, + }, + ] as SessionEvent[] + expect(() => foldSurface(events)).not.toThrow() }) it('rejects provenance on a non-surface event', () => { const event = { type: 'turn/start', - seq: 1, + seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }, sourceEventSeqs: [0], } as unknown as SessionEvent - expect(validateSurfaceMetadata(event, new Set([0]))) - .toEqual({ - kind: 'provenance', - message: 'turn/start cannot carry sourceEventSeqs (non-surface event)', - }) + expect(() => foldSurface([event])).toThrow(/cannot carry sourceEventSeqs/) }) it.each([ - ['a non-array', 1, 'invalid', new Set([0]), [], /must be an array/], - ['an empty array', 1, [], new Set([0]), [], /must not be empty/], - ['duplicates', 1, [0, 0], new Set([0]), [], /must not contain duplicates/], - ['a non-number', 1, ['0'], new Set([0]), [], /non-negative safe integers/], - ['a fractional number', 1, [0.5], new Set([0]), [], /non-negative safe integers/], - ['a negative number', 1, [-1], new Set([0]), [], /non-negative safe integers/], - ['a self reference', 1, [1], new Set([0]), [], /must reference earlier events/], - ['an unknown earlier seq', 2, [1], new Set([0]), [], /references unknown seq 1/], - ['incomplete replacement coverage', 2, [0], new Set([0, 1]), [0, 1], /missing 1/], + ['a non-array', [{ ...provenanceEvent(0, undefined), sourceEventSeqs: 'invalid' }], /must be an array/], + ['an empty array', [provenanceEvent(0, [])], /must not be empty/], + ['duplicates', [provenanceEvent(0, undefined), provenanceEvent(1, [0, 0])], /must not contain duplicates/], + ['a non-number', [{ ...provenanceEvent(0, undefined), sourceEventSeqs: ['0'] }], /non-negative safe integers/], + ['a fractional number', [provenanceEvent(0, [0.5])], /non-negative safe integers/], + ['a negative number', [provenanceEvent(0, [-1])], /non-negative safe integers/], + ['a self reference', [provenanceEvent(0, [0])], /must reference earlier events/], + ['an unknown earlier seq', [provenanceEvent(0, undefined), provenanceEvent(2, [1])], /references unknown seq 1/], + ['incomplete replacement coverage', [ + provenanceEvent(0, undefined), + provenanceEvent(1, undefined), + { ...provenanceEvent(2, [0]), surfaceOp: { op: 'replace', start: 0, end: 1 } }, + ], /missing 1/], ] as const)( - 'returns the first violation for %s', - (_name, seq, sources, knownSeqs, shadowedSeqs, expected) => { - const violation = validateSurfaceMetadata( - provenanceEvent(seq, sources), - knownSeqs, - shadowedSeqs, - ) - expect(violation?.kind).toBe('provenance') - expect(violation?.message).toMatch(expected) + 'rejects %s', + (_name, events, expected) => { + expect(() => foldSurface(events as unknown as SessionEvent[])).toThrow(expected) }, ) }) @@ -101,7 +99,7 @@ describe('SurfaceManager', () => { it('does not retain fold-only replacement history in incremental state', () => { const s = new Session(SessionId('incremental-state')) s.append('user/message', { content: [{ type: 'text', text: 'a' }], source: { kind: 'user' } }, { surfaceOp: 'append' }) - s.append('assistant/message', { turn: 1, step: 1, content: [{ type: 'text', text: 'b' }] }, { surfaceOp: { op: 'replace', start: 0, end: 0 } }) + s.append('assistant/message', { turn: 1, step: 1, content: [{ type: 'text', text: 'b' }] }, { surfaceOp: { op: 'replace', start: 0, end: 0 }, sourceEventSeqs: [0] }) expect(s.surface.nodes).toEqual([{ seq: 1, prev: null, next: null }]) const manager = s.surface as unknown as { _state: object } @@ -112,12 +110,29 @@ describe('SurfaceManager', () => { }) it('foldSurface reports the same invalid replacement failures as the incremental manager', () => { - const s = new Session(SessionId('shared-fold-invalid')) - s.append('user/message', { content: [{ type: 'text', text: 'a' }], source: { kind: 'user' } }, { surfaceOp: 'append' }) - s.append('assistant/message', { turn: 1, step: 1, content: [] }, { surfaceOp: { op: 'replace', start: 42, end: 0 }, sourceEventSeqs: [0] }) + const events = [ + provenanceEvent(0, undefined), + { ...provenanceEvent(1, [0]), surfaceOp: { op: 'replace', start: 42, end: 0 } }, + ] as SessionEvent[] - expect(() => foldSurface(s.events)).toThrow(/start seq 42 not found/) - expect(() => s.surface.nodes).toThrow(/start seq 42 not found/) + expect(() => foldSurface(events)).toThrow(/start seq 42 not found/) + expect(() => new Session(SessionId('shared-fold-invalid'), events)) + .toThrow(/start seq 42 not found/) + }) + + it('leaves incremental state unchanged when candidate validation fails', () => { + const s = new Session(SessionId('atomic-validation')) + s.append('user/message', { content: [{ type: 'text', text: 'a' }], source: { kind: 'user' } }, { surfaceOp: 'append' }) + + expect(() => s.append( + 'assistant/message', + { turn: 1, step: 1, content: [{ type: 'text', text: 'invalid' }] }, + { surfaceOp: { op: 'replace', start: 0, end: 0 } }, + )).toThrow(/missing 0/) + + expect(s.events).toHaveLength(1) + s.append('user/message', { content: [{ type: 'text', text: 'b' }], source: { kind: 'user' } }, { surfaceOp: 'append' }) + expect(s.surface.nodes.map(node => node.seq)).toEqual([0, 1]) }) it('foldSurface rejects a surface-eligible event without its mandatory marker', () => { @@ -250,21 +265,19 @@ describe('SurfaceManager', () => { it('throws when replace start is not found', () => { const s = new Session(SessionId('bad-start')) s.append('user/message', { content: [{ type: 'text', text: 'a' }], source: { kind: 'user' } }, { surfaceOp: 'append' }) // seq 0 - s.append('assistant/message', + expect(() => s.append('assistant/message', { turn: 1, step: 1, content: [{ type: 'text', text: 'y' }] }, - { surfaceOp: { op: 'replace', start: 5, end: 0 }, sourceEventSeqs: [5, 0] }, - ) - expect(() => s.surface.nodes).toThrow(/surface replace: start seq 5 not found/) + { surfaceOp: { op: 'replace', start: 5, end: 0 }, sourceEventSeqs: [0] }, + )).toThrow(/surface replace: start seq 5 not found/) }) it('throws when replace end is not found', () => { const s = new Session(SessionId('bad-end')) s.append('user/message', { content: [{ type: 'text', text: 'a' }], source: { kind: 'user' } }, { surfaceOp: 'append' }) // seq 0 - s.append('assistant/message', + expect(() => s.append('assistant/message', { turn: 1, step: 1, content: [{ type: 'text', text: 'y' }] }, { surfaceOp: { op: 'replace', start: 0, end: 99 }, sourceEventSeqs: [0] }, - ) - expect(() => s.surface.nodes).toThrow(/surface replace: end seq 99 not found/) + )).toThrow(/surface replace: end seq 99 not found/) }) it('throws when start is after end', () => { @@ -272,22 +285,22 @@ describe('SurfaceManager', () => { s.append('user/message', { content: [{ type: 'text', text: 'a' }], source: { kind: 'user' } }, { surfaceOp: 'append' }) // seq 0 s.append('user/message', { content: [{ type: 'text', text: 'b' }], source: { kind: 'user' } }, { surfaceOp: 'append' }) // seq 1 // start=1, end=0 would be reversed order. - s.append('assistant/message', + expect(() => s.append('assistant/message', { turn: 1, step: 1, content: [{ type: 'text', text: 'y' }] }, { surfaceOp: { op: 'replace', start: 1, end: 0 }, sourceEventSeqs: [1, 0] }, - ) - expect(() => s.surface.nodes).toThrow(/start seq 1.*after end seq 0/) + )).toThrow(/start seq 1.*after end seq 0/) }) it('sourceEventSeqs is snapshot so caller mutation does not affect logged event', () => { const s = new Session(SessionId('immutable')) - const sources = [10, 20] + s.append('user/message', { content: [{ type: 'text', text: 'source' }], source: { kind: 'user' } }, { surfaceOp: 'append' }) + const sources = [0] s.append('assistant/message', { turn: 1, step: 1, content: [{ type: 'text', text: 'h' }] }, { surfaceOp: 'append', sourceEventSeqs: sources }) // Mutate caller's array after append. - sources.push(30) + sources.push(1) sources[0] = 99 - const logged = s.events[0]! as SurfaceEvent - expect(logged.sourceEventSeqs).toEqual([10, 20]) + const logged = s.events[1]! as SurfaceEvent + expect(logged.sourceEventSeqs).toEqual([0]) }) it('replace starting at non-head position links to previous node correctly', () => { @@ -369,15 +382,17 @@ describe('deriveMessages with surface', () => { describe('Session.append surface opts', () => { it('records sourceEventSeqs and surfaceOp on the event', () => { const s = new Session(SessionId('opts')) + s.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) + s.append('step/start', { turn: 1, step: 1 }) const event = s.append('assistant/message', { turn: 1, step: 1, content: [{ type: 'text', text: 'h' }] }, - { surfaceOp: 'append', sourceEventSeqs: [3, 5, 7] }, + { surfaceOp: 'append', sourceEventSeqs: [0, 1] }, ) - expect(event.sourceEventSeqs).toEqual([3, 5, 7]) + expect(event.sourceEventSeqs).toEqual([0, 1]) expect(event.surfaceOp).toBe('append') // The logged event matches the returned event. - expect((s.events[0]! as SurfaceEvent).sourceEventSeqs).toEqual([3, 5, 7]) - expect((s.events[0]! as SurfaceEvent).surfaceOp).toBe('append') + expect((s.events[2]! as SurfaceEvent).sourceEventSeqs).toEqual([0, 1]) + expect((s.events[2]! as SurfaceEvent).surfaceOp).toBe('append') }) it('deriveMessages skips a surface node that derives to null (empty assistant/message)', () => { diff --git a/packages/core/session/tests/tool-pairing.spec.ts b/packages/core/session/tests/tool-pairing.spec.ts index 307b0d8658..cc2acd1a2c 100644 --- a/packages/core/session/tests/tool-pairing.spec.ts +++ b/packages/core/session/tests/tool-pairing.spec.ts @@ -261,10 +261,11 @@ describe('isToolPairingBalanced — CBR-001: a head checkpoint left by a replace // summary user/message — appended now, so it carries a high log seq. const u1 = seqOf(s, 'user/message') const result = s.events.find(e => e.type === 'tool/result')!.seq + const shadowedSeqs = s.surface.nodes.map(node => node.seq) s.append('user/message', { content: [{ type: 'text', text: 'CHECKPOINT' }], source: { kind: 'plugin', plugin: 'compact' }, - }, { surfaceOp: { op: 'replace', start: u1, end: result } }) + }, { surfaceOp: { op: 'replace', start: u1, end: result }, sourceEventSeqs: shadowedSeqs }) // The step's own assistant/message lands AFTER the checkpoint in the log, // still inside the open step. s.append('assistant/message', { turn: 2, step: 1, content: [{ type: 'text', text: 'a2' }] }, SURFACE) diff --git a/packages/session-query/session-query/README.md b/packages/session-query/session-query/README.md index 2dcdec085e..4f9f8122b4 100644 --- a/packages/session-query/session-query/README.md +++ b/packages/session-query/session-query/README.md @@ -14,9 +14,9 @@ This is trusted context-wide infrastructure. It performs no caller authorization Persistence is optional and may mount or unmount dynamically. Cross-corpus listing and lineage tracing fail with `SESSION_QUERY_PERSISTENCE_FAILED` while mounted persistence is unreadable. An event read or trace targeting a known live session does not consult persistence, so durable backend health cannot make current in-memory history unreadable. Persisted event operations list before loading and reject a metadata mismatch rather than combining inconsistent observations. -`traceEvent()` validates the whole loaded log with `dsh-session`'s shared surface-metadata checker before returning relationships: surface markers obey event-type eligibility, provenance arrays are nonempty and duplicate-free, references name known earlier events, and each positional replacement names every surface node it removed. Surface-marker and positional-fold violations fail with `SESSION_QUERY_INVALID_SURFACE`; provenance violations use `SESSION_QUERY_INVALID_PROVENANCE`. `listEvents()` only needs surface classification and deliberately does not enforce the trace-specific provenance contract. +`listEvents()` and `traceEvent()` run the same one-pass `dsh-session` surface fold. A loaded log is valid only when surface markers obey event-type eligibility, provenance arrays are nonempty and duplicate-free, references name known earlier events, and each positional replacement names and cites every surface node it removes; every violation fails with `SESSION_QUERY_INVALID_SURFACE`. -`SessionQueryError.code` is a closed union: `SESSION_QUERY_EVENT_NOT_FOUND`, `SESSION_QUERY_INVALID_CONFIG`, `SESSION_QUERY_INVALID_LINEAGE`, `SESSION_QUERY_INVALID_PROVENANCE`, `SESSION_QUERY_INVALID_SURFACE`, `SESSION_QUERY_INVALID_WINDOW`, `SESSION_QUERY_PERSISTENCE_FAILED`, `SESSION_QUERY_SESSION_NOT_FOUND`, and `SESSION_QUERY_SOURCE_CONFLICT`. +`SessionQueryError.code` is a closed union: `SESSION_QUERY_EVENT_NOT_FOUND`, `SESSION_QUERY_INVALID_CONFIG`, `SESSION_QUERY_INVALID_LINEAGE`, `SESSION_QUERY_INVALID_SURFACE`, `SESSION_QUERY_INVALID_WINDOW`, `SESSION_QUERY_PERSISTENCE_FAILED`, `SESSION_QUERY_SESSION_NOT_FOUND`, and `SESSION_QUERY_SOURCE_CONFLICT`. ## Configuration diff --git a/packages/session-query/session-query/src/config.ts b/packages/session-query/session-query/src/config.ts index a6d0ab10fa..4a15366c2c 100644 --- a/packages/session-query/session-query/src/config.ts +++ b/packages/session-query/session-query/src/config.ts @@ -16,7 +16,6 @@ export type SessionQueryErrorCode = | 'SESSION_QUERY_EVENT_NOT_FOUND' | 'SESSION_QUERY_INVALID_CONFIG' | 'SESSION_QUERY_INVALID_LINEAGE' - | 'SESSION_QUERY_INVALID_PROVENANCE' | 'SESSION_QUERY_INVALID_SURFACE' | 'SESSION_QUERY_INVALID_WINDOW' | 'SESSION_QUERY_PERSISTENCE_FAILED' diff --git a/packages/session-query/session-query/src/tracing.ts b/packages/session-query/session-query/src/tracing.ts index 09d3ed65f6..c7b5143b67 100644 --- a/packages/session-query/session-query/src/tracing.ts +++ b/packages/session-query/session-query/src/tracing.ts @@ -1,7 +1,7 @@ /** One-shot session-lineage and event-relationship tracing helpers. */ -import { foldSurface, validateSurfaceMetadata } from '@deepseek-ai/dsh-session' -import type { SessionEvent, SessionId } from '@deepseek-ai/dsh-session' +import { foldSurface } from '@deepseek-ai/dsh-session' +import type { SessionEvent, SessionId, SurfaceEventType } from '@deepseek-ai/dsh-session' import { SessionQueryError } from './config.ts' import type { SessionEventRecord, @@ -51,21 +51,6 @@ export function traceEventLog( } const analysis = analyzeEventLog(sessionId, events) - const knownSeqs = new Set() - for (const event of events) { - const violation = validateSurfaceMetadata( - event, - knownSeqs, - analysis.replacedEventSeqs.get(event.seq), - ) - if (violation !== undefined) { - throw new SessionQueryError( - `invalid session provenance: ${violation.message}`, - 'SESSION_QUERY_INVALID_PROVENANCE', - ) - } - knownSeqs.add(event.seq) - } const replacementChain: number[] = [] let replacement = analysis.replacedBy.get(seq) @@ -143,7 +128,9 @@ export function traceLineage( children.push(record) childrenByParent.set(parent, children) } - for (const children of childrenByParent.values()) children.sort(compareSessionsAscending) + for (const children of childrenByParent.values()) { + children.sort((a, b) => a.header.createdAt - b.header.createdAt || a.header.id.localeCompare(b.header.id)) + } const descendants = buildDescendants(childrenByParent, sessionId) const common = { @@ -203,13 +190,8 @@ function analyzeEventLog( } } -function rawEventSources(event: SessionEvent): unknown { - return (event as SessionEvent & { sourceEventSeqs?: unknown }).sourceEventSeqs -} - function eventSources(event: SessionEvent): number[] { - const sources = rawEventSources(event) - return Array.isArray(sources) ? sources as number[] : [] + return (event as SessionEvent).sourceEventSeqs ?? [] } function buildDescendants( @@ -238,10 +220,6 @@ function buildDescendants( return descendants } -function compareSessionsAscending(a: SessionRecord, b: SessionRecord): number { - return a.header.createdAt - b.header.createdAt || a.header.id.localeCompare(b.header.id) -} - function cloneRecord(record: SessionRecord): SessionRecord { return { ...record, header: structuredClone(record.header) } } diff --git a/packages/session-query/session-query/tests/session-query.spec.ts b/packages/session-query/session-query/tests/session-query.spec.ts index 3b50feee45..bb7d01ec73 100644 --- a/packages/session-query/session-query/tests/session-query.spec.ts +++ b/packages/session-query/session-query/tests/session-query.spec.ts @@ -110,7 +110,7 @@ describe('session-query exact reads', () => { session.append( 'assistant/message', { turn: 1, step: 1, content: [{ type: 'text', text: 'replacement' }] }, - { surfaceOp: { op: 'replace', start: first.seq, end: first.seq } }, + { surfaceOp: { op: 'replace', start: first.seq, end: first.seq }, sourceEventSeqs: [first.seq] }, ) expect((await ctx.sessionQuery.listEvents(session.id)).map(record => record.surface)) @@ -227,11 +227,13 @@ describe('session-query exact reads', () => { it('turns malformed surfaces and direct invalid config into typed errors', async () => { const ctx = await liveContext() const session = ctx.sessions.create(SessionId('bad-surface')) - session.append( - 'assistant/message', - { turn: 1, step: 1, content: [] }, - { surfaceOp: { op: 'replace', start: 9, end: 9 } }, - ) + ;(session as unknown as { log: SessionEvent[] }).log.push({ + type: 'assistant/message', + seq: 0, + time: 1, + data: { turn: 1, step: 1, content: [] }, + surfaceOp: { op: 'replace', start: 9, end: 9 }, + }) await expect(ctx.sessionQuery.listEvents(session.id)) .rejects.toThrow(expectCode('SESSION_QUERY_INVALID_SURFACE')) diff --git a/packages/session-query/session-query/tests/tracing.spec.ts b/packages/session-query/session-query/tests/tracing.spec.ts index 0a8b8e6da1..efc32d2216 100644 --- a/packages/session-query/session-query/tests/tracing.spec.ts +++ b/packages/session-query/session-query/tests/tracing.spec.ts @@ -379,7 +379,7 @@ describe('session event tracing', () => { appendEvent(1), { ...appendEvent(2, [0]), surfaceOp: { op: 'replace', start: 1, end: 1 } }, ]], - ] as const)('rejects invalid whole-log provenance: %s', async (_name, rawEvents) => { + ] as const)('rejects an invalid surface log: %s', async (_name, rawEvents) => { const durable = header('invalid-provenance') const events = structuredClone(rawEvents) as unknown as SessionEvent[] TracePersistence.reset([{ meta: durable, events }]) @@ -387,7 +387,7 @@ describe('session event tracing', () => { await ctx.plugin(TracePersistence) await expect(ctx.sessionQuery.traceEvent({ sessionId: durable.id, seq: 0 })) - .rejects.toThrow(expectCode('SESSION_QUERY_INVALID_PROVENANCE')) + .rejects.toThrow(expectCode('SESSION_QUERY_INVALID_SURFACE')) }) it('rejects surfaceOp on a non-surface event as an invalid surface', async () => { @@ -407,15 +407,13 @@ describe('session event tracing', () => { .rejects.toThrow(expectCode('SESSION_QUERY_INVALID_SURFACE')) }) - it('keeps listEvents tolerant of malformed provenance alone', async () => { + it('applies the same surface contract to listEvents', async () => { const durable = header('list-regression') TracePersistence.reset([{ meta: durable, events: [appendEvent(0), appendEvent(1, [0, 0])] }]) const ctx = await queryContext() await ctx.plugin(TracePersistence) - await expect(ctx.sessionQuery.listEvents(durable.id)).resolves.toMatchObject([ - { seq: 0, surface: 'current' }, - { seq: 1, surface: 'current' }, - ]) + await expect(ctx.sessionQuery.listEvents(durable.id)) + .rejects.toThrow(expectCode('SESSION_QUERY_INVALID_SURFACE')) }) }) diff --git a/packages/support/invariants/README.md b/packages/support/invariants/README.md index 2fe2db1f6c..562df0bf3a 100644 --- a/packages/support/invariants/README.md +++ b/packages/support/invariants/README.md @@ -4,7 +4,7 @@ Dev-mode event-contract assertions. This pure-listener plugin checks relationshi **Off in production.** Enable it in tests and the demos, where a contract violation should fail loudly. It costs nothing when not registered, and doubles as executable documentation of the event taxonomy — the assertions *are* the contract. -Session itself owns immutable log storage in every composition: it takes one lossless JSON snapshot of each accepted event, deep-freezes that record, and exposes the log through immutable array snapshots. The invariants plugin checks the cross-record and cross-seam rules that storage immutability cannot express. +Session itself owns immutable, surface-valid log storage in every composition: it takes one lossless JSON snapshot of each candidate, validates the complete surface transition, deep-freezes the accepted record, and exposes the log through immutable array snapshots. The invariants plugin checks the remaining cross-record and cross-seam rules that Session does not own. Session-log assertions run during Cordis `internal/dispatch`, while `Session.append()` is resolving the `session/event` callback snapshot but before it pushes the candidate into the log. A valid transition is staged by exact event identity and applied to the live trace only when that same committed event reaches the plugin's contained post-commit listener. A later internal dispatch check can therefore veto without advancing either the log or the invariant trace, while ordinary `session/event` observer failures remain observe-only. @@ -28,7 +28,6 @@ await ctx.plugin(Invariants) Session log (per session): - **`seq` strictly increases** — the spine of replay equivalence. -- **surface metadata is valid** — `surfaceOp` and `sourceEventSeqs` use the shared `dsh-session` checker for type eligibility, structural shape, nonempty unique earlier references, and complete replacement coverage. - **turns pair and nest** — `turn/start` opens a turn, `turn/end` closes the matching one; no overlapping turns. - **steps nest in turns** — `step/start` opens a step in the open turn; `step/end` closes the matching step. - **chunks belong to an open step** — `step/start` precedes its `assistant/chunk`s. diff --git a/packages/support/invariants/src/index.ts b/packages/support/invariants/src/index.ts index b2fa15080f..db25c42b89 100644 --- a/packages/support/invariants/src/index.ts +++ b/packages/support/invariants/src/index.ts @@ -6,13 +6,12 @@ * `session/event`, `agent/status`, and the scoped dispatch and request seams. * It is **off in production**: enable it in tests and demos, where a contract * violation should be a loud failure rather than a subtle one. It doubles as - * executable documentation of the event taxonomy: these assertions and the - * shared session validators they invoke are the contract. + * executable documentation of the relational event taxonomy. * - * Session owns immutable log storage: it snapshots and deep-freezes every - * accepted event at the source. This plugin checks relationships that one - * event's types and immutability cannot express, including turn/step nesting, - * scoped dispatch, status transitions, and request reconstructability. + * Session owns immutable, surface-valid log storage: it validates, snapshots, + * and deep-freezes every accepted event at the source. This plugin checks the + * remaining relationships that acceptance cannot express, including turn/step + * nesting, scoped dispatch, status transitions, and request reconstructability. * * @module @deepseek-ai/dsh-invariants */ @@ -26,9 +25,8 @@ import { Session, SessionId, foldRequestHeader, - validateSurfaceMetadata, } from '@deepseek-ai/dsh-session' -import type { SessionEvent, SurfaceEventType } from '@deepseek-ai/dsh-session' +import type { SessionEvent } from '@deepseek-ai/dsh-session' export const name = 'invariants' export const inject = ['sessions'] @@ -62,15 +60,6 @@ interface SessionTrace { * `step/end` — a result must arrive in the same step as its call. */ pendingCalls: Set - /** Every seq seen so far — validates `sourceEventSeqs` references. */ - knownSeqs: Set - /** - * The seqs currently on the surface linked list, in linked-list order - * (head to tail). A replace reorders this relative to seq order (the new - * node takes the replaced range's position), so range validation is - * positional, not by seq comparison. - */ - surface: number[] } /** One accepted event's deferred mutation of a live session trace. */ @@ -82,12 +71,6 @@ interface SessionTraceTransition { | { kind: 'none' } | { kind: 'add' | 'delete'; callId: CallId } | { kind: 'clear' } - /** The event's mutation of the derived surface order. */ - surface: - | { kind: 'none' | 'append' } - | { kind: 'replace'; start: number; count: number } - /** The committed event sequence to add to the known-sequence set. */ - seq: number } /** Event payload prefix for scoped seams whose first argument names its agent. */ @@ -122,50 +105,6 @@ function validateEvent(trace: SessionTrace, event: SessionEvent): SessionTraceTr let nextTurn = trace.nextTurn let nextStep = trace.nextStep let pendingCalls: SessionTraceTransition['pendingCalls'] = { kind: 'none' } - let surface: SessionTraceTransition['surface'] = { kind: 'none' } - - // --- Surface invariants --- - // Cast to surface-eligible event type so we can access surfaceOp and - // sourceEventSeqs (optional on SessionEvent, mandatory on SurfaceEvent). - // SurfaceEvent's mandatory surfaceOp is too strict here — we need to - // CHECK whether surface metadata is present, not assume it. - const se = event as SessionEvent - const metadataViolation = validateSurfaceMetadata(event) - if (metadataViolation !== undefined) throw new InvariantError(metadataViolation.message) - - // Fold this event into the tracked surface linked list, validating the - // replace contract as we go. `append` adds a tail node; `replace` shadows a - // positional range — every shadowed node must appear in sourceEventSeqs. - let shadowed: number[] | undefined - if (se.surfaceOp !== undefined) { - if (se.surfaceOp === 'append') { - surface = { kind: 'append' } - } else { - const { start, end } = se.surfaceOp - const startIdx = trace.surface.indexOf(start) - if (startIdx === -1) { - throw new InvariantError(`surface replace: start seq ${start} is not on the surface`) - } - const endIdx = trace.surface.indexOf(end) - if (endIdx === -1) { - throw new InvariantError(`surface replace: end seq ${end} is not on the surface`) - } - if (startIdx > endIdx) { - throw new InvariantError(`surface replace: start seq ${start} (pos ${startIdx}) is after end seq ${end} (pos ${endIdx}) on the surface`) - } - shadowed = trace.surface.slice(startIdx, endIdx + 1) - surface = { kind: 'replace', start: startIdx, count: shadowed.length } - } - } - - const provenanceViolation = validateSurfaceMetadata( - event, - trace.knownSeqs, - shadowed, - ) - if (provenanceViolation !== undefined) { - throw new InvariantError(provenanceViolation.message) - } // Boundary/step-scoped events have explicit cases; every OTHER event type — // including plugin-added (merge-extensible) SessionEventMap keys — is caught @@ -265,8 +204,6 @@ function validateEvent(trace: SessionTrace, event: SessionEvent): SessionTraceTr return { scalars: { lastSeq: event.seq, openTurn, openStep, nextTurn, nextStep }, pendingCalls, - surface, - seq: event.seq, } } @@ -289,20 +226,6 @@ function applyTransition(trace: SessionTrace, transition: SessionTraceTransition default: assertNever(transition.pendingCalls, 'session trace pending-call transition') } - switch (transition.surface.kind) { - case 'none': - break - case 'append': - trace.surface.push(transition.seq) - break - case 'replace': - trace.surface.splice(transition.surface.start, transition.surface.count, transition.seq) - break - /* v8 ignore next -- validateEvent produces this closed transition union */ - default: - assertNever(transition.surface, 'session trace surface transition') - } - trace.knownSeqs.add(transition.seq) } /** Validate and apply one event while rebuilding an already-committed log. */ @@ -351,8 +274,6 @@ export function apply(ctx: Context): void { nextTurn: 1, nextStep: 1, pendingCalls: new Set(), - knownSeqs: new Set(), - surface: [], }) /** Build (or rebuild) a session's trace by replaying its whole log. */ diff --git a/packages/support/invariants/tests/invariants.spec.ts b/packages/support/invariants/tests/invariants.spec.ts index a1e824524b..d5f43d1568 100644 --- a/packages/support/invariants/tests/invariants.spec.ts +++ b/packages/support/invariants/tests/invariants.spec.ts @@ -466,7 +466,7 @@ describe('HMR safety', () => { }) }) -describe('surface invariants', () => { +describe('surface contract under the invariants composition', () => { it('accepts well-formed surface metadata', async () => { const { ctx } = await setup() const session = ctx.sessions.create() @@ -495,7 +495,7 @@ describe('surface invariants', () => { session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) expect(() => { session.append('assistant/message', { turn: 1, step: 1, content: [] }, { surfaceOp: 'append', sourceEventSeqs: [] }) - }).toThrow(InvariantError) + }).toThrow(/must not be empty/) }) it('rejects duplicate sourceEventSeqs', async () => { @@ -542,15 +542,15 @@ describe('surface invariants', () => { it('rejects sourceEventSeqs referencing unknown seq (gap in event log)', async () => { // The unknown-seq check fires when a ref passes the "earlier" test but is - // not in knownSeqs — only possible with a gap in seqs. We create a gap by + // not in the folded log — only possible with a gap in seqs. We create a gap by // directly manipulating the private log array to skip a seq. const { ctx } = await setup() const session = ctx.sessions.create() session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) session.append('step/start', { turn: 1, step: 1 }) // Push a fake event at seq 3 into the internal log, creating a gap at seq 2. - // The invariants plugin replays session.events on every append, so it sees - // this gap during trace reconstruction. + // The canonical surface validator folds the committed delta before checking + // the next append, so it sees this gap. ;(session as unknown as { log: unknown[] }).log.push({ type: 'assistant/chunk', seq: 3, @@ -575,7 +575,7 @@ describe('surface invariants', () => { // Reversed range: start seq 3 is at a later surface position than end seq 2. expect(() => { session.append('assistant/message', { turn: 1, step: 1, content: [] }, { surfaceOp: { op: 'replace', start: 3, end: 2 }, sourceEventSeqs: [2, 3] }) - }).toThrow(/is after end seq 2 .* on the surface/) + }).toThrow(/is after end seq 2/) }) it('rejects a replace whose sourceEventSeqs omits a shadowed surface node', async () => { @@ -612,7 +612,7 @@ describe('surface invariants', () => { // seq 1 (step/start) is a real earlier event but never entered the surface. expect(() => { session.append('assistant/message', { turn: 1, step: 1, content: [] }, { surfaceOp: { op: 'replace', start: 1, end: 2 }, sourceEventSeqs: [1, 2] }) - }).toThrow(/start seq 1 is not on the surface/) + }).toThrow(/start seq 1 not found in surface/) }) it('rejects a replace naming an end seq that is not on the surface', async () => { @@ -624,7 +624,7 @@ describe('surface invariants', () => { // start (2) is on the surface but end (99) never entered it. expect(() => { session.append('assistant/message', { turn: 1, step: 1, content: [] }, { surfaceOp: { op: 'replace', start: 2, end: 99 }, sourceEventSeqs: [2] }) - }).toThrow(/end seq 99 is not on the surface/) + }).toThrow(/end seq 99 not found in surface/) }) it('rejects a replace whose range is reversed in surface position after a prior replace reordered it', async () => { @@ -641,7 +641,7 @@ describe('surface invariants', () => { // reversed positionally (3 is at pos 1, 4 is at pos 0). expect(() => { session.append('assistant/message', { turn: 1, step: 1, content: [] }, { surfaceOp: { op: 'replace', start: 3, end: 4 }, sourceEventSeqs: [3, 4] }) // seq 5 - }).toThrow(/is after end seq 4 .* on the surface/) + }).toThrow(/is after end seq 4/) }) it('accepts a replace whose start seq exceeds its end seq when the surface position order is valid', async () => { @@ -685,25 +685,6 @@ describe('surface invariants', () => { expect(() => ctx.sessions.create(undefined, { seed: badSeed })).toThrow(/must include every shadowed surface node; missing 3/) }) - it('rejects sourceEventSeqs on a non-surface event', async () => { - const { ctx } = await setup() - const session = ctx.sessions.create() - session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) - // Session rejects this at its own acceptance boundary. Emit a hand-built - // record to cover the listener's defensive check for alternate producers. - const event = { type: 'turn/end', seq: 1, time: 1, data: { turn: 1, reason: { kind: 'completed' } }, sourceEventSeqs: [0] } - expect(() => { ctx.emit(scopeTarget(session, undefined), 'session/event', session, event as never) }) - .toThrow(/cannot carry sourceEventSeqs/) - }) - - it('rejects surfaceOp on a non-surface event', async () => { - const { ctx } = await setup() - const session = ctx.sessions.create() - session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) - const event = { type: 'turn/end', seq: 1, time: 1, data: { turn: 1, reason: { kind: 'completed' } }, surfaceOp: 'append' } - expect(() => { ctx.emit(scopeTarget(session, undefined), 'session/event', session, event as never) }) - .toThrow(/cannot carry surfaceOp/) - }) }) describe('request-reconstruction cross-check (llm/stream)', () => {