Merge refreshed schema DSL into canonical tool outputs

This commit is contained in:
Tianyi Cui
2026-07-22 22:46:10 +08:00
32 changed files with 1208 additions and 61 deletions

View File

@@ -48,6 +48,10 @@ Plain class (not a Cordis Service). Create via `ctx.sessions.create()`.
Durable values need one accepted representation, not a check followed by a second read. `isJsonValue(value)` is the boolean predicate; `snapshotJsonValue(value)` iteratively validates and copies a plain value in one pass, returning `undefined` for invalid input and propagating a throwing getter. The snapshot helper accepts finite JSON numbers except `-0` (JSON rewrites it to `0`), dense ordinary arrays, and plain or null-prototype objects; it rejects cycles, unsupported scalars, and exotic prototypes before normalization without imposing a call-stack depth limit.
### Chunk-row storage codec (`chunk-rows.ts`)
Providers stream token-sized deltas, so a raw log stores hundreds of `assistant/chunk` lines whose JSON envelopes dwarf their payloads. `packChunkRuns(events)` packs each run of ≥3 consecutive same-block delta chunks into one storage row — `text-chunks`, `reasoning-chunks`, or `tool-call-chunks` (bare slash-less tags: storage vocabulary, not `SessionEventMap` members) — and `decodeStorageRecord(value)` expands a parsed line back into its exact events (`seq0`/`time0` + per-member `dt` gaps reconstruct every `seq`/`time`). The encoder whitelists exact shapes and stores anything unrecognized verbatim; the decoder validates row-tagged values and throws on malformation. Owned here so the JSONL backend and the fixture readers (`dsh-llm-replay`, `dsh-acp-snapshot`) share one codec; the write-side switch is the backend's `packChunks` config.
### Surface types
- `SurfaceOp` — how an event entered the ordered surface: `'append'` (normal tail append) or `{ op: 'replace', start, end }` (replace entries from `start` through `end` inclusive — both must be valid surface seqs; `start === end` replaces one entry). Used by compaction to shadow old events without deleting them.

View File

@@ -0,0 +1,347 @@
/**
* Lossless storage packing for `assistant/chunk` delta runs. Providers stream
* token-sized deltas, so a log stores hundreds of near-identical event lines
* whose JSON envelopes dwarf their payloads (~56× measured on a real DeepSeek
* session). This module packs each run of consecutive same-block delta chunks
* into ONE storage row — `text-chunks`, `reasoning-chunks`, or
* `tool-call-chunks` — and expands rows back to the exact original events.
*
* Storage rows are a durable-encoding vocabulary, NOT session events: they
* never enter `Session.events`, have no `SessionEventMap` entry, and use bare
* (slash-less) type tags so a reader cannot confuse them with the event
* taxonomy (precedent: the JSONL header line's `session` tag). The encoder
* whitelists exact shapes — anything it does not fully recognize is stored
* verbatim, so unknown fields or future chunk variants lose compression, never
* data. The decoder validates before expanding and fails loud on a malformed
* row-tagged value instead of silently dropping a whole run.
*
* @module @deepseek-ai/dsh-session/chunk-rows
*/
import { CallId, assertNever } from '@deepseek-ai/dsh-llm'
import type { StreamChunk } from '@deepseek-ai/dsh-llm'
import type { SessionEvent } from './types.ts'
/** The chunk kinds that may pack; block boundaries, usage, and finish chunks always stay one event per line. */
type DeltaKind = 'text-delta' | 'reasoning-delta' | 'tool-call-delta'
/** A run member: an `assistant/chunk` event whose exact shape the encoder whitelisted. */
type DeltaEvent = SessionEvent<'assistant/chunk'>
/**
* Fields shared by every packed run: placement, block correlation, and member
* timestamps as gaps. Member `k` reconstructs as seq `seq0 + k` and time
* `time0` plus the first `k` gaps; a gap may be negative when the wall clock
* stepped backwards between events.
*/
interface RunDataBase {
turn: number
step: number
/** The stream block index every member shares. */
index: number
/** Epoch-ms gaps between consecutive members; length is one less than the member count. */
dt: number[]
}
/** Payload of a `text-chunks`/`reasoning-chunks` row: one entry per member, never joined — token boundaries are data. */
interface TextRunData extends RunDataBase {
texts: string[]
}
/** Payload of a `tool-call-chunks` row: the run-constant call identity plus each member's raw arguments fragment. */
interface ToolCallRunData extends RunDataBase {
id: CallId
/** Present iff every member carried it, with one uniform value (a mixed run never packs). */
name?: string
args: string[]
}
/**
* A packed run of consecutive delta chunk events, discriminated on `type`.
* `seq0`/`time0` anchor the first member; text and reasoning rows share the
* {@link TextRunData} payload, tool-call rows carry {@link ToolCallRunData}.
*/
export type ChunkRow =
| { type: 'text-chunks'; seq0: number; time0: number; data: TextRunData }
| { type: 'reasoning-chunks'; seq0: number; time0: number; data: TextRunData }
| { type: 'tool-call-chunks'; seq0: number; time0: number; data: ToolCallRunData }
/** One durable log line's JSON value: a session event verbatim, or a packed chunk row. */
export type StorageRecord = SessionEvent | ChunkRow
/**
* Minimum members before a run packs. Below it a row's envelope rivals the
* event lines it replaces. A format constant, not a tunable: both layouts
* decode identically, so changing it never invalidates stored logs.
*/
const MIN_RUN = 3
function isRecord(value: unknown): value is Record<string, unknown> {
return typeof value === 'object' && value !== null
}
/** Exact-key check: `value` has every key in `keys` and nothing else. */
function hasExactKeys(value: object, keys: readonly string[]): boolean {
return Object.keys(value).length === keys.length && keys.every(k => Object.hasOwn(value, k))
}
/**
* Classify an event for packing: its delta kind when the ENTIRE shape
* (envelope, data, chunk — exact keys, primitive types, integer seq/time) is
* whitelisted, else `undefined` (store verbatim). Inputs come from live typed
* appends AND parsed fixture files, so the checks are structural, not
* type-trusted. Integer times keep gap encoding exact: a fractional time would
* reconstruct through float subtraction/addition, which need not round-trip.
*/
function classify(event: SessionEvent): DeltaKind | undefined {
if (event.type !== 'assistant/chunk') return undefined
if (!hasExactKeys(event, ['type', 'seq', 'time', 'data'])) return undefined
if (!Number.isSafeInteger(event.seq) || event.seq < 0 || !Number.isSafeInteger(event.time)) return undefined
const data: unknown = event.data
if (!isRecord(data) || !hasExactKeys(data, ['turn', 'step', 'chunk'])) return undefined
if (typeof data.turn !== 'number' || typeof data.step !== 'number') return undefined
const chunk = data.chunk
if (!isRecord(chunk) || typeof chunk.index !== 'number') return undefined
switch (chunk.type) {
case 'text-delta':
case 'reasoning-delta':
return hasExactKeys(chunk, ['type', 'index', 'text']) && typeof chunk.text === 'string'
? chunk.type
: undefined
case 'tool-call-delta': {
const shapeOk = hasExactKeys(chunk, ['type', 'index', 'id', 'argumentsDelta'])
|| (hasExactKeys(chunk, ['type', 'index', 'id', 'name', 'argumentsDelta']) && typeof chunk.name === 'string')
return shapeOk && typeof chunk.id === 'string' && typeof chunk.argumentsDelta === 'string'
? chunk.type
: undefined
}
// Whitelist fall-through over parsed data: block-start/end, usage, finish,
// and any future chunk variant stay one event per line.
default:
return undefined
}
}
/** The tool-call fields of a whitelisted delta chunk (only after {@link classify} returned `'tool-call-delta'`). */
function toolCallOf(event: DeltaEvent): { id: string; name?: string } {
return event.data.chunk as { id: string; name?: string }
}
/** The block index of a whitelisted delta chunk (not every {@link StreamChunk} variant carries one). */
function indexOf(event: DeltaEvent): number {
return (event.data.chunk as { index: number }).index
}
/** Whether `next` extends a run ending in `prev` (same kind already checked by the caller). */
function continues(prev: DeltaEvent, next: DeltaEvent, kind: DeltaKind): boolean {
if (next.seq !== prev.seq + 1) return false
// Two safe-integer times can sit further apart than a double subtracts
// exactly (2^53-1 and its negation differ by ~2^54); a rounded gap would
// decode to a different timestamp. The check is exact in both directions: a
// true gap within safe range subtracts without rounding and passes, while a
// true gap beyond it rounds to a value that is itself beyond and fails.
if (!Number.isSafeInteger(next.time - prev.time)) return false
if (next.data.turn !== prev.data.turn || next.data.step !== prev.data.step) return false
if (indexOf(next) !== indexOf(prev)) return false
if (kind !== 'tool-call-delta') return true
const a = toolCallOf(prev)
const b = toolCallOf(next)
// `name` must match in presence AND value — a mixed run is not representable.
return a.id === b.id && Object.hasOwn(a, 'name') === Object.hasOwn(b, 'name') && a.name === b.name
}
/** Build the row for a completed run (`run.length >= MIN_RUN`, uniform per {@link continues}). */
function buildRow(kind: DeltaKind, run: readonly DeltaEvent[]): ChunkRow {
const first = run[0] as DeltaEvent
const base = {
turn: first.data.turn,
step: first.data.step,
index: indexOf(first),
dt: run.slice(1).map((event, i) => event.time - (run[i] as DeltaEvent).time),
}
const envelope = { seq0: first.seq, time0: first.time }
if (kind === 'tool-call-delta') {
const call = toolCallOf(first)
return {
type: 'tool-call-chunks',
...envelope,
data: {
...base,
id: CallId(call.id),
...Object.hasOwn(call, 'name') ? { name: call.name as string } : {},
args: run.map(event => (event.data.chunk as { argumentsDelta: string }).argumentsDelta),
},
}
}
const data = { ...base, texts: run.map(event => (event.data.chunk as { text: string }).text) }
return kind === 'text-delta'
? { type: 'text-chunks', ...envelope, data }
: { type: 'reasoning-chunks', ...envelope, data }
}
/**
* Pack an event batch for storage: each run of at least {@link MIN_RUN}
* consecutive whitelisted same-kind, same-block delta chunk events becomes one
* {@link ChunkRow}; every other event passes through verbatim, in order.
* Pure and stateless — safe over any array, including a batch whose runs were
* split by flush boundaries (the split runs simply pack per batch).
*
* @param events - the batch to encode, in log order.
* @returns the storage records to write, one JSONL line each.
*/
export function packChunkRuns(events: readonly SessionEvent[]): StorageRecord[] {
const out: StorageRecord[] = []
let kind: DeltaKind | undefined
let run: DeltaEvent[] = []
const flush = (): void => {
if (kind !== undefined && run.length >= MIN_RUN) out.push(buildRow(kind, run))
else out.push(...run)
kind = undefined
run = []
}
for (const event of events) {
const k = classify(event)
if (k === undefined) {
flush()
out.push(event)
continue
}
const delta = event as DeltaEvent
const last = run[run.length - 1]
if (k === kind && last !== undefined && continues(last, delta, k)) {
run.push(delta)
continue
}
flush()
kind = k
run = [delta]
}
flush()
return out
}
/** Throw the uniform malformed-row diagnostic. */
function malformed(tag: string, why: string): never {
throw new Error(`malformed ${tag} storage row: ${why}`)
}
/** Validate the shared run-data fields and the payload/dt arity; returns the member payload. */
function validateRunData(tag: string, data: Record<string, unknown>, payloadKey: 'texts' | 'args'): string[] {
if (typeof data.turn !== 'number' || typeof data.step !== 'number' || typeof data.index !== 'number') {
malformed(tag, 'turn/step/index must be numbers')
}
const payload = data[payloadKey]
if (!Array.isArray(payload) || payload.length === 0 || payload.some(entry => typeof entry !== 'string')) {
malformed(tag, `${payloadKey} must be a non-empty string array`)
}
const dt = data.dt
if (!Array.isArray(dt) || dt.some(gap => !Number.isSafeInteger(gap))) {
malformed(tag, 'dt must be an array of safe integers')
}
if (dt.length !== payload.length - 1) {
malformed(tag, `dt length ${dt.length} does not match ${payload.length} members`)
}
return payload as string[]
}
/** Validate a row-tagged parsed value's envelope and data, throwing on any malformation. */
function validateRow(value: Record<string, unknown>, tag: ChunkRow['type']): ChunkRow {
if (!hasExactKeys(value, ['type', 'seq0', 'time0', 'data'])) {
malformed(tag, 'envelope must be exactly {type, seq0, time0, data}')
}
if (!Number.isSafeInteger(value.seq0) || (value.seq0 as number) < 0) {
malformed(tag, 'seq0 must be a non-negative safe integer')
}
if (!Number.isSafeInteger(value.time0)) {
malformed(tag, 'time0 must be a safe integer')
}
const data = value.data
if (!isRecord(data)) malformed(tag, 'data must be an object')
let payload: string[]
if (tag === 'tool-call-chunks') {
const withName = hasExactKeys(data, ['turn', 'step', 'index', 'id', 'name', 'dt', 'args'])
if (!withName && !hasExactKeys(data, ['turn', 'step', 'index', 'id', 'dt', 'args'])) {
malformed(tag, 'data must be exactly {turn, step, index, id, name?, dt, args}')
}
if (typeof data.id !== 'string' || (withName && typeof data.name !== 'string')) {
malformed(tag, 'id (and name when present) must be strings')
}
payload = validateRunData(tag, data, 'args')
} else {
if (!hasExactKeys(data, ['turn', 'step', 'index', 'dt', 'texts'])) {
malformed(tag, 'data must be exactly {turn, step, index, dt, texts}')
}
payload = validateRunData(tag, data, 'texts')
}
// Reconstruction bounds. The encoder only packs runs whose member seqs and
// times are all safe integers, so a running value that leaves safe range is
// outside any encoder's image: float arithmetic would round it to a
// different number than exact arithmetic, a silent corruption. Within safe
// range every step is exact, so the first departure is always caught.
if (!Number.isSafeInteger((value.seq0 as number) + payload.length - 1)) {
malformed(tag, 'member seqs must stay safe integers')
}
let time = value.time0 as number
for (const gap of data.dt as number[]) {
time += gap
if (!Number.isSafeInteger(time)) malformed(tag, 'member times must stay safe integers')
}
return value as unknown as ChunkRow
}
/** Expand a validated row back into its exact original events, in order. */
function expandRow(row: ChunkRow): SessionEvent[] {
const members = row.type === 'tool-call-chunks' ? row.data.args : row.data.texts
const events: SessionEvent[] = []
let time = row.time0
for (let k = 0; k < members.length; k++) {
if (k > 0) time += row.data.dt[k - 1] as number
let chunk: StreamChunk
switch (row.type) {
case 'text-chunks':
chunk = { type: 'text-delta', index: row.data.index, text: members[k] as string }
break
case 'reasoning-chunks':
chunk = { type: 'reasoning-delta', index: row.data.index, text: members[k] as string }
break
case 'tool-call-chunks':
chunk = {
type: 'tool-call-delta',
index: row.data.index,
id: row.data.id,
...Object.hasOwn(row.data, 'name') ? { name: row.data.name as string } : {},
argumentsDelta: members[k] as string,
}
break
/* v8 ignore next 2 -- validateRow only returns the three row tags */
default:
return assertNever(row, 'chunk-rows expandRow')
}
events.push({
type: 'assistant/chunk',
seq: row.seq0 + k,
time,
data: { turn: row.data.turn, step: row.data.step, chunk },
})
}
return events
}
/**
* Decode one parsed JSONL line value into the session event(s) it stores.
* Chunk-row-tagged values validate and expand (a malformed row throws — it is
* corrupt storage, and treating it as an event would silently drop a whole
* run); every other value passes through as a single event, unvalidated,
* exactly as readers treated event lines before packing existed.
*
* @param value - one line's `JSON.parse` result.
* @returns the stored events, in log order.
*/
export function decodeStorageRecord(value: unknown): SessionEvent[] {
if (!isRecord(value)) return [value as SessionEvent]
const tag = value.type
if (tag !== 'text-chunks' && tag !== 'reasoning-chunks' && tag !== 'tool-call-chunks') {
return [value as SessionEvent]
}
return expandRow(validateRow(value, tag))
}

View File

@@ -23,6 +23,8 @@ export * from './types.ts'
export { isJsonValue, snapshotJsonValue } from './json.ts'
export type { JsonValue } from './json.ts'
export { interruptedTurnClosers, TOOL_NOT_STARTED, TOOL_OUTCOME_UNKNOWN } from './repair.ts'
export { decodeStorageRecord, packChunkRuns } from './chunk-rows.ts'
export type { ChunkRow, StorageRecord } from './chunk-rows.ts'
export type { SessionSurface, SurfaceFoldReplacement, SurfaceFoldResult } from './surface.ts'
export { foldSurface, isSurfaceEvent, isSurfaceEligibleType } from './surface.ts'
export { canonicalHeader, foldRequestHeader, headerEquals } from './request-header.ts'

View File

@@ -0,0 +1,236 @@
/**
* Chunk-row codec tests: pack/expand round-trip losslessness (example-based and
* property-based), run-boundary rules, whitelist fall-through, and decoder
* validation failures.
*/
import { describe, expect, it } from 'vitest'
import fc from 'fast-check'
import { CallId } from '@deepseek-ai/dsh-llm'
import type { StreamChunk } from '@deepseek-ai/dsh-llm'
import { decodeStorageRecord, packChunkRuns } from '@deepseek-ai/dsh-session'
import type { ChunkRow, SessionEvent, StorageRecord } from '@deepseek-ai/dsh-session'
/** Build an `assistant/chunk` event with the exact live-append shape. */
function chunkEvent(seq: number, time: number, chunk: StreamChunk, turn = 1, step = 1): SessionEvent {
return { type: 'assistant/chunk', seq, time, data: { turn, step, chunk } }
}
/** Sequential delta events (contiguous seqs, fixed 10ms gaps) of one kind. */
function deltaRun(kind: 'text-delta' | 'reasoning-delta', count: number, seq0 = 0, index = 0): SessionEvent[] {
return Array.from({ length: count }, (_, k) =>
chunkEvent(seq0 + k, 1000 + 10 * k, { type: kind, index, text: `t${k}` }))
}
/** Decode a packed record list back to a flat event list. */
function decodeAll(records: readonly StorageRecord[]): SessionEvent[] {
return records.flatMap(record => decodeStorageRecord(JSON.parse(JSON.stringify(record))))
}
describe('packChunkRuns', () => {
it('packs a text-delta run into one text-chunks row and round-trips it', () => {
const events = deltaRun('text-delta', 5)
const packed = packChunkRuns(events)
expect(packed).toHaveLength(1)
const row = packed[0] as ChunkRow
expect(row.type).toBe('text-chunks')
expect(row.seq0).toBe(0)
expect(row.time0).toBe(1000)
expect(row.data).toMatchObject({ turn: 1, step: 1, index: 0, dt: [10, 10, 10, 10], texts: ['t0', 't1', 't2', 't3', 't4'] })
expect(decodeAll(packed)).toStrictEqual(events)
})
it('packs reasoning and tool-call runs under their own tags', () => {
const reasoning = deltaRun('reasoning-delta', 3)
const toolCall = [4, 5, 6].map(seq =>
chunkEvent(seq, 1000 + seq, { type: 'tool-call-delta', index: 1, id: CallId('c1'), name: 'write', argumentsDelta: `a${seq}` }))
const packed = packChunkRuns([...reasoning, ...toolCall])
expect(packed.map(r => (r as ChunkRow).type)).toStrictEqual(['reasoning-chunks', 'tool-call-chunks'])
const row = packed[1] as ChunkRow & { type: 'tool-call-chunks' }
expect(row.data).toMatchObject({ id: 'c1', name: 'write', args: ['a4', 'a5', 'a6'] })
expect(decodeAll(packed)).toStrictEqual([...reasoning, ...toolCall])
})
it('packs a name-less tool-call run and round-trips field absence', () => {
const events = [0, 1, 2].map(seq =>
chunkEvent(seq, 1000, { type: 'tool-call-delta', index: 0, id: CallId('c1'), argumentsDelta: `a${seq}` }))
const packed = packChunkRuns(events)
expect(packed).toHaveLength(1)
expect(Object.hasOwn((packed[0] as ChunkRow).data, 'name')).toBe(false)
const decoded = decodeAll(packed)
expect(decoded).toStrictEqual(events)
expect(decoded.every(e => !Object.hasOwn((e.data as { chunk: object }).chunk, 'name'))).toBe(true)
})
it('leaves runs shorter than three events verbatim', () => {
const events = deltaRun('text-delta', 2)
expect(packChunkRuns(events)).toStrictEqual(events)
})
it('leaves non-delta chunks and non-chunk events verbatim between runs', () => {
const events: SessionEvent[] = [
chunkEvent(0, 1000, { type: 'block-start', index: 0, blockType: 'text' }),
...deltaRun('text-delta', 3, 1),
chunkEvent(4, 1040, { type: 'block-end', index: 0, block: { type: 'text', text: 't0t1t2' } }),
{ type: 'step/end', seq: 5, time: 1050, data: { turn: 1, step: 1 } },
]
const packed = packChunkRuns(events)
expect(packed).toHaveLength(4)
expect((packed[1] as ChunkRow).type).toBe('text-chunks')
expect(decodeAll(packed)).toStrictEqual(events)
})
it.each([
['a seq gap', deltaRun('text-delta', 3).map((e, k) => ({ ...e, seq: k === 2 ? 9 : e.seq }))],
['a kind switch', [...deltaRun('text-delta', 2), ...deltaRun('reasoning-delta', 1, 2)]],
['a block-index switch', [...deltaRun('text-delta', 2), ...deltaRun('text-delta', 1, 2, 7)]],
['a step switch', deltaRun('text-delta', 3).map((e, k) => k === 2 ? chunkEvent(e.seq, e.time, (e.data as { chunk: StreamChunk }).chunk, 1, 2) : e)],
])('breaks a run on %s (both halves too short to pack)', (_label, events) => {
expect(packChunkRuns(events as SessionEvent[])).toStrictEqual(events)
})
it('breaks a tool-call run on call-id or name change', () => {
const call = (seq: number, id: string, name?: string): SessionEvent =>
chunkEvent(seq, 1000, { type: 'tool-call-delta', index: 0, id: CallId(id), ...name !== undefined ? { name } : {}, argumentsDelta: 'a' })
const idSwitch = [call(0, 'c1', 'w'), call(1, 'c1', 'w'), call(2, 'c2', 'w')]
expect(packChunkRuns(idSwitch)).toStrictEqual(idSwitch)
const namePresence = [call(0, 'c1', 'w'), call(1, 'c1', 'w'), call(2, 'c1')]
expect(packChunkRuns(namePresence)).toStrictEqual(namePresence)
})
it('stores an off-whitelist delta verbatim (extra field, bad type, fractional time)', () => {
const extraField = { ...chunkEvent(0, 1000, { type: 'text-delta', index: 0, text: 'x' }), surfaceOp: 'append' }
const badText = chunkEvent(1, 1001, { type: 'text-delta', index: 0, text: 7 as unknown as string })
const fractionalTime = chunkEvent(2, 1001.5, { type: 'text-delta', index: 0, text: 'y' })
const events = [extraField, badText, fractionalTime] as SessionEvent[]
expect(packChunkRuns(events)).toStrictEqual(events)
})
it('breaks a run on a time gap beyond safe-integer range (subtraction would round)', () => {
// Both endpoints are safe integers, but their true difference (~2^54)
// exceeds exact double range: b - a rounds, so a + (b - a) !== b and a
// packed row would decode to a different timestamp.
const a = Number.MIN_SAFE_INTEGER
const b = Number.MAX_SAFE_INTEGER - 1
expect(a + (b - a)).not.toBe(b) // the rounding this guard exists for
const events = [
chunkEvent(0, a, { type: 'text-delta', index: 0, text: 'x' }),
chunkEvent(1, b, { type: 'text-delta', index: 0, text: 'y' }),
chunkEvent(2, b + 1, { type: 'text-delta', index: 0, text: 'z' }),
]
expect(packChunkRuns(events)).toStrictEqual(events) // split at the gap; halves too short
expect(decodeAll(packChunkRuns(events))).toStrictEqual(events)
})
it('stores a delta with an off-whitelist data envelope verbatim (parsed-fixture shapes)', () => {
const mk = (seq: number, data: unknown): SessionEvent =>
({ type: 'assistant/chunk', seq, time: 1000, data } as SessionEvent)
const events = [
mk(0, 'not-an-object'),
mk(1, { turn: 1, step: 1, chunk: { type: 'text-delta', index: 0, text: 'a' }, extra: 1 }),
mk(2, { turn: 'x', step: 1, chunk: { type: 'text-delta', index: 0, text: 'a' } }),
mk(3, { turn: 1, step: 1, chunk: 'not-an-object' }),
mk(4, { turn: 1, step: 1, chunk: { type: 'text-delta', index: 'x', text: 'a' } }),
mk(5, { turn: 1, step: 1, chunk: { type: 'tool-call-delta', index: 0, id: 7, argumentsDelta: 'a' } }),
mk(6, { turn: 1, step: 1, chunk: { type: 'tool-call-delta', index: 0, id: 'c', name: 7, argumentsDelta: 'a' } }),
]
expect(packChunkRuns(events)).toStrictEqual(events)
})
})
describe('decodeStorageRecord', () => {
it('passes non-row values through as single events, unvalidated', () => {
const event = { type: 'turn/start', seq: 0, time: 1, data: { turn: 1 } }
expect(decodeStorageRecord(event)).toStrictEqual([event])
expect(decodeStorageRecord('junk')).toStrictEqual(['junk'])
expect(decodeStorageRecord(null)).toStrictEqual([null])
})
it('reconstructs timestamps through negative dt gaps (clock stepped back)', () => {
const events = [
chunkEvent(0, 1000, { type: 'text-delta', index: 0, text: 'a' }),
chunkEvent(1, 990, { type: 'text-delta', index: 0, text: 'b' }),
chunkEvent(2, 995, { type: 'text-delta', index: 0, text: 'c' }),
]
expect(decodeAll(packChunkRuns(events))).toStrictEqual(events)
})
it.each([
['a non-object data', { type: 'text-chunks', seq0: 0, time0: 1, data: 'x' }],
['an envelope with extra keys', { type: 'text-chunks', seq0: 0, time0: 1, data: { turn: 1, step: 1, index: 0, dt: [], texts: ['a'] }, extra: 1 }],
['a negative seq0', { type: 'text-chunks', seq0: -1, time0: 1, data: { turn: 1, step: 1, index: 0, dt: [], texts: ['a'] } }],
['a non-finite time0', { type: 'text-chunks', seq0: 0, time0: Infinity, data: { turn: 1, step: 1, index: 0, dt: [], texts: ['a'] } }],
['a fractional time0', { type: 'text-chunks', seq0: 0, time0: 1.5, data: { turn: 1, step: 1, index: 0, dt: [], texts: ['a'] } }],
['a data shape mismatch', { type: 'text-chunks', seq0: 0, time0: 1, data: { turn: 1, step: 1, index: 0, dt: [], args: ['a'] } }],
['a non-string member', { type: 'text-chunks', seq0: 0, time0: 1, data: { turn: 1, step: 1, index: 0, dt: [], texts: [7] } }],
['an empty member list', { type: 'text-chunks', seq0: 0, time0: 1, data: { turn: 1, step: 1, index: 0, dt: [], texts: [] } }],
['a dt arity mismatch', { type: 'text-chunks', seq0: 0, time0: 1, data: { turn: 1, step: 1, index: 0, dt: [1, 2], texts: ['a', 'b'] } }],
['a non-finite dt gap', { type: 'text-chunks', seq0: 0, time0: 1, data: { turn: 1, step: 1, index: 0, dt: [NaN], texts: ['a', 'b'] } }],
['a fractional dt gap', { type: 'text-chunks', seq0: 0, time0: 1, data: { turn: 1, step: 1, index: 0, dt: [0.5], texts: ['a', 'b'] } }],
['a member seq leaving safe range', { type: 'text-chunks', seq0: Number.MAX_SAFE_INTEGER, time0: 1, data: { turn: 1, step: 1, index: 0, dt: [0, 0], texts: ['a', 'b', 'c'] } }],
['a member time leaving safe range', { type: 'text-chunks', seq0: 0, time0: Number.MAX_SAFE_INTEGER, data: { turn: 1, step: 1, index: 0, dt: [1], texts: ['a', 'b'] } }],
['a non-numeric turn', { type: 'text-chunks', seq0: 0, time0: 1, data: { turn: 'x', step: 1, index: 0, dt: [], texts: ['a'] } }],
['a tool-call row without id', { type: 'tool-call-chunks', seq0: 0, time0: 1, data: { turn: 1, step: 1, index: 0, dt: [], args: ['a'] } }],
['a tool-call row with non-string id', { type: 'tool-call-chunks', seq0: 0, time0: 1, data: { turn: 1, step: 1, index: 0, id: 7, dt: [], args: ['a'] } }],
['a tool-call row with non-string name', { type: 'tool-call-chunks', seq0: 0, time0: 1, data: { turn: 1, step: 1, index: 0, id: 'c', name: 7, dt: [], args: ['a'] } }],
])('throws on %s', (_label, row) => {
expect(() => decodeStorageRecord(row)).toThrow(/malformed .* storage row/)
})
})
// --- Property: pack∘decode is the identity over arbitrary event batches ---
const deltaChunkArb: fc.Arbitrary<StreamChunk> = fc.oneof(
fc.record({ type: fc.constant<'text-delta'>('text-delta'), index: fc.nat(2), text: fc.string() }),
fc.record({ type: fc.constant<'reasoning-delta'>('reasoning-delta'), index: fc.nat(2), text: fc.string() }),
fc.record({
type: fc.constant<'tool-call-delta'>('tool-call-delta'),
index: fc.nat(2),
id: fc.constantFrom(CallId('c1'), CallId('c2')),
argumentsDelta: fc.string(),
}),
fc.record({
type: fc.constant<'tool-call-delta'>('tool-call-delta'),
index: fc.nat(2),
id: fc.constantFrom(CallId('c1'), CallId('c2')),
name: fc.constantFrom('write', 'read'),
argumentsDelta: fc.string(),
}),
)
const boundaryChunkArb: fc.Arbitrary<StreamChunk> = fc.oneof(
fc.record({ type: fc.constant<'block-start'>('block-start'), index: fc.nat(2), blockType: fc.constant<'text'>('text') }),
fc.record({ type: fc.constant<'finish'>('finish'), reason: fc.constant({ kind: 'stop' as const }) }),
)
/**
* Batches with contiguous seqs, arbitrary timestamps, mixed chunk kinds and
* turn/step placement. Times draw from the FULL safe-integer range (not just
* realistic clocks) so the property exercises the gap-overflow guard: two safe
* endpoints can differ by more than a double subtracts exactly.
*/
const batchArb: fc.Arbitrary<SessionEvent[]> = fc.array(
fc.record({
chunk: fc.oneof({ weight: 4, arbitrary: deltaChunkArb }, { weight: 1, arbitrary: boundaryChunkArb }),
time: fc.oneof(
{ weight: 4, arbitrary: fc.integer({ min: 995, max: 9000 }) },
{ weight: 1, arbitrary: fc.integer({ min: Number.MIN_SAFE_INTEGER, max: Number.MAX_SAFE_INTEGER }) },
),
turn: fc.nat(1),
step: fc.nat(1),
}),
{ maxLength: 40 },
// JSON round-trip normalizes fast-check's null-prototype records into the
// plain objects real log events are (the log is JSON), so equality compares
// values, not prototypes.
).map(entries => JSON.parse(JSON.stringify(
entries.map((entry, k) => chunkEvent(k, entry.time, entry.chunk, entry.turn, entry.step)),
)) as SessionEvent[])
describe('chunk-row codec properties', () => {
it('JSON-serialized pack∘decode reproduces every batch exactly', () => {
fc.assert(fc.property(batchArb, (events) => {
expect(decodeAll(packChunkRuns(events))).toStrictEqual(events)
}))
})
})