From 11a29fdefe5a3fbc71e22758fee110d2aae5cf83 Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Sat, 13 Jun 2026 23:25:12 +0800 Subject: [PATCH 1/5] feat(invariants): dev-mode event-contract assertions + session-log freeze (RFC 005 pt 3, RFC 008) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit New @deepseek-ai/dsh-invariants plugin (pure listeners, off in prod) asserts the event taxonomy at runtime — seq monotonicity, turn/step nesting, a tool/result needs a prior tool/call (NOT the converse), legal agent/status transitions — and deep-freezes logged event data so mutating history throws. Seeded sessions are checked + frozen on session/created. The real RFC 008 fix is always-on: deriveMessages now structured-clones the content it emits, so the loop's sanctioned request/adapter mutation can no longer reach back and rewrite the append-only log. The pervasive DeepReadonly type flip is rejected (compile-only, high-noise, castable) — recorded in ADR 0012, which folds in RFC 008. Wired into both demos. --- .../0012-dev-invariants-over-deep-readonly.md | 27 ++ docs/adr/README.md | 1 + ...5-runtime-validation-and-error-taxonomy.md | 2 +- docs/rfc/008-immutable-public-surfaces.md | 2 +- docs/rfc/README.md | 2 +- examples/coding-agent/cordis.yml | 4 + examples/echo-agent/cordis.yml | 5 + packages/invariants/README.md | 46 ++++ packages/invariants/package.json | 33 +++ packages/invariants/src/index.ts | 193 ++++++++++++++ packages/invariants/tests/invariants.spec.ts | 239 ++++++++++++++++++ packages/invariants/tsconfig.json | 15 ++ packages/session/src/index.ts | 18 +- packages/session/tests/session.spec.ts | 25 ++ scripts/publint-all.ts | 1 + tsconfig.base.json | 3 +- tsconfig.build.json | 3 +- tsconfig.typecheck.json | 3 +- yarn.lock | 15 ++ 19 files changed, 626 insertions(+), 11 deletions(-) create mode 100644 docs/adr/0012-dev-invariants-over-deep-readonly.md create mode 100644 packages/invariants/README.md create mode 100644 packages/invariants/package.json create mode 100644 packages/invariants/src/index.ts create mode 100644 packages/invariants/tests/invariants.spec.ts create mode 100644 packages/invariants/tsconfig.json diff --git a/docs/adr/0012-dev-invariants-over-deep-readonly.md b/docs/adr/0012-dev-invariants-over-deep-readonly.md new file mode 100644 index 0000000000..01d711e38c --- /dev/null +++ b/docs/adr/0012-dev-invariants-over-deep-readonly.md @@ -0,0 +1,27 @@ +# ADR 0012: Dev-mode invariants over compile-time deep-readonly + +Status: accepted (2026-06-13) + +## Context + +The session log is append-only by contract, but the types don't enforce it: `session.events` returns `readonly SessionEvent[]` whose *elements* are mutable, and `deriveMessages()` handed the logged `content` arrays/blocks out by reference. The loop then passes those derived messages into the `agent/request` waterfall and on to adapters, where mutating the request is sanctioned — so a request middleware could reach back and rewrite history, silently breaking replay equivalence and the derived-history guarantee. Separately, the event taxonomy (turn/step nesting, seq monotonicity, tool-call/result pairing, legal status transitions) was asserted only where individual tests happened to look. + +Two ways to defend the log: make immutability part of the type (`DeepReadonly` on the way out), or catch corruption at runtime in dev. The RFC (005) proposed the runtime route; RFC 008 proposed the type route. + +## Decision + +Reject the pervasive `DeepReadonly` type flip. Instead: + +1. **Always-on:** `deriveMessages()` deep-clones the content it emits (one `structuredClone` per derived message). In-flight mutation of a request can no longer reach the log — this is the real fix, and it costs nothing meaningful next to a model call. +2. **Dev-mode:** a new `dsh-invariants` plugin (pure listeners, off in production, on in tests and demos) asserts the event contract and `Object.freeze`s logged event data so any *other* code that mutates a logged event throws instead of corrupting silently. Seeded sessions are frozen and checked on `session/created` (the constructor copies the seed without emitting `session/event`). + +The invariants encode the *real* contract, not an idealized one: a `tool/call` may have no `tool/result` (a thrown `tools/execute` waterfall ends the step), and both `idle→disposed` and `running→disposed` are legal. + +`DeepReadonly` was rejected because it is compile-time only (a plugin casts straight through it), high type-noise across every log/message consumer and adapter, and would force readonly types through code where mutation is the sanctioned API. The clone draws the mutable/immutable boundary exactly at "logged vs in-flight" without any of that noise. + +## Consequences + +- History corruption is caught loudly in tests and demos, at zero production cost and zero type noise. The trade-off is that the guarantee is dynamic (a dev-mode tripwire) rather than static. +- The invariants plugin doubles as executable documentation of the event taxonomy — the assertions are the contract. +- `Session.events` keeps its `readonly SessionEvent[]` type; no consumer churn. +- This folds in RFC 008 — there is no separate deep-readonly ADR; this records the decision to *not* pursue that approach. `InvariantError` is a plain `Error` with a `code` for now; a later taxonomy change can promote it. diff --git a/docs/adr/README.md b/docs/adr/README.md index a4c7a6e84f..ee3ebde6c6 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -23,3 +23,4 @@ Do NOT write an ADR for: a mechanical or local choice (a variable name, a one-fi | [0009](0009-capability-seams.md) | Capability seams — interface / implementation / consumer split | accepted | | [0010](0010-twin-llm-adapters.md) | Two LLM adapters as a design-verification twin | accepted | | [0011](0011-runtime-arg-validation.md) | Runtime arg validation at the model boundary | accepted | +| [0012](0012-dev-invariants-over-deep-readonly.md) | Dev-mode invariants over compile-time deep-readonly | accepted | diff --git a/docs/rfc/005-runtime-validation-and-error-taxonomy.md b/docs/rfc/005-runtime-validation-and-error-taxonomy.md index a84bb4c264..a754c638ff 100644 --- a/docs/rfc/005-runtime-validation-and-error-taxonomy.md +++ b/docs/rfc/005-runtime-validation-and-error-taxonomy.md @@ -1,6 +1,6 @@ # RFC 005: Runtime validation at the model boundary, error taxonomy, dev-mode invariants -Status: partially implemented — part 1 (arg validation) → [ADR 0011](../adr/0011-runtime-arg-validation.md); parts 2-3 in progress +Status: partially implemented — part 1 (arg validation) → [ADR 0011](../adr/0011-runtime-arg-validation.md); part 3 (dev invariants) → [ADR 0012](../adr/0012-dev-invariants-over-deep-readonly.md); part 2 (error taxonomy) in progress ## Problem diff --git a/docs/rfc/008-immutable-public-surfaces.md b/docs/rfc/008-immutable-public-surfaces.md index e4faf9a0a4..fae635a78e 100644 --- a/docs/rfc/008-immutable-public-surfaces.md +++ b/docs/rfc/008-immutable-public-surfaces.md @@ -1,6 +1,6 @@ # RFC 008: Deep-readonly public surfaces -Status: proposed +Status: implemented (revised) — the pervasive `DeepReadonly` type flip was rejected in favor of an always-on `deriveMessages` clone plus dev-mode `Object.freeze` + invariants. See [ADR 0012](../adr/0012-dev-invariants-over-deep-readonly.md). ## Problem diff --git a/docs/rfc/README.md b/docs/rfc/README.md index 46613a73b4..4760bb98f6 100644 --- a/docs/rfc/README.md +++ b/docs/rfc/README.md @@ -11,4 +11,4 @@ Proposals for substantial future work — reviewed before implementation, unlike | [005](005-runtime-validation-and-error-taxonomy.md) | Runtime arg validation, structured error taxonomy, dev-mode invariants | partially implemented | | [006](006-doc-sync-and-api-reports.md) | Doc-sync enforcement and API extractor reports | proposed | | [007](007-supply-chain-and-vendor-drift.md) | Supply chain checks and vendor drift verification | proposed | -| [008](008-immutable-public-surfaces.md) | Deep-readonly public surfaces | proposed | +| [008](008-immutable-public-surfaces.md) | Deep-readonly public surfaces | implemented (revised) | diff --git a/examples/coding-agent/cordis.yml b/examples/coding-agent/cordis.yml index 96a6fef4b1..1e1104254c 100644 --- a/examples/coding-agent/cordis.yml +++ b/examples/coding-agent/cordis.yml @@ -30,6 +30,10 @@ - id: agents name: '@deepseek-ai/dsh-agent' +# Dev-mode event-contract assertions + session-log freeze (off in prod). +- id: invariants + name: '@deepseek-ai/dsh-invariants' + # The DeepSeek adapter. Swap to '@deepseek-ai/dsh-llm-pi-ai' for the # pi-ai-backed twin (same config shape; `reasoning: high` replaces # thinking/reasoningEffort). diff --git a/examples/echo-agent/cordis.yml b/examples/echo-agent/cordis.yml index 62e42f9ec6..276dd700ef 100644 --- a/examples/echo-agent/cordis.yml +++ b/examples/echo-agent/cordis.yml @@ -27,6 +27,11 @@ - id: agents name: '@deepseek-ai/dsh-agent' +# Dev-mode event-contract assertions + session-log freeze (off in prod; +# on here so the demo smoke test exercises the contract). +- id: invariants + name: '@deepseek-ai/dsh-invariants' + - id: agent-loop name: '@deepseek-ai/dsh-agent-loop' config: diff --git a/packages/invariants/README.md b/packages/invariants/README.md new file mode 100644 index 0000000000..ee12612006 --- /dev/null +++ b/packages/invariants/README.md @@ -0,0 +1,46 @@ +# dsh-invariants + +Dev-mode event-contract invariants and session-log freeze. A pure-listener plugin (everything is a plugin) that asserts the harness event contract at runtime and, optionally, freezes logged session-event data so any code that mutates history throws instead of corrupting silently. + +**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. + +## Plugin + +```ts +import Invariants from '@deepseek-ai/dsh-invariants' + +await ctx.plugin(Invariants) // freeze on (default) +await ctx.plugin(Invariants, { freeze: false }) // assert contract, don't freeze +``` + +`inject`: none required — it listens on `session/created`, `session/event`, and `agent/status`, all emitted by services it does not depend on directly. + +### Config + +| Key | Default | Meaning | +|---|---|---| +| `freeze` | `true` | Deep-freeze each logged event's data so mutating a logged event throws. Set `false` to assert the contract without freezing. | + +## Invariants asserted + +Session log (per session): + +- **`seq` strictly increases** — the spine of replay equivalence. +- **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. +- **a `tool/result` needs a prior `tool/call`** — but NOT the converse: a `tool/call` may have no result (a thrown `tools/execute` waterfall ends the step with no `tool/result`, which is legal). + +Agent status (per agent): + +- **legal transitions only** — `idle↔running` and `(idle|running)→disposed`. A no-op transition (`setStatus` dedups, so it never fires) and leaving the terminal `disposed` state are violations. + +On any violation it throws `InvariantError` (`code: 'INVARIANT'`). + +## Why runtime, not deep-readonly types + +A `DeepReadonly` is high type-noise across every log consumer, and a plugin can cast straight through it. A dev-mode freeze plus these assertions catch real corruption at zero production cost and zero type noise. The always-on half of that defense — cloning derived messages so request/adapter mutation can't reach back into the log — lives in `dsh-session`'s `deriveMessages`. This package is the dev-mode tripwire. See [ADR 0012](../../docs/adr/0012-dev-invariants-over-deep-readonly.md). + +## Seeded sessions + +A seeded/forked session arrives with events already in its log (the `Session` constructor copies the seed without emitting `session/event`). On `session/created` the plugin replays the existing log through the checker and freezes those entries, so seeded history is held to the same contract. diff --git a/packages/invariants/package.json b/packages/invariants/package.json new file mode 100644 index 0000000000..5e8bc76d91 --- /dev/null +++ b/packages/invariants/package.json @@ -0,0 +1,33 @@ +{ + "name": "@deepseek-ai/dsh-invariants", + "description": "Dev-mode event-contract invariants + session-log freeze for the DeepSeek Harness", + "version": "0.0.1", + "private": true, + "type": "module", + "main": "lib/index.js", + "types": "lib/index.d.ts", + "exports": { + ".": { + "types": "./lib/index.d.ts", + "default": "./lib/index.js" + }, + "./src/*": "./src/*", + "./package.json": "./package.json" + }, + "files": [ + "lib", + "src" + ], + "license": "BSD-3-Clause", + "peerDependencies": { + "@deepseek-ai/dsh-agent": "^0.0.1", + "@deepseek-ai/dsh-session": "^0.0.1", + "cordis": "^4.0.0-rc.6" + }, + "devDependencies": { + "@deepseek-ai/dsh-agent": "^0.0.1", + "@deepseek-ai/dsh-llm": "^0.0.1", + "@deepseek-ai/dsh-session": "^0.0.1", + "cordis": "^4.0.0-rc.6" + } +} diff --git a/packages/invariants/src/index.ts b/packages/invariants/src/index.ts new file mode 100644 index 0000000000..4f04f70209 --- /dev/null +++ b/packages/invariants/src/index.ts @@ -0,0 +1,193 @@ +/** + * Dev-mode invariants: a pure-listener plugin that asserts the harness event + * contract at runtime, and (optionally) freezes logged session-event data so + * any code that mutates history throws instead of corrupting silently. + * + * Everything is a plugin — this is just listeners on `session/created`, + * `session/event`, and `agent/status`. It is **off in production**: enable it + * in tests and the demos, where a contract violation should be a loud failure, + * not a subtle one. It doubles as executable documentation of the event + * taxonomy: the assertions below ARE the contract. + * + * Why runtime assertions instead of compile-time deep-readonly types? See + * ADR 0012. Briefly: a `DeepReadonly` is high type-noise across + * every log consumer and a plugin casts straight through it; a dev-mode freeze + * + assertions catch real corruption at zero production cost and zero type + * noise. The always-on half of that defense (cloning derived messages) lives + * in dsh-session; this package is the dev-mode tripwire. + * + * @module @deepseek-ai/dsh-invariants + */ + +import type { Context } from 'cordis' +import type { Agent, AgentStatus } from '@deepseek-ai/dsh-agent' +import type { Session, SessionEvent } from '@deepseek-ai/dsh-session' + +export const name = 'invariants' + +/** + * Thrown when a harness event-contract invariant is violated. Plain `Error` + * with a `code` for now; a later change promotes the harness error taxonomy. + */ +export class InvariantError extends Error { + readonly code = 'INVARIANT' + constructor(message: string) { + super(`invariant violated: ${message}`) + this.name = 'InvariantError' + } +} + +/** Plugin config. */ +export interface Config { + /** + * Deep-freeze logged session-event data so mutating a logged event throws. + * Default true — this plugin only runs in dev/test, where freezing is the + * point. Set false to assert the event contract without freezing. + */ + freeze?: boolean +} + +/** Per-session bookkeeping for the session-log invariants. */ +interface SessionTrace { + /** Highest `seq` seen so far (must strictly increase). */ + lastSeq: number + /** Open turn number, or null between turns. */ + openTurn: number | null + /** Open step within the current turn, or null between steps. */ + openStep: number | null + /** Outstanding tool-call ids awaiting a result (a result needs a prior call). */ + pendingCalls: Set +} + +/** Deep-freeze a value and everything reachable from it. Idempotent. */ +function deepFreeze(value: unknown): void { + if (value === null || typeof value !== 'object') return + if (Object.isFrozen(value)) return + Object.freeze(value) + for (const key of Object.keys(value)) { + deepFreeze((value as Record)[key]) + } +} + +/** Assert one appended event against the per-session invariants. */ +function checkEvent(trace: SessionTrace, event: SessionEvent): void { + // seq is strictly monotonic — the spine of replay equivalence. lastSeq + // starts at -1, so the first event (seq 0) passes. + if (event.seq <= trace.lastSeq) { + throw new InvariantError(`seq must strictly increase: saw ${event.seq} after ${trace.lastSeq}`) + } + trace.lastSeq = event.seq + + // Intentionally non-exhaustive: only events that carry ordering structure + // are checked; the rest are trace/replay data with no nesting contract. + // eslint-disable-next-line @typescript-eslint/switch-exhaustiveness-check + switch (event.type) { + case 'turn/start': { + if (trace.openTurn !== null) { + throw new InvariantError(`turn/start ${event.data.turn} while turn ${trace.openTurn} is still open`) + } + trace.openTurn = event.data.turn + break + } + case 'turn/end': { + if (trace.openTurn !== event.data.turn) { + throw new InvariantError(`turn/end ${event.data.turn} does not match open turn ${trace.openTurn}`) + } + trace.openTurn = null + trace.openStep = null + break + } + case 'step/start': { + if (trace.openTurn !== event.data.turn) { + throw new InvariantError(`step/start in turn ${event.data.turn} but open turn is ${trace.openTurn}`) + } + trace.openStep = event.data.step + break + } + case 'step/end': { + if (trace.openStep !== event.data.step) { + throw new InvariantError(`step/end ${event.data.step} does not match open step ${trace.openStep}`) + } + trace.openStep = null + break + } + case 'assistant/chunk': { + // A chunk belongs to an open step — step/start must precede it. + if (trace.openStep === null) { + throw new InvariantError('assistant/chunk outside an open step (step/start must precede its chunks)') + } + break + } + case 'tool/call': { + trace.pendingCalls.add(event.data.callId) + break + } + case 'tool/result': { + // A result needs a prior matching call. (The converse does NOT hold: a + // call may have no result — a thrown tools/execute waterfall ends the + // step with no tool/result, which is legal.) + if (!trace.pendingCalls.delete(event.data.callId)) { + throw new InvariantError(`tool/result for ${event.data.callId} with no prior tool/call`) + } + break + } + } +} + +/** Legal agent status transitions (the only state machine the loop guarantees). */ +function checkTransition(from: AgentStatus | undefined, to: AgentStatus): void { + // First observation: any status is a valid starting point. + if (from === undefined) return + // A no-op transition is illegal — setStatus dedups, so we never see it. + if (from === to) { + throw new InvariantError(`agent/status repeated ${to} (no-op transition)`) + } + // Leaving `disposed` is illegal — disposal is terminal. + if (from === 'disposed') { + throw new InvariantError(`agent/status left terminal state disposed → ${to}`) + } + // idle↔running and (idle|running)→disposed are all legal; nothing else exists. +} + +/** + * Register the dev-mode invariants. Returns nothing — contributions are + * effect-scoped, so disposing the plugin fiber removes all listeners and + * stops freezing (HMR-safe). + */ +export function apply(ctx: Context, config: Config = {}): void { + const freeze = config.freeze ?? true + const traces = new WeakMap() + const lastStatus = new WeakMap() + + const traceFor = (session: Session): SessionTrace => { + let trace = traces.get(session) + if (!trace) { + trace = { lastSeq: -1, openTurn: null, openStep: null, pendingCalls: new Set() } + traces.set(session, trace) + } + return trace + } + + ctx.on('session/created', (session) => { + // A seeded/forked session arrives with events already in its log — the + // constructor copies the seed WITHOUT emitting session/event, so replay + // them through the checker here and freeze the existing entries. + const trace = traceFor(session) + for (const event of session.events) { + checkEvent(trace, event) + if (freeze) deepFreeze(event) + } + }) + + ctx.on('session/event', (session, event) => { + checkEvent(traceFor(session), event) + if (freeze) deepFreeze(event) + }) + + ctx.on('agent/status', (agent, status) => { + checkTransition(lastStatus.get(agent), status) + lastStatus.set(agent, status) + }) +} + +export default apply diff --git a/packages/invariants/tests/invariants.spec.ts b/packages/invariants/tests/invariants.spec.ts new file mode 100644 index 0000000000..3b7f88f6d4 --- /dev/null +++ b/packages/invariants/tests/invariants.spec.ts @@ -0,0 +1,239 @@ +import { describe, expect, it, vi } from 'vitest' +import { Context } from 'cordis' +import { CallId } from '@deepseek-ai/dsh-llm' +import type { Agent } from '@deepseek-ai/dsh-agent' +import SessionStore from '@deepseek-ai/dsh-session' +import Invariants, { InvariantError } from '@deepseek-ai/dsh-invariants' + +/** A Context with the session store and the invariants plugin registered. */ +async function setup(config?: { freeze?: boolean }) { + const ctx = new Context() + await ctx.plugin(SessionStore) + const fiber = await ctx.plugin(Invariants, config ?? {}) + return { ctx, fiber } +} + +/** A minimal Agent stand-in for agent/status emission. */ +function mockAgent(id: string): Agent { + return { id } as unknown as Agent +} + +describe('session-log invariants', () => { + it('accepts a well-formed turn/step/tool sequence', async () => { + const { ctx } = await setup({ freeze: false }) + const session = ctx.sessions.create() + expect(() => { + session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) + session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }) + session.append('step/start', { turn: 1, step: 1 }) + session.append('assistant/chunk', { turn: 1, step: 1, chunk: { type: 'text-delta', index: 0, text: 'h' } }) + session.append('assistant/message', { turn: 1, step: 1, content: [{ type: 'tool-call', id: CallId('c1'), name: 'echo', arguments: '{}' }] }) + session.append('tool/call', { turn: 1, step: 1, callId: CallId('c1'), name: 'echo', arguments: '{}' }) + session.append('tool/result', { turn: 1, step: 1, callId: CallId('c1'), content: [{ type: 'text', text: 'ok' }], isError: false }) + session.append('step/end', { turn: 1, step: 1 }) + session.append('turn/end', { turn: 1, reason: { kind: 'completed' } }) + }).not.toThrow() + }) + + it('rejects a turn/start while another turn is open', async () => { + const { ctx } = await setup({ freeze: false }) + const session = ctx.sessions.create() + session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) + expect(() => session.append('turn/start', { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } })) + .toThrow(/turn 1 is still open/) + }) + + it('rejects a turn/end that does not match the open turn', async () => { + const { ctx } = await setup({ freeze: false }) + const session = ctx.sessions.create() + session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) + expect(() => session.append('turn/end', { turn: 2, reason: { kind: 'completed' } })) + .toThrow(/does not match open turn 1/) + }) + + it('rejects a step/start outside its declared turn', async () => { + const { ctx } = await setup({ freeze: false }) + const session = ctx.sessions.create() + session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) + expect(() => session.append('step/start', { turn: 2, step: 1 })).toThrow(/open turn is 1/) + }) + + it('rejects a step/end that does not match the open step', async () => { + const { ctx } = await setup({ freeze: false }) + 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 }) + expect(() => session.append('step/end', { turn: 1, step: 2 })).toThrow(/does not match open step 1/) + }) + + it('rejects an assistant/chunk outside an open step', async () => { + const { ctx } = await setup({ freeze: false }) + const session = ctx.sessions.create() + session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) + expect(() => session.append('assistant/chunk', { turn: 1, step: 1, chunk: { type: 'text-delta', index: 0, text: 'x' } })) + .toThrow(/outside an open step/) + }) + + it('rejects a tool/result with no prior tool/call', async () => { + const { ctx } = await setup({ freeze: false }) + 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 }) + expect(() => session.append('tool/result', { turn: 1, step: 1, callId: CallId('ghost'), content: [], isError: false })) + .toThrow(/no prior tool\/call/) + }) + + it('allows a tool/call with no matching tool/result (thrown waterfall ends the step)', async () => { + const { ctx } = await setup({ freeze: false }) + const session = ctx.sessions.create() + expect(() => { + session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) + session.append('step/start', { turn: 1, step: 1 }) + session.append('tool/call', { turn: 1, step: 1, callId: CallId('c1'), name: 'echo', arguments: '{}' }) + session.append('step/end', { turn: 1, step: 1 }) + session.append('turn/end', { turn: 1, reason: { kind: 'error', message: 'boom' } }) + }).not.toThrow() + }) + + it('holds seeded sessions to the contract on session/created', async () => { + const { ctx } = await setup({ freeze: false }) + // A seed whose seq is non-monotonic must be rejected when the session is + // created (the constructor copies the seed without emitting session/event). + const badSeed = [ + { type: 'turn/start' as const, seq: 0, time: 0, data: { turn: 1, trigger: { kind: 'message' as const, source: { kind: 'user' as const } } } }, + { type: 'turn/start' as const, seq: 0, time: 0, data: { turn: 2, trigger: { kind: 'message' as const, source: { kind: 'user' as const } } } }, + ] + expect(() => ctx.sessions.create(undefined, badSeed)).toThrow(InvariantError) + }) + + it('tracks turns per session independently', async () => { + const { ctx } = await setup({ freeze: false }) + const a = ctx.sessions.create('a') + const b = ctx.sessions.create('b') + a.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) + // b is a fresh session — its own turn/start must not see a's open turn. + expect(() => b.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })).not.toThrow() + }) +}) + +describe('dev-freeze', () => { + it('freezes appended event data so mutating a logged event throws', async () => { + const { ctx } = await setup() // freeze defaults true + const session = ctx.sessions.create() + const event = session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }) + expect(Object.isFrozen(event)).toBe(true) + expect(Object.isFrozen(event.data)).toBe(true) + expect(Object.isFrozen(event.data.content)).toBe(true) + expect(() => { (event.data.content[0] as { text: string }).text = 'HACKED' }).toThrow() + }) + + it('does not freeze when freeze:false', async () => { + const { ctx } = await setup({ freeze: false }) + const session = ctx.sessions.create() + const event = session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }) + expect(Object.isFrozen(event)).toBe(false) + }) + + it('freezes seeded events on session/created', async () => { + const { ctx } = await setup() + const seed = [ + { type: 'user/message' as const, seq: 0, time: 0, data: { content: [{ type: 'text' as const, text: 'seeded' }], source: { kind: 'user' as const } } }, + ] + const session = ctx.sessions.create(undefined, seed) + expect(Object.isFrozen(session.events[0])).toBe(true) + }) + + it('is idempotent over already-frozen sub-structures', async () => { + const { ctx } = await setup() + const session = ctx.sessions.create() + // Pre-freeze a block before appending; deepFreeze must short-circuit on it + // (the already-frozen guard) while still freezing the enclosing event. + const block = Object.freeze({ type: 'text' as const, text: 'pre-frozen' }) + const event = session.append('user/message', { content: [block], source: { kind: 'user' } }) + expect(Object.isFrozen(event)).toBe(true) + expect(Object.isFrozen(event.data.content)).toBe(true) + expect(Object.isFrozen(event.data.content[0])).toBe(true) + }) +}) + +describe('agent status invariants', () => { + it('accepts legal transitions: idle→running→idle and →disposed', async () => { + const { ctx } = await setup({ freeze: false }) + const agent = mockAgent('a1') + expect(() => { + ctx.emit('agent/status', agent, 'idle') + ctx.emit('agent/status', agent, 'running') + ctx.emit('agent/status', agent, 'idle') + ctx.emit('agent/status', agent, 'disposed') + }).not.toThrow() + }) + + it('accepts running→disposed', async () => { + const { ctx } = await setup({ freeze: false }) + const agent = mockAgent('a2') + ctx.emit('agent/status', agent, 'running') + expect(() => { ctx.emit('agent/status', agent, 'disposed') }).not.toThrow() + }) + + it('rejects a no-op transition', async () => { + const { ctx } = await setup({ freeze: false }) + const agent = mockAgent('a3') + ctx.emit('agent/status', agent, 'running') + expect(() => { ctx.emit('agent/status', agent, 'running') }).toThrow(/no-op transition/) + }) + + it('rejects leaving the terminal disposed state', async () => { + const { ctx } = await setup({ freeze: false }) + const agent = mockAgent('a4') + ctx.emit('agent/status', agent, 'disposed') + expect(() => { ctx.emit('agent/status', agent, 'idle') }).toThrow(/left terminal state disposed/) + }) + + it('tracks status per agent independently', async () => { + const { ctx } = await setup({ freeze: false }) + const a = mockAgent('a5') + const b = mockAgent('b5') + ctx.emit('agent/status', a, 'running') + // b's first observation is independent of a. + expect(() => { ctx.emit('agent/status', b, 'running') }).not.toThrow() + }) +}) + +describe('HMR safety', () => { + it('removes all listeners when the plugin fiber is disposed', async () => { + const { ctx, fiber } = await setup() + const session = ctx.sessions.create() + session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) + + await fiber.dispose() + + // After disposal: no freezing, no assertions. An event that WOULD have + // violated the open-turn rule now passes silently, and is not frozen. + const event = session.append('turn/start', { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } }) + expect(Object.isFrozen(event)).toBe(false) + // A no-op status transition no longer throws either. + const agent = mockAgent('hmr') + ctx.emit('agent/status', agent, 'idle') + expect(() => { ctx.emit('agent/status', agent, 'idle') }).not.toThrow() + }) + + it('InvariantError carries a stable code', () => { + const err = new InvariantError('seq must strictly increase') + expect(err).toBeInstanceOf(Error) + expect(err.name).toBe('InvariantError') + expect(err.code).toBe('INVARIANT') + expect(err.message).toBe('invariant violated: seq must strictly increase') + }) + + it('does not leak listeners across dispose (no stale freezing)', async () => { + const { ctx, fiber } = await setup() + await fiber.dispose() + const spy = vi.fn() + ctx.on('session/event', spy) + const session = ctx.sessions.create() + session.append('user/message', { content: [{ type: 'text', text: 'x' }], source: { kind: 'user' } }) + // our own spy fires, proving events still flow — but the plugin's frozen. + expect(spy).toHaveBeenCalledOnce() + expect(Object.isFrozen(session.events[0])).toBe(false) + }) +}) diff --git a/packages/invariants/tsconfig.json b/packages/invariants/tsconfig.json new file mode 100644 index 0000000000..54fbb4adac --- /dev/null +++ b/packages/invariants/tsconfig.json @@ -0,0 +1,15 @@ +{ + "extends": "../../tsconfig.base.json", + "compilerOptions": { + "rootDir": "src", + "outDir": "lib" + }, + "include": ["src"], + "references": [ + { "path": "../../vendor/cosmokit" }, + { "path": "../../vendor/cordis" }, + { "path": "../llm" }, + { "path": "../session" }, + { "path": "../agent" } + ] +} diff --git a/packages/session/src/index.ts b/packages/session/src/index.ts index 7b3b04812c..75147899f0 100644 --- a/packages/session/src/index.ts +++ b/packages/session/src/index.ts @@ -94,6 +94,14 @@ export class Session { * - `tool/result` → user message carrying a tool-result block * - `context/message` / `steering/message` → tagged synthetic user messages * at their chronological position + * + * The returned `content` is **deep-cloned** off the logged events: the loop + * hands these messages into the mutable `agent/request` waterfall and on to + * adapters, where mutating the request is sanctioned — but the session log + * is append-only by contract. Cloning at this boundary keeps in-flight + * mutation from reaching back and rewriting history (which would silently + * break replay equivalence). Cost is one structured clone per step, + * negligible next to a model call. */ deriveMessages(): Message[] { const messages: Message[] = [] @@ -104,29 +112,29 @@ export class Session { // eslint-disable-next-line @typescript-eslint/switch-exhaustiveness-check switch (event.type) { case 'user/message': { - messages.push({ role: 'user', content: event.data.content }) + messages.push({ role: 'user', content: structuredClone(event.data.content) }) break } case 'assistant/message': { - messages.push({ role: 'assistant', content: event.data.content }) + messages.push({ role: 'assistant', content: structuredClone(event.data.content) }) break } case 'tool/result': { const { callId, content, isError } = event.data messages.push({ role: 'user', - content: [{ type: 'tool-result', toolCallId: callId, content, isError }], + content: [{ type: 'tool-result', toolCallId: callId, content: structuredClone(content), isError }], }) break } case 'context/message': { const { content, source } = event.data - messages.push({ role: 'user', content: renderTagged('context', content, source) }) + messages.push({ role: 'user', content: renderTagged('context', structuredClone(content), source) }) break } case 'steering/message': { const { content, source } = event.data - messages.push({ role: 'user', content: renderTagged('steering', content, source) }) + messages.push({ role: 'user', content: renderTagged('steering', structuredClone(content), source) }) break } } diff --git a/packages/session/tests/session.spec.ts b/packages/session/tests/session.spec.ts index 44f2bae980..740ce78bda 100644 --- a/packages/session/tests/session.spec.ts +++ b/packages/session/tests/session.spec.ts @@ -54,6 +54,31 @@ describe('Session', () => { expect(replayed.deriveMessages()).toEqual(original.deriveMessages()) expect(replayed.seq).toBe(original.seq) }) + + it('isolates the log from mutation through a derived message (append-only contract)', () => { + const session = new Session(SessionId('s4')) + session.append('user/message', { content: [{ type: 'text', text: 'original' }], source: { kind: 'user' } }) + session.append('tool/result', { + turn: 1, step: 1, callId: CallId('c1'), + content: [{ type: 'text', text: 'tool out' }], isError: false, + }) + const before = structuredClone(session.events) + + // A request middleware / adapter mutates the messages it was handed. + const messages = session.deriveMessages() + const userBlock = messages[0]!.content[0]! + if (userBlock.type === 'text') userBlock.text = 'HACKED' + const toolBlock = messages[1]!.content[0]! + if (toolBlock.type === 'tool-result') { + toolBlock.content.push({ type: 'text', text: 'injected' }) + } + messages[0]!.content.push({ type: 'text', text: 'extra' }) + + // The log is unchanged: deep-equal to the snapshot taken before mutation. + expect(session.events).toEqual(before) + // And a fresh derivation still reflects the original content. + expect(session.deriveMessages()[0]!.content).toEqual([{ type: 'text', text: 'original' }]) + }) }) describe('SessionStore', () => { diff --git a/scripts/publint-all.ts b/scripts/publint-all.ts index 45e5a3e0be..4b967e9c4b 100644 --- a/scripts/publint-all.ts +++ b/scripts/publint-all.ts @@ -15,6 +15,7 @@ const packages = [ 'packages/llm-pi-ai', 'packages/bash-local', 'packages/tool-bash', + 'packages/invariants', ] const root = resolve(import.meta.dirname, '..') diff --git a/tsconfig.base.json b/tsconfig.base.json index 8adcf0313d..6311883ba8 100644 --- a/tsconfig.base.json +++ b/tsconfig.base.json @@ -44,7 +44,8 @@ "@deepseek-ai/dsh-llm-deepseek": ["./packages/llm-deepseek/src"], "@deepseek-ai/dsh-llm-pi-ai": ["./packages/llm-pi-ai/src"], "@deepseek-ai/dsh-bash-local": ["./packages/bash-local/src"], - "@deepseek-ai/dsh-tool-bash": ["./packages/tool-bash/src"] + "@deepseek-ai/dsh-tool-bash": ["./packages/tool-bash/src"], + "@deepseek-ai/dsh-invariants": ["./packages/invariants/src"] } } } diff --git a/tsconfig.build.json b/tsconfig.build.json index d99cc777cd..7f62071870 100644 --- a/tsconfig.build.json +++ b/tsconfig.build.json @@ -20,6 +20,7 @@ { "path": "./packages/llm-deepseek" }, { "path": "./packages/llm-pi-ai" }, { "path": "./packages/bash-local" }, - { "path": "./packages/tool-bash" } + { "path": "./packages/tool-bash" }, + { "path": "./packages/invariants" } ] } diff --git a/tsconfig.typecheck.json b/tsconfig.typecheck.json index 0d36220af6..181f319dfd 100644 --- a/tsconfig.typecheck.json +++ b/tsconfig.typecheck.json @@ -26,7 +26,8 @@ "@deepseek-ai/dsh-llm-deepseek": ["./packages/llm-deepseek/src"], "@deepseek-ai/dsh-llm-pi-ai": ["./packages/llm-pi-ai/src"], "@deepseek-ai/dsh-bash-local": ["./packages/bash-local/src"], - "@deepseek-ai/dsh-tool-bash": ["./packages/tool-bash/src"] + "@deepseek-ai/dsh-tool-bash": ["./packages/tool-bash/src"], + "@deepseek-ai/dsh-invariants": ["./packages/invariants/src"] } }, "include": ["packages/*/src", "packages/*/tests", "examples", "scripts"] diff --git a/yarn.lock b/yarn.lock index 2b82f1a35a..a57f97dadc 100644 --- a/yarn.lock +++ b/yarn.lock @@ -611,6 +611,21 @@ __metadata: languageName: unknown linkType: soft +"@deepseek-ai/dsh-invariants@workspace:packages/invariants": + version: 0.0.0-use.local + resolution: "@deepseek-ai/dsh-invariants@workspace:packages/invariants" + dependencies: + "@deepseek-ai/dsh-agent": "npm:^0.0.1" + "@deepseek-ai/dsh-llm": "npm:^0.0.1" + "@deepseek-ai/dsh-session": "npm:^0.0.1" + cordis: "npm:^4.0.0-rc.6" + peerDependencies: + "@deepseek-ai/dsh-agent": ^0.0.1 + "@deepseek-ai/dsh-session": ^0.0.1 + cordis: ^4.0.0-rc.6 + languageName: unknown + linkType: soft + "@deepseek-ai/dsh-llm-deepseek@npm:^0.0.1, @deepseek-ai/dsh-llm-deepseek@workspace:packages/llm-deepseek": version: 0.0.0-use.local resolution: "@deepseek-ai/dsh-llm-deepseek@workspace:packages/llm-deepseek" From 89e63f143612fe4ee9bfa5efa5b9d32027af5b0a Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Sat, 13 Jun 2026 23:50:43 +0800 Subject: [PATCH 2/5] fix(invariants): address Codex review of dev invariants (PR 2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - HMR state soundness: inject sessions, rebuild per-session trace by replaying each existing session's log at (re-)apply, so a reload mid-turn no longer falsely rejects the next event - tighten nesting: turn/end rejects an open step; step/start rejects an open step; chunk/message/tool events must name the open turn+step; pendingCalls clears at step/end so a cross-step tool/result can't satisfy a stale call - drop the default export (it stripped the inject metadata when loaded by name; functional plugins expose named exports only — matches tool-bash) - document deepFreeze's top-down precondition; sync RFC 005/008 bodies to the as-implemented decision --- ...5-runtime-validation-and-error-taxonomy.md | 2 +- docs/rfc/008-immutable-public-surfaces.md | 2 + packages/invariants/README.md | 10 +- packages/invariants/src/index.ts | 109 ++++++++++++------ packages/invariants/tests/invariants.spec.ts | 85 +++++++++++++- 5 files changed, 167 insertions(+), 41 deletions(-) diff --git a/docs/rfc/005-runtime-validation-and-error-taxonomy.md b/docs/rfc/005-runtime-validation-and-error-taxonomy.md index a754c638ff..c3382abe22 100644 --- a/docs/rfc/005-runtime-validation-and-error-taxonomy.md +++ b/docs/rfc/005-runtime-validation-and-error-taxonomy.md @@ -14,7 +14,7 @@ Three gaps where compile-time guarantees stop: 1. **Schema validation in defineTool**: before `execute`, validate parsed args against the SchemaSpec (the converter already encodes the structure — a small interpreter walks it: presence of required keys, primitive type checks, enum membership, recursion into objects/arrays). On mismatch, return an `isError` ToolExecutionResult describing the violation — the model can self-correct. Raw-registered tools (MCP) keep validating their own input. 2. **Structured error taxonomy**: per-package error classes extending a common `HarnessError` (name, `code`, `cause` chaining). `ToolExecutionResult` gains optional `error: { name, code }` alongside the model-facing text. The loop's `errorData` consumes it; session `error` events carry the code. This also properly fixes the non-Error-throw message degradation found in review. -3. **Dev-mode invariants**: a `dsh-invariants` debug plugin (everything is a plugin — it's just listeners) asserting, when enabled: session seq strictly increases; `step/start` precedes its chunks; `turn/start`/`turn/end` pair and nest; tool/call has a matching tool/result; status transitions are legal. Enabled in tests and the demo; off in production. Doubles as executable documentation of the event contract. +3. **Dev-mode invariants**: a `dsh-invariants` debug plugin (everything is a plugin — it's just listeners) asserting, when enabled: session seq strictly increases; `step/start` precedes its chunks; `turn/start`/`turn/end` pair and nest; tool/call has a matching tool/result; status transitions are legal. Enabled in tests and the demo; off in production. Doubles as executable documentation of the event contract. _(As implemented, the tool rule is one-directional — a `tool/result` requires a prior `tool/call`, but NOT the converse: a throwing `tools/execute` waterfall ends a step with no result. See [ADR 0012](../adr/0012-dev-invariants-over-deep-readonly.md).)_ ## Plan diff --git a/docs/rfc/008-immutable-public-surfaces.md b/docs/rfc/008-immutable-public-surfaces.md index fae635a78e..263c274f01 100644 --- a/docs/rfc/008-immutable-public-surfaces.md +++ b/docs/rfc/008-immutable-public-surfaces.md @@ -8,6 +8,8 @@ The session log is append-only by contract, but `session.events` returns `readon ## Proposal +> **Implemented differently — see the Status line and [ADR 0012](../adr/0012-dev-invariants-over-deep-readonly.md).** The `DeepReadonly` design below was rejected as written (compile-only, high type-noise, castable). What shipped: an always-on deep clone in `deriveMessages` (closing the request/adapter aliasing path) plus a dev-mode `Object.freeze` + invariants plugin. The proposal text is kept for the record. + Make immutability part of the type where mutation is corruption: - `SessionEvent` data becomes `DeepReadonly` on the way OUT of a session (`events`, `session/event` listeners); `append()` keeps taking plain mutable input. A `DeepReadonly` utility type lands in dsh-llm next to the brand/never helpers. diff --git a/packages/invariants/README.md b/packages/invariants/README.md index ee12612006..3e7337fe11 100644 --- a/packages/invariants/README.md +++ b/packages/invariants/README.md @@ -6,14 +6,16 @@ Dev-mode event-contract invariants and session-log freeze. A pure-listener plugi ## Plugin -```ts -import Invariants from '@deepseek-ai/dsh-invariants' +A functional plugin — register the module namespace (this is what loading by name in `cordis.yml` does): -await ctx.plugin(Invariants) // freeze on (default) +```ts +import * as Invariants from '@deepseek-ai/dsh-invariants' + +await ctx.plugin(Invariants) // freeze on (default) await ctx.plugin(Invariants, { freeze: false }) // assert contract, don't freeze ``` -`inject`: none required — it listens on `session/created`, `session/event`, and `agent/status`, all emitted by services it does not depend on directly. +`inject`: `['sessions']` — it reads `ctx.sessions.list()` at apply time to rebuild trace state for sessions that already exist (so a hot reload mid-turn doesn't falsely reject the next event). It listens on `session/created`, `session/event`, and `agent/status`. ### Config diff --git a/packages/invariants/src/index.ts b/packages/invariants/src/index.ts index 4f04f70209..6824ea76af 100644 --- a/packages/invariants/src/index.ts +++ b/packages/invariants/src/index.ts @@ -24,6 +24,7 @@ import type { Agent, AgentStatus } from '@deepseek-ai/dsh-agent' import type { Session, SessionEvent } from '@deepseek-ai/dsh-session' export const name = 'invariants' +export const inject = ['sessions'] /** * Thrown when a harness event-contract invariant is violated. Plain `Error` @@ -55,11 +56,23 @@ interface SessionTrace { openTurn: number | null /** Open step within the current turn, or null between steps. */ openStep: number | null - /** Outstanding tool-call ids awaiting a result (a result needs a prior call). */ + /** + * Tool-call ids issued in the OPEN step awaiting a result. Cleared at + * `step/end` — a result must arrive in the same step as its call. + */ pendingCalls: Set } -/** Deep-freeze a value and everything reachable from it. Idempotent. */ +/** + * Deep-freeze a value and everything reachable from it. + * + * Sound because this is only ever called top-down on event objects we just + * appended: by the time a node is frozen, this same walk has already frozen + * its descendants, so a frozen node implies frozen descendants — skipping it + * is correct and avoids re-walking on HMR replay. (We never pass an + * externally shallow-frozen object, which is the only input that would make + * the early-return unsound.) + */ function deepFreeze(value: unknown): void { if (value === null || typeof value !== 'object') return if (Object.isFrozen(value)) return @@ -69,6 +82,15 @@ function deepFreeze(value: unknown): void { } } +/** Assert that a step-scoped event names the currently open turn and step. */ +function requireOpenStep(trace: SessionTrace, kind: string, turn: number, step: number): void { + if (trace.openTurn !== turn || trace.openStep !== step) { + throw new InvariantError( + `${kind} names turn ${turn}/step ${step} but open is turn ${trace.openTurn}/step ${trace.openStep}`, + ) + } +} + /** Assert one appended event against the per-session invariants. */ function checkEvent(trace: SessionTrace, event: SessionEvent): void { // seq is strictly monotonic — the spine of replay equivalence. lastSeq @@ -80,6 +102,8 @@ function checkEvent(trace: SessionTrace, event: SessionEvent): void { // Intentionally non-exhaustive: only events that carry ordering structure // are checked; the rest are trace/replay data with no nesting contract. + // SessionEventMap is merge-extensible, so no assertNever — unknown event + // types fall through untouched. // eslint-disable-next-line @typescript-eslint/switch-exhaustiveness-check switch (event.type) { case 'turn/start': { @@ -93,41 +117,50 @@ function checkEvent(trace: SessionTrace, event: SessionEvent): void { if (trace.openTurn !== event.data.turn) { throw new InvariantError(`turn/end ${event.data.turn} does not match open turn ${trace.openTurn}`) } + if (trace.openStep !== null) { + throw new InvariantError(`turn/end ${event.data.turn} while step ${trace.openStep} is still open`) + } trace.openTurn = null - trace.openStep = null break } case 'step/start': { if (trace.openTurn !== event.data.turn) { throw new InvariantError(`step/start in turn ${event.data.turn} but open turn is ${trace.openTurn}`) } + if (trace.openStep !== null) { + throw new InvariantError(`step/start ${event.data.step} while step ${trace.openStep} is still open`) + } trace.openStep = event.data.step break } case 'step/end': { - if (trace.openStep !== event.data.step) { - throw new InvariantError(`step/end ${event.data.step} does not match open step ${trace.openStep}`) - } + requireOpenStep(trace, 'step/end', event.data.turn, event.data.step) + // A result must arrive in the step that issued the call; orphan calls + // (a step that errored before its result) do not carry to the next step. + trace.pendingCalls.clear() trace.openStep = null break } case 'assistant/chunk': { - // A chunk belongs to an open step — step/start must precede it. - if (trace.openStep === null) { - throw new InvariantError('assistant/chunk outside an open step (step/start must precede its chunks)') - } + requireOpenStep(trace, 'assistant/chunk', event.data.turn, event.data.step) + break + } + case 'assistant/message': { + requireOpenStep(trace, 'assistant/message', event.data.turn, event.data.step) break } case 'tool/call': { + requireOpenStep(trace, 'tool/call', event.data.turn, event.data.step) trace.pendingCalls.add(event.data.callId) break } case 'tool/result': { - // A result needs a prior matching call. (The converse does NOT hold: a - // call may have no result — a thrown tools/execute waterfall ends the - // step with no tool/result, which is legal.) + requireOpenStep(trace, 'tool/result', event.data.turn, event.data.step) + // A result needs a prior matching call in the same step. (The converse + // does NOT hold: a call may have no result — a throwing tools/execute + // waterfall ends the step with no tool/result, which is legal.) if (!trace.pendingCalls.delete(event.data.callId)) { - throw new InvariantError(`tool/result for ${event.data.callId} with no prior tool/call`) + throw new InvariantError(`tool/result for ${event.data.callId} with no prior tool/call in this step`) } break } @@ -150,34 +183,46 @@ function checkTransition(from: AgentStatus | undefined, to: AgentStatus): void { } /** - * Register the dev-mode invariants. Returns nothing — contributions are - * effect-scoped, so disposing the plugin fiber removes all listeners and - * stops freezing (HMR-safe). + * Register the dev-mode invariants. Contributions are effect-scoped, so + * disposing the plugin fiber removes all listeners and stops freezing + * (HMR-safe). On (re-)apply the trace state is rebuilt by replaying each + * existing session's log, so a hot reload mid-turn does not falsely reject the + * next event. */ export function apply(ctx: Context, config: Config = {}): void { const freeze = config.freeze ?? true const traces = new WeakMap() + // Agent status has no stored history to replay; the first observation after + // (re-)apply seeds the baseline, so a reload never produces a false positive. const lastStatus = new WeakMap() - const traceFor = (session: Session): SessionTrace => { - let trace = traces.get(session) - if (!trace) { - trace = { lastSeq: -1, openTurn: null, openStep: null, pendingCalls: new Set() } - traces.set(session, trace) - } - return trace - } + const freshTrace = (): SessionTrace => ({ lastSeq: -1, openTurn: null, openStep: null, pendingCalls: new Set() }) - ctx.on('session/created', (session) => { - // A seeded/forked session arrives with events already in its log — the - // constructor copies the seed WITHOUT emitting session/event, so replay - // them through the checker here and freeze the existing entries. - const trace = traceFor(session) + /** Build (or rebuild) a session's trace by replaying its whole log; freeze it. */ + const seedSession = (session: Session): SessionTrace => { + const trace = freshTrace() + traces.set(session, trace) for (const event of session.events) { checkEvent(trace, event) if (freeze) deepFreeze(event) } - }) + return trace + } + + // Every store-created session (the only kind that emits session/event) is + // seeded first — via ctx.sessions.list() at apply or session/created — so + // the fallback is a defensive guard, never hit in practice. + /* v8 ignore next -- traceFor's fallback: session/event always follows a seed */ + const traceFor = (session: Session): SessionTrace => traces.get(session) ?? seedSession(session) + + // Rebuild state for sessions that already exist at (re-)apply time — HMR + // reload starts a fresh fiber, and a mid-turn session would otherwise look + // like it began with a stray chunk/step-end. + for (const session of ctx.sessions.list()) seedSession(session) + + // A newly created session may arrive seeded/forked (the constructor copies + // the seed WITHOUT emitting session/event), so replay its log here too. + ctx.on('session/created', (session) => { seedSession(session) }) ctx.on('session/event', (session, event) => { checkEvent(traceFor(session), event) @@ -189,5 +234,3 @@ export function apply(ctx: Context, config: Config = {}): void { lastStatus.set(agent, status) }) } - -export default apply diff --git a/packages/invariants/tests/invariants.spec.ts b/packages/invariants/tests/invariants.spec.ts index 3b7f88f6d4..7c8e383141 100644 --- a/packages/invariants/tests/invariants.spec.ts +++ b/packages/invariants/tests/invariants.spec.ts @@ -3,7 +3,8 @@ import { Context } from 'cordis' import { CallId } from '@deepseek-ai/dsh-llm' import type { Agent } from '@deepseek-ai/dsh-agent' import SessionStore from '@deepseek-ai/dsh-session' -import Invariants, { InvariantError } from '@deepseek-ai/dsh-invariants' +import * as Invariants from '@deepseek-ai/dsh-invariants' +import { InvariantError } from '@deepseek-ai/dsh-invariants' /** A Context with the session store and the invariants plugin registered. */ async function setup(config?: { freeze?: boolean }) { @@ -63,7 +64,7 @@ describe('session-log invariants', () => { 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 }) - expect(() => session.append('step/end', { turn: 1, step: 2 })).toThrow(/does not match open step 1/) + expect(() => session.append('step/end', { turn: 1, step: 2 })).toThrow(/open is turn 1\/step 1/) }) it('rejects an assistant/chunk outside an open step', async () => { @@ -71,7 +72,7 @@ describe('session-log invariants', () => { const session = ctx.sessions.create() session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) expect(() => session.append('assistant/chunk', { turn: 1, step: 1, chunk: { type: 'text-delta', index: 0, text: 'x' } })) - .toThrow(/outside an open step/) + .toThrow(/open is turn 1\/step null/) }) it('rejects a tool/result with no prior tool/call', async () => { @@ -114,6 +115,84 @@ describe('session-log invariants', () => { // b is a fresh session — its own turn/start must not see a's open turn. expect(() => b.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })).not.toThrow() }) + + it('accepts multiple steps in a turn and consecutive turns', async () => { + const { ctx } = await setup({ freeze: false }) + const session = ctx.sessions.create() + expect(() => { + session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) + session.append('step/start', { turn: 1, step: 1 }) + session.append('assistant/message', { turn: 1, step: 1, content: [] }) + session.append('step/end', { turn: 1, step: 1 }) + session.append('step/start', { turn: 1, step: 2 }) + session.append('assistant/message', { turn: 1, step: 2, content: [] }) + session.append('step/end', { turn: 1, step: 2 }) + session.append('turn/end', { turn: 1, reason: { kind: 'completed' } }) + session.append('turn/start', { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } }) + session.append('turn/end', { turn: 2, reason: { kind: 'completed' } }) + }).not.toThrow() + }) + + it('rejects a turn/end while a step is still open', async () => { + const { ctx } = await setup({ freeze: false }) + 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 }) + expect(() => session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })) + .toThrow(/while step 1 is still open/) + }) + + it('rejects a step/start while a step is still open', async () => { + const { ctx } = await setup({ freeze: false }) + 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 }) + expect(() => session.append('step/start', { turn: 1, step: 2 })).toThrow(/while step 1 is still open/) + }) + + it('rejects a tool/result satisfying a call from a previous step', async () => { + const { ctx } = await setup({ freeze: false }) + 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 }) + session.append('tool/call', { turn: 1, step: 1, callId: CallId('c1'), name: 'echo', arguments: '{}' }) + // step ends with the call unresolved — pendingCalls is cleared. + session.append('step/end', { turn: 1, step: 1 }) + session.append('step/start', { turn: 1, step: 2 }) + expect(() => session.append('tool/result', { turn: 1, step: 2, callId: CallId('c1'), content: [], isError: false })) + .toThrow(/no prior tool\/call in this step/) + }) + + it('rejects an assistant/message naming the wrong step', async () => { + const { ctx } = await setup({ freeze: false }) + 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 }) + expect(() => session.append('assistant/message', { turn: 1, step: 2, content: [] })) + .toThrow(/open is turn 1\/step 1/) + }) +}) + +describe('HMR state rebuild', () => { + it('rebuilds trace state for a session that exists at (re-)apply time', async () => { + const ctx = new Context() + await ctx.plugin(SessionStore) + // First registration, mid-turn: a turn is open when the plugin reloads. + const first = await ctx.plugin(Invariants, { freeze: false }) + 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 }) + await first.dispose() + + // Re-apply (HMR): the fresh fiber must replay the existing log so the open + // step is known — the next chunk must NOT be a false positive. + await ctx.plugin(Invariants, { freeze: false }) + expect(() => session.append('assistant/chunk', { turn: 1, step: 1, chunk: { type: 'text-delta', index: 0, text: 'h' } })) + .not.toThrow() + // And a genuine violation is still caught after the rebuild. + expect(() => session.append('turn/start', { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } })) + .toThrow(/turn 1 is still open/) + }) }) describe('dev-freeze', () => { From 2f6d3b8539c482bb0fc3ce845eaa569479788284 Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Sun, 14 Jun 2026 00:06:25 +0800 Subject: [PATCH 3/5] test: property-based tests for protocol-shaped code (RFC 001) Adds fast-check + one tests/properties.spec.ts per protocol-shaped package (llm/BlockAssembler, session, tools/schema DSL, agent-loop scheduling). The tools suite includes the RFC 001<->005 composition property (generated args satisfying a spec pass validateArgs), closing the validator/InferArgs drift risk from ADR 0011. Loop properties are deterministic (settle on agent/status, no sleeps). The BlockAssembler suite found a real bug on first run: a duplicate block-end at the same index overwrote an already-flushed block, so the streamed prefix disagreed with final blocks(). Fixed (first close wins, matching the existing straggler rule) + regression test. Graduates RFC 001 -> ADR 0013. --- docs/adr/0013-property-based-testing.md | 23 +++ docs/adr/README.md | 1 + docs/rfc/001-property-based-testing.md | 2 +- docs/rfc/README.md | 2 +- package.json | 1 + packages/agent-loop/tests/properties.spec.ts | 138 ++++++++++++++++++ packages/llm/src/assembler.ts | 5 + packages/llm/tests/assembler.spec.ts | 35 +++++ packages/llm/tests/properties.spec.ts | 125 ++++++++++++++++ packages/session/tests/properties.spec.ts | 110 ++++++++++++++ packages/tools/tests/properties.spec.ts | 144 +++++++++++++++++++ yarn.lock | 17 +++ 12 files changed, 601 insertions(+), 2 deletions(-) create mode 100644 docs/adr/0013-property-based-testing.md create mode 100644 packages/agent-loop/tests/properties.spec.ts create mode 100644 packages/llm/tests/properties.spec.ts create mode 100644 packages/session/tests/properties.spec.ts create mode 100644 packages/tools/tests/properties.spec.ts diff --git a/docs/adr/0013-property-based-testing.md b/docs/adr/0013-property-based-testing.md new file mode 100644 index 0000000000..3d9a7e924f --- /dev/null +++ b/docs/adr/0013-property-based-testing.md @@ -0,0 +1,23 @@ +# ADR 0013: Property-based testing for protocol-shaped code + +Status: accepted (2026-06-14) + +## Context + +Example-based tests pin the cases we thought of. The harness's core is protocol-shaped — chunk streams, event logs, schema conversion, inbox scheduling — where the input space is combinatorial and the interesting bugs live in interleavings nobody wrote an example for. The motivating evidence: a `streamBlocks` ordering bug once survived 100% line coverage of the happy paths. Per-file 100% coverage proves every line ran, not that every interleaving is correct. + +## Decision + +Adopt `fast-check` (a root devDependency) with one `tests/properties.spec.ts` per protocol-shaped package, generators tuned for *realistic-but-adversarial* inputs (not uniform noise) and `numRuns` kept so the suite stays well under ~10s locally. Failures print a reproducible seed. + +- **dsh-llm / BlockAssembler:** arbitrary chunk streams (valid + malformed: duplicate indices, stragglers, missing block-start). Invariants: `flushReady()+flushRemaining() ≡ blocks()` in order; the streamed prefix is always a prefix of final `blocks()`; partial count ≤ distinct indices; re-assembly idempotent. +- **dsh-session:** arbitrary event logs. Invariants: `deriveMessages` deterministic; replay-from-seed identical; seq strictly monotonic; non-message events never affect derived history; derived content is decoupled from the log. +- **dsh-tools:** arbitrary `SchemaSpec`. Invariants: JSON Schema `required` equals the `required:true` keys at every level; conversion total; **and the RFC 001↔005 composition** — generated args satisfying a spec pass `validateArgs`, and targeted corruptions (dropped required key, non-object top level) are rejected. This closes the validator/`InferArgs` drift risk from ADR 0011. +- **dsh-agent-loop:** arbitrary send schedules against a never-exhausting adapter, driven through the `agent/status` settle signal (no wall-clock sleeps). Invariants: no message lost; turn numbers strictly increase; status transitions stay on the legal machine. + +## Consequences + +- Generator quality is the value lever — the generators bias toward small index pools and short strings so collisions and interleavings are common. +- **It already paid off:** the BlockAssembler stream found a real bug — a duplicate `block-end` at the same index overwrote an already-flushed block, so the streamed prefix disagreed with final `blocks()`. Fixed (first close wins, matching the existing straggler rule) with a dedicated regression test. +- A property flake from a timeout is a finding, not something to retry away. The loop properties are deterministic by construction (settle on `agent/status`), so a hang is a real defect. +- Property tests supplement, not replace, the example tests that pin specific branches for the 100%-coverage gate. diff --git a/docs/adr/README.md b/docs/adr/README.md index ee3ebde6c6..1842301e9e 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -24,3 +24,4 @@ Do NOT write an ADR for: a mechanical or local choice (a variable name, a one-fi | [0010](0010-twin-llm-adapters.md) | Two LLM adapters as a design-verification twin | accepted | | [0011](0011-runtime-arg-validation.md) | Runtime arg validation at the model boundary | accepted | | [0012](0012-dev-invariants-over-deep-readonly.md) | Dev-mode invariants over compile-time deep-readonly | accepted | +| [0013](0013-property-based-testing.md) | Property-based testing for protocol-shaped code | accepted | diff --git a/docs/rfc/001-property-based-testing.md b/docs/rfc/001-property-based-testing.md index cb519129de..17795ebfd7 100644 --- a/docs/rfc/001-property-based-testing.md +++ b/docs/rfc/001-property-based-testing.md @@ -1,6 +1,6 @@ # RFC 001: Property-based testing for protocol-shaped code -Status: proposed +Status: implemented — see [ADR 0013](../adr/0013-property-based-testing.md). (It found a real BlockAssembler duplicate-`block-end` bug on first run.) ## Problem diff --git a/docs/rfc/README.md b/docs/rfc/README.md index 4760bb98f6..fbbe442d31 100644 --- a/docs/rfc/README.md +++ b/docs/rfc/README.md @@ -4,7 +4,7 @@ Proposals for substantial future work — reviewed before implementation, unlike | # | Title | Status | |---|---|---| -| [001](001-property-based-testing.md) | Property-based testing for protocol-shaped code | proposed | +| [001](001-property-based-testing.md) | Property-based testing for protocol-shaped code | implemented | | [002](002-mutation-testing.md) | Mutation testing as the coverage counterweight | proposed | | [003](003-deterministic-and-stress-testing.md) | Deterministic tests + replay invariant fixture + race stress | proposed | | [004](004-architectural-conformance.md) | Architectural rules: dependency-cruiser, adapter conformance kit | proposed | diff --git a/package.json b/package.json index f4c3279082..4f752ac364 100644 --- a/package.json +++ b/package.json @@ -32,6 +32,7 @@ "@vitest/coverage-v8": "^4.1.8", "@yarnpkg/types": "^4.0.1", "eslint": "^10.4.1", + "fast-check": "^4.8.0", "knip": "^6.16.1", "lefthook": "^2.1.9", "publint": "^0.3.21", diff --git a/packages/agent-loop/tests/properties.spec.ts b/packages/agent-loop/tests/properties.spec.ts new file mode 100644 index 0000000000..8e6768d402 --- /dev/null +++ b/packages/agent-loop/tests/properties.spec.ts @@ -0,0 +1,138 @@ +/** + * Property-based tests for the agent loop's inbox/turn scheduling (RFC 001 → + * ADR 0013). Deterministic by construction: schedules are driven through the + * `agent/status` settle signal (no wall-clock sleeps), so a flake is a finding, + * not timing noise. + * + * Invariants: every sent message appears exactly once in the log (none lost); + * turn numbers strictly increase; status transitions follow the legal machine + * idle→running→idle (and →disposed at teardown). + */ + +import { describe, expect, it } from 'vitest' +import { Context } from 'cordis' +import LlmService from '@deepseek-ai/dsh-llm' +import type { GenerateOptions, StreamChunk } from '@deepseek-ai/dsh-llm' +import { LlmAdapter } from '@deepseek-ai/dsh-llm' +import SessionStore from '@deepseek-ai/dsh-session' +import SystemPrompt from '@deepseek-ai/dsh-system-prompt' +import ToolRegistry from '@deepseek-ai/dsh-tools' +import AgentRegistry from '@deepseek-ai/dsh-agent' +import AgentLoop, { type LoopAgent } from '@deepseek-ai/dsh-agent-loop' +import fc from 'fast-check' + +/** A never-exhausting adapter: every model call returns the same short reply. */ +class EchoAdapter extends LlmAdapter { + async * stream(options: GenerateOptions): AsyncIterable { + if (options.signal?.aborted) throw new Error('aborted') + const text = 'ok' + yield { type: 'block-start', index: 0, blockType: 'text' } + yield { type: 'text-delta', index: 0, text } + yield { type: 'block-end', index: 0, block: { type: 'text', text } } + yield { type: 'usage', usage: { inputTokens: 1, outputTokens: 1 } } + yield { type: 'finish', reason: { kind: 'stop' } } + } +} + +async function harness() { + const ctx = new Context() + await ctx.plugin(LlmService) + await ctx.plugin(SessionStore) + await ctx.plugin(SystemPrompt) + await ctx.plugin(ToolRegistry) + await ctx.plugin(AgentRegistry) + await ctx.plugin(AgentLoop, { agents: [] }) + ctx.llm.registerAdapter(['mock'], new EchoAdapter()) + return ctx +} + +/** Resolve on the agent's next transition to idle (event-based, not polled). */ +function nextIdle(ctx: Context, agent: LoopAgent): Promise { + return new Promise((resolve) => { + const dispose = ctx.on('agent/status', (subject, status) => { + if (subject === agent && status === 'idle') { + dispose() + resolve() + } + }) + }) +} + +/** Record every status transition for the legal-machine assertion. */ +function recordStatus(ctx: Context, agent: LoopAgent): string[] { + const seen: string[] = [] + ctx.on('agent/status', (subject, status) => { + if (subject === agent) seen.push(status) + }) + return seen +} + +function userMessageTexts(agent: LoopAgent): string[] { + return agent.session.events + .filter(e => e.type === 'user/message') + .map(e => (e.data as { content: { type: string; text?: string }[] }).content.map(b => b.text ?? '').join('')) +} + +function turnNumbers(agent: LoopAgent): number[] { + return agent.session.events + .filter(e => e.type === 'turn/start') + .map(e => (e.data as { turn: number }).turn) +} + +/** Assert a status trace is a legal run: idle/running alternating, ending idle. */ +function assertLegalStatusTrace(trace: string[]): void { + for (let i = 1; i < trace.length; i++) { + expect(trace[i]).not.toBe(trace[i - 1]) // no repeats (setStatus dedups) + } + for (const s of trace) expect(['idle', 'running']).toContain(s) +} + +describe('agent loop scheduling properties', () => { + it('a synchronous burst loses no message and uses strictly increasing turns', async () => { + await fc.assert(fc.asyncProperty( + fc.array(fc.string({ minLength: 1 }), { minLength: 1, maxLength: 6 }), + async (texts) => { + const ctx = await harness() + try { + const agent = ctx.agentLoop.create('a', { model: 'mock' }) + const trace = recordStatus(ctx, agent) + const idle = nextIdle(ctx, agent) + // Send all in one synchronous tick: they queue before the loop wakes. + for (const text of texts) agent.send([{ type: 'text', text }]) + await idle + + // No message lost: every send appears as a user/message, in order. + expect(userMessageTexts(agent)).toEqual(texts) + // Turn numbers strictly increase. + const turns = turnNumbers(agent) + for (let i = 1; i < turns.length; i++) expect(turns[i]!).toBeGreaterThan(turns[i - 1]!) + assertLegalStatusTrace(trace) + } finally { + await ctx.fiber.dispose() + } + }, + ), { numRuns: 25 }) + }) + + it('sequential sends each get their own turn with increasing numbers', async () => { + await fc.assert(fc.asyncProperty( + fc.array(fc.string({ minLength: 1 }), { minLength: 1, maxLength: 5 }), + async (texts) => { + const ctx = await harness() + try { + const agent = ctx.agentLoop.create('a', { model: 'mock' }) + for (const text of texts) { + const idle = nextIdle(ctx, agent) + agent.send([{ type: 'text', text }]) + await idle + } + // Each send was drained at a separate turn start: N turns, 1..N. + expect(turnNumbers(agent)).toEqual(texts.map((_, i) => i + 1)) + expect(userMessageTexts(agent)).toEqual(texts) + } finally { + await ctx.fiber.dispose() + } + }, + ), { numRuns: 20 }) + }) +}) diff --git a/packages/llm/src/assembler.ts b/packages/llm/src/assembler.ts index e18071d7e9..a61d6cf044 100644 --- a/packages/llm/src/assembler.ts +++ b/packages/llm/src/assembler.ts @@ -72,6 +72,11 @@ export class BlockAssembler { } case 'block-end': { const partial = this.ensure(chunk.index, chunk.block.type) + // First close wins: a second block-end for an already-closed index is + // a straggler (same rule as post-close deltas). Ignoring it keeps the + // streamed prefix and the final blocks() in agreement — otherwise a + // re-close could rewrite a block already flushed downstream. + if (partial.block) return partial.block = chunk.block return chunk.block } diff --git a/packages/llm/tests/assembler.spec.ts b/packages/llm/tests/assembler.spec.ts index e02e547e99..24639a04a2 100644 --- a/packages/llm/tests/assembler.spec.ts +++ b/packages/llm/tests/assembler.spec.ts @@ -166,3 +166,38 @@ describe('assertNever', () => { .toThrow('unreachable variant in BlockAssembler.push') }) }) + +describe('BlockAssembler regressions (property-test findings)', () => { + it('first block-end wins: a duplicate block-end for a closed index is ignored', () => { + // Found by fast-check (RFC 001): two block-ends at the same index made the + // streamed prefix (first block) disagree with final blocks() (second + // block). The first close must win — same straggler rule as post-close + // deltas — so streaming and one-shot assembly stay identical. + const chunks: StreamChunk[] = [ + { type: 'block-end', index: 0, block: { type: 'reasoning', text: 'first' } }, + { type: 'block-end', index: 0, block: { type: 'text', text: 'second' } }, + ] + const streaming = new BlockAssembler() + const flushed = [] + for (const chunk of chunks) { + streaming.push(chunk) + flushed.push(...streaming.flushReady()) + } + flushed.push(...streaming.flushRemaining()) + + const oneShot = new BlockAssembler() + for (const chunk of chunks) oneShot.push(chunk) + + expect(flushed).toEqual([{ type: 'reasoning', text: 'first' }]) + expect(oneShot.blocks()).toEqual([{ type: 'reasoning', text: 'first' }]) + expect(flushed).toEqual(oneShot.blocks()) + }) + + it('push returns undefined for a duplicate block-end (it closed nothing)', () => { + const a = new BlockAssembler() + expect(a.push({ type: 'block-end', index: 0, block: { type: 'text', text: 'x' } })) + .toEqual({ type: 'text', text: 'x' }) + expect(a.push({ type: 'block-end', index: 0, block: { type: 'text', text: 'y' } })) + .toBeUndefined() + }) +}) diff --git a/packages/llm/tests/properties.spec.ts b/packages/llm/tests/properties.spec.ts new file mode 100644 index 0000000000..60c546aa39 --- /dev/null +++ b/packages/llm/tests/properties.spec.ts @@ -0,0 +1,125 @@ +/** + * Property-based tests for the BlockAssembler (RFC 001 → ADR 0013). + * + * The assembler is protocol-shaped: arbitrary interleavings of block-start, + * deltas, block-end, usage, and finish — valid and malformed (duplicate + * indices, stragglers after block-end, missing block-start, delta-only). The + * invariants below are the contract the agent loop and LlmService rely on. + */ + +import { describe, expect, it } from 'vitest' +import fc from 'fast-check' +import { BlockAssembler } from '@deepseek-ai/dsh-llm' +import type { ContentBlock, StreamChunk } from '@deepseek-ai/dsh-llm' +import { CallId } from '@deepseek-ai/dsh-llm' + +// A small pool of indices so collisions (duplicate-index bugs) are common. +const indexArb = fc.integer({ min: 0, max: 4 }) + +const blockEndArb = (index: number): fc.Arbitrary => fc.oneof( + fc.record({ text: fc.string() }).map((r): StreamChunk => ( + { type: 'block-end', index, block: { type: 'text', text: r.text } } + )), + fc.record({ text: fc.string() }).map((r): StreamChunk => ( + { type: 'block-end', index, block: { type: 'reasoning', text: r.text } } + )), + fc.record({ id: fc.string({ minLength: 1 }), name: fc.string(), args: fc.string() }).map((r): StreamChunk => ( + { type: 'block-end', index, block: { type: 'tool-call', id: CallId(r.id), name: r.name, arguments: r.args } } + )), +) + +/** One arbitrary chunk over the small index pool — valid and malformed mixes. */ +const chunkArb: fc.Arbitrary = indexArb.chain(index => fc.oneof( + fc.constant({ type: 'block-start', index, blockType: 'text' }), + fc.constant({ type: 'block-start', index, blockType: 'reasoning' }), + fc.constant({ type: 'block-start', index, blockType: 'tool-call' }), + fc.string().map((text): StreamChunk => ({ type: 'text-delta', index, text })), + fc.string().map((text): StreamChunk => ({ type: 'reasoning-delta', index, text })), + fc.record({ id: fc.string({ minLength: 1 }), argumentsDelta: fc.string() }) + .map((r): StreamChunk => ({ type: 'tool-call-delta', index, id: CallId(r.id), argumentsDelta: r.argumentsDelta })), + blockEndArb(index), + fc.constant({ type: 'usage', usage: { inputTokens: 1, outputTokens: 1 } }), +)) + +/** A stream is a list of chunks; we add the terminal `finish` ourselves. */ +const streamArb = fc.array(chunkArb, { maxLength: 30 }) + +/** Feed a fresh assembler, return it. */ +function feed(chunks: StreamChunk[]): BlockAssembler { + const a = new BlockAssembler() + for (const chunk of chunks) a.push(chunk) + return a +} + +describe('BlockAssembler properties', () => { + it('flushReady() ++ flushRemaining() === blocks(), in order', () => { + fc.assert(fc.property(streamArb, (chunks) => { + const streaming = new BlockAssembler() + const flushed: ContentBlock[] = [] + for (const chunk of chunks) { + streaming.push(chunk) + flushed.push(...streaming.flushReady()) + } + flushed.push(...streaming.flushRemaining()) + + const oneShot = feed(chunks).blocks() + expect(flushed).toEqual(oneShot) + })) + }) + + it('streamBlocks-style flush never yields a block before an earlier open one', () => { + // flushReady is strict-order: once it stops at an open index, no later + // index may be emitted until that one closes. We assert the flushed prefix + // is always a prefix of the final blocks() order. + fc.assert(fc.property(streamArb, (chunks) => { + const streaming = new BlockAssembler() + const flushed: ContentBlock[] = [] + for (const chunk of chunks) { + streaming.push(chunk) + flushed.push(...streaming.flushReady()) + } + const finalSoFar = streaming.blocks() + // Everything flushed mid-stream is a prefix of the full ordered blocks. + expect(finalSoFar.slice(0, flushed.length)).toEqual(flushed) + })) + }) + + it('partials map size never exceeds the number of distinct indices seen', () => { + fc.assert(fc.property(streamArb, (chunks) => { + const distinct = new Set() + for (const chunk of chunks) { + if ('index' in chunk) distinct.add(chunk.index) + } + const a = feed(chunks) + // blocks() length equals the number of distinct indices that became + // partials (block-bearing chunks). It can never exceed distinct indices. + expect(a.blocks().length).toBeLessThanOrEqual(distinct.size) + })) + }) + + it('re-assembly is idempotent: blocks() is stable across repeated calls', () => { + fc.assert(fc.property(streamArb, (chunks) => { + const a = feed(chunks) + expect(a.blocks()).toEqual(a.blocks()) + // And message().content mirrors blocks(). + expect(a.message().content).toEqual(a.blocks()) + })) + }) + + it('blocks() never throws and yields only valid content-block tags', () => { + fc.assert(fc.property(streamArb, (chunks) => { + const blocks = feed(chunks).blocks() + for (const block of blocks) { + expect(['text', 'reasoning', 'tool-call', 'tool-result', 'image']).toContain(block.type) + } + })) + }) + + it('result().finish defaults to stop when no finish chunk arrives', () => { + fc.assert(fc.property(streamArb, (chunks) => { + const a = feed(chunks) + const hasFinish = chunks.some(c => c.type === 'finish') + if (!hasFinish) expect(a.finish).toEqual({ kind: 'stop' }) + })) + }) +}) diff --git a/packages/session/tests/properties.spec.ts b/packages/session/tests/properties.spec.ts new file mode 100644 index 0000000000..2b31d6e83b --- /dev/null +++ b/packages/session/tests/properties.spec.ts @@ -0,0 +1,110 @@ +/** + * Property-based tests for the Session event log (RFC 001 → ADR 0013). + * + * Generates arbitrary event logs and asserts the derivation invariants the + * agent loop and replay depend on: deriveMessages is deterministic and + * replay-from-seed reproduces it; seq is strictly monotonic; non-message + * events never affect derived history. + */ + +import { describe, expect, it } from 'vitest' +import fc from 'fast-check' +import { CallId } from '@deepseek-ai/dsh-llm' +import { Session, SessionId } from '@deepseek-ai/dsh-session' +import type { SessionEventMap, SessionEventType } from '@deepseek-ai/dsh-session' + +type Appendable = { [T in SessionEventType]: { type: T; data: SessionEventMap[T] } }[SessionEventType] + +const textContentArb = fc.array( + fc.record({ type: fc.constant<'text'>('text'), text: fc.string() }), + { maxLength: 3 }, +) + +// A message-producing event (these DO affect derived history). +const messageEventArb: fc.Arbitrary = fc.oneof( + textContentArb.map((content): Appendable => ({ type: 'user/message', data: { content, source: { kind: 'user' } } })), + textContentArb.map((content): Appendable => ({ type: 'assistant/message', data: { turn: 1, step: 1, content } })), + fc.record({ id: fc.string({ minLength: 1 }), content: textContentArb, isError: fc.boolean() }) + .map((r): Appendable => ({ type: 'tool/result', data: { turn: 1, step: 1, callId: CallId(r.id), content: r.content, isError: r.isError } })), +) + +// A non-message event (trace/replay data — must NOT affect derived history). +const nonMessageEventArb: fc.Arbitrary = fc.oneof( + fc.constant({ type: 'turn/start', data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } }), + fc.constant({ type: 'turn/end', data: { turn: 1, reason: { kind: 'completed' } } }), + fc.constant({ type: 'step/start', data: { turn: 1, step: 1 } }), + fc.constant({ type: 'step/end', data: { turn: 1, step: 1 } }), + fc.string().map((text): Appendable => ({ type: 'assistant/chunk', data: { turn: 1, step: 1, chunk: { type: 'text-delta', index: 0, text } } })), + fc.constant({ type: 'usage', data: { turn: 1, step: 1, usage: { inputTokens: 1, outputTokens: 1 } } }), + fc.constant({ type: 'error', data: { turn: 1, step: 1, message: 'x' } }), +) + +const anyEventArb = fc.oneof(messageEventArb, nonMessageEventArb) +const logArb = fc.array(anyEventArb, { maxLength: 25 }) + +let counter = 0 +function build(events: Appendable[]): Session { + const session = new Session(SessionId(`prop-${counter++}`)) + for (const e of events) session.append(e.type, e.data) + return session +} + +describe('Session properties', () => { + it('deriveMessages is deterministic (same log → identical derivation)', () => { + fc.assert(fc.property(logArb, (events) => { + const a = build(events) + expect(a.deriveMessages()).toEqual(a.deriveMessages()) + })) + }) + + it('seq is strictly monotonic and zero-based contiguous', () => { + fc.assert(fc.property(logArb, (events) => { + const session = build(events) + session.events.forEach((event, i) => { expect(event.seq).toBe(i) }) + expect(session.seq).toBe(events.length) + })) + }) + + it('replay-from-seed reproduces the derivation identically', () => { + fc.assert(fc.property(logArb, (events) => { + const original = build(events) + const replayed = new Session(SessionId(`replay-${counter++}`), [...original.events]) + expect(replayed.deriveMessages()).toEqual(original.deriveMessages()) + expect(replayed.seq).toBe(original.seq) + })) + }) + + it('non-message events never affect derived history', () => { + fc.assert(fc.property( + fc.array(messageEventArb, { maxLength: 12 }), + fc.array(nonMessageEventArb, { maxLength: 12 }), + (messages, noise) => { + // The same message events, with and without interleaved noise, derive + // the same history (noise is inserted at arbitrary positions). + const clean = build(messages).deriveMessages() + const interleaved: Appendable[] = [] + const maxLen = Math.max(messages.length, noise.length) + for (let i = 0; i < maxLen; i++) { + if (i < noise.length) interleaved.push(noise[i]!) + if (i < messages.length) interleaved.push(messages[i]!) + } + const withNoise = build(interleaved).deriveMessages() + expect(withNoise).toEqual(clean) + }, + )) + }) + + it('every derived message has a known role and decoupled content', () => { + fc.assert(fc.property(logArb, (events) => { + const session = build(events) + const messages = session.deriveMessages() + const before = structuredClone(session.events) + for (const m of messages) { + expect(['user', 'assistant', 'system']).toContain(m.role) + // Mutating derived content must not touch the log (append-only). + m.content.push({ type: 'text', text: 'mutation' }) + } + expect(session.events).toEqual(before) + })) + }) +}) diff --git a/packages/tools/tests/properties.spec.ts b/packages/tools/tests/properties.spec.ts new file mode 100644 index 0000000000..a5d44baf2d --- /dev/null +++ b/packages/tools/tests/properties.spec.ts @@ -0,0 +1,144 @@ +/** + * Property-based tests for the tool-schema DSL (RFC 001 → ADR 0013), including + * the RFC 001 ↔ 005 composition: generated args that satisfy a SchemaSpec must + * pass validateArgs, and targeted corruptions must be rejected. This closes the + * validator/InferArgs drift risk noted in ADR 0011. + */ + +import { describe, expect, it } from 'vitest' +import fc from 'fast-check' +import { schemaSpecToJsonSchema, validateArgs } from '@deepseek-ai/dsh-tools' +import type { SchemaProp, SchemaSpec } from '@deepseek-ai/dsh-tools' + +// A leaf prop arbitrary (no nesting) with optional required/enum. +function leafPropArb(): fc.Arbitrary { + return fc.oneof( + fc.record({ required: fc.boolean() }).map(({ required }): SchemaProp => ({ type: 'string', ...required ? { required: true } : {} })), + fc.record({ required: fc.boolean() }).map(({ required }): SchemaProp => ({ type: 'number', ...required ? { required: true } : {} })), + fc.record({ required: fc.boolean() }).map(({ required }): SchemaProp => ({ type: 'boolean', ...required ? { required: true } : {} })), + fc.record({ values: fc.uniqueArray(fc.string({ minLength: 1 }), { minLength: 1, maxLength: 3 }), required: fc.boolean() }) + .map(({ values, required }): SchemaProp => ({ type: 'string', enum: values, ...required ? { required: true } : {} })), + ) +} + +/** A prop arbitrary up to `depth` levels of nesting (objects and arrays). */ +function propArb(depth: number): fc.Arbitrary { + if (depth <= 0) return leafPropArb() + return fc.oneof( + { weight: 3, arbitrary: leafPropArb() }, + { + weight: 1, + arbitrary: fc.record({ properties: specArb(depth - 1), required: fc.boolean() }) + .map(({ properties, required }): SchemaProp => ({ type: 'object', properties, ...required ? { required: true } : {} })), + }, + { + weight: 1, + arbitrary: fc.record({ items: propArb(depth - 1), required: fc.boolean() }) + .map(({ items, required }): SchemaProp => ({ type: 'array', items, ...required ? { required: true } : {} })), + }, + ) +} + +function specArb(depth: number): fc.Arbitrary { + return fc.dictionary(fc.string({ minLength: 1, maxLength: 6 }), propArb(depth), { maxKeys: 4 }) +} + +/** Generate a value that satisfies a prop (used to build valid args). */ +function valueForProp(prop: SchemaProp): fc.Arbitrary { + switch (prop.type) { + case 'string': return prop.enum ? fc.constantFrom(...prop.enum) : fc.string() + case 'number': return fc.double({ noNaN: true }) + case 'boolean': return fc.boolean() + case 'object': return prop.properties ? validArgsForSpec(prop.properties) : fc.constant({}) + case 'array': return prop.items ? fc.array(valueForProp(prop.items), { maxLength: 3 }) : fc.constant([]) + } +} + +/** Generate args satisfying every required key of a spec (optionals included randomly). */ +function validArgsForSpec(spec: SchemaSpec): fc.Arbitrary> { + const entries = Object.entries(spec) + return fc.tuple(...entries.map(([key, prop]) => + fc.tuple( + fc.constant(key), + // required keys are always present; optional keys are present ~half the time + prop.required === true + ? valueForProp(prop).map(v => ({ include: true, value: v })) + : fc.oneof( + valueForProp(prop).map(v => ({ include: true, value: v })), + fc.constant({ include: false, value: undefined }), + ), + ), + )).map((pairs) => { + const out: Record = {} + for (const [key, { include, value }] of pairs) if (include) out[key] = value + return out + }) +} + +/** Collect the `required: true` keys at the top level of a spec. */ +function requiredKeys(spec: SchemaSpec): string[] { + return Object.entries(spec).filter(([, p]) => p.required === true).map(([k]) => k) +} + +describe('schema DSL properties', () => { + it('JSON Schema `required` equals the required:true keys at every level', () => { + fc.assert(fc.property(specArb(2), (spec) => { + const checkLevel = (s: SchemaSpec, json: { required?: string[]; properties: Record }) => { + expect(new Set(json.required ?? [])).toEqual(new Set(requiredKeys(s))) + for (const [key, prop] of Object.entries(s)) { + const propJson = json.properties[key] as Record + if (prop.type === 'object' && prop.properties) { + checkLevel(prop.properties, propJson as { required?: string[]; properties: Record }) + } + } + } + checkLevel(spec, schemaSpecToJsonSchema(spec)) + })) + }) + + it('conversion is total (never throws) for any spec', () => { + fc.assert(fc.property(specArb(3), (spec) => { + expect(() => schemaSpecToJsonSchema(spec)).not.toThrow() + })) + }) + + it('validateArgs is total (never throws) for any spec and any input', () => { + fc.assert(fc.property(specArb(2), fc.anything(), (spec, args) => { + expect(() => validateArgs(spec, args)).not.toThrow() + })) + }) + + it('RFC 001↔005: args satisfying the spec pass validateArgs', () => { + fc.assert(fc.property( + specArb(2).chain(spec => fc.tuple(fc.constant(spec), validArgsForSpec(spec))), + ([spec, args]) => { + expect(validateArgs(spec, args)).toEqual([]) + }, + )) + }) + + it('RFC 001↔005: dropping a required key is always rejected', () => { + fc.assert(fc.property( + specArb(1) + .filter(spec => requiredKeys(spec).length > 0) + .chain(spec => fc.tuple(fc.constant(spec), validArgsForSpec(spec))), + ([spec, args]) => { + const required = requiredKeys(spec) + const victim = required[0]! + const broken = Object.fromEntries(Object.entries(args).filter(([k]) => k !== victim)) + const violations = validateArgs(spec, broken) + expect(violations.some(v => v.includes(`"${victim}"`))).toBe(true) + }, + )) + }) + + it('RFC 001↔005: a non-object top level is always rejected', () => { + fc.assert(fc.property( + specArb(1), + fc.oneof(fc.string(), fc.integer(), fc.boolean(), fc.constant(null), fc.array(fc.anything())), + (spec, notAnObject) => { + expect(validateArgs(spec, notAnObject).length).toBeGreaterThan(0) + }, + )) + }) +}) diff --git a/yarn.lock b/yarn.lock index a57f97dadc..ee8d65477f 100644 --- a/yarn.lock +++ b/yarn.lock @@ -673,6 +673,7 @@ __metadata: "@vitest/coverage-v8": "npm:^4.1.8" "@yarnpkg/types": "npm:^4.0.1" eslint: "npm:^10.4.1" + fast-check: "npm:^4.8.0" knip: "npm:^6.16.1" lefthook: "npm:^2.1.9" publint: "npm:^0.3.21" @@ -2829,6 +2830,15 @@ __metadata: languageName: node linkType: hard +"fast-check@npm:^4.8.0": + version: 4.8.0 + resolution: "fast-check@npm:4.8.0" + dependencies: + pure-rand: "npm:^8.0.0" + checksum: 10c0/f72556a29db4ff386a8b6e50d420b06c7e5eaafff7db5560a99136c57d8d4777998155eb02d1bbeff396f575cc0b1442c8a1c4ddb798c4a919b542de1a1904ff + languageName: node + linkType: hard + "fast-deep-equal@npm:^3.1.1, fast-deep-equal@npm:^3.1.3": version: 3.1.3 resolution: "fast-deep-equal@npm:3.1.3" @@ -4048,6 +4058,13 @@ __metadata: languageName: node linkType: hard +"pure-rand@npm:^8.0.0": + version: 8.4.0 + resolution: "pure-rand@npm:8.4.0" + checksum: 10c0/6414bbc1c6f45fb774173431c7205e79783b77cfae0e2145e741b6999363554dbd2f4210d2a5bc08683e0b2f6823198c9308766b1d0911e1dccd7beb8842f860 + languageName: node + linkType: hard + "quansync@npm:^1.0.0": version: 1.0.0 resolution: "quansync@npm:1.0.0" From 7b07b70750bbf7bed3410e1e58a9db57c8896166 Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Sun, 14 Jun 2026 00:24:23 +0800 Subject: [PATCH 4/5] test: address Codex review of property tests (PR 3) - llm: generator now emits finish chunks (the finish-defaults property was vacuously green); add a property asserting streaming and one-shot assembly agree on usage and finish - agent-loop: assert the synchronous burst batches into exactly one turn; add a mixed-schedule property (send/settle interleavings); recordStatus returns its disposer; per-run timeouts so a hang loses no seed - session: randomize the noise/message interleaving (was a fixed alternation) - tools: exclude non-finite doubles from generated numeric args (JSON-real) --- packages/agent-loop/tests/properties.spec.ts | 58 ++++++++++++++++---- packages/llm/tests/properties.spec.ts | 33 +++++++++-- packages/session/tests/properties.spec.ts | 21 ++++--- packages/tools/tests/properties.spec.ts | 2 +- 4 files changed, 91 insertions(+), 23 deletions(-) diff --git a/packages/agent-loop/tests/properties.spec.ts b/packages/agent-loop/tests/properties.spec.ts index 8e6768d402..23c0f77b46 100644 --- a/packages/agent-loop/tests/properties.spec.ts +++ b/packages/agent-loop/tests/properties.spec.ts @@ -58,13 +58,14 @@ function nextIdle(ctx: Context, agent: LoopAgent): Promise { }) } -/** Record every status transition for the legal-machine assertion. */ -function recordStatus(ctx: Context, agent: LoopAgent): string[] { +/** Record every status transition for the legal-machine assertion. Returns + * the seen list plus a disposer for the listener (per the registry convention). */ +function recordStatus(ctx: Context, agent: LoopAgent): { seen: string[]; dispose: () => void } { const seen: string[] = [] - ctx.on('agent/status', (subject, status) => { + const dispose = ctx.on('agent/status', (subject, status) => { if (subject === agent) seen.push(status) }) - return seen + return { seen, dispose } } function userMessageTexts(agent: LoopAgent): string[] { @@ -95,7 +96,7 @@ describe('agent loop scheduling properties', () => { const ctx = await harness() try { const agent = ctx.agentLoop.create('a', { model: 'mock' }) - const trace = recordStatus(ctx, agent) + const { seen: trace } = recordStatus(ctx, agent) const idle = nextIdle(ctx, agent) // Send all in one synchronous tick: they queue before the loop wakes. for (const text of texts) agent.send([{ type: 'text', text }]) @@ -103,15 +104,14 @@ describe('agent loop scheduling properties', () => { // No message lost: every send appears as a user/message, in order. expect(userMessageTexts(agent)).toEqual(texts) - // Turn numbers strictly increase. - const turns = turnNumbers(agent) - for (let i = 1; i < turns.length; i++) expect(turns[i]!).toBeGreaterThan(turns[i - 1]!) + // A synchronous burst batches into exactly one turn. + expect(turnNumbers(agent)).toEqual([1]) assertLegalStatusTrace(trace) } finally { await ctx.fiber.dispose() } }, - ), { numRuns: 25 }) + ), { numRuns: 25, timeout: 2000 }) }) it('sequential sends each get their own turn with increasing numbers', async () => { @@ -133,6 +133,44 @@ describe('agent loop scheduling properties', () => { await ctx.fiber.dispose() } }, - ), { numRuns: 20 }) + ), { numRuns: 20, timeout: 2000 }) + }) + + it('mixed schedule (send, optionally settle) loses no message and orders turns', async () => { + // Each step is a (text, settle?) pair: settle=true awaits idle before the + // next send (own turn); settle=false sends in the same tick (batches). + const stepArb = fc.record({ text: fc.string({ minLength: 1 }), settle: fc.boolean() }) + await fc.assert(fc.asyncProperty( + fc.array(stepArb, { minLength: 1, maxLength: 6 }), + async (steps) => { + const ctx = await harness() + try { + const agent = ctx.agentLoop.create('a', { model: 'mock' }) + // Capture an idle waiter before EACH send; the last one is guaranteed + // to resolve because the final send always triggers (or joins) a turn + // that ends idle. Awaiting an already-resolved waiter is a no-op, so a + // trailing settle step can't cause a hang. + let lastIdle: Promise | undefined + for (const step of steps) { + const idle = nextIdle(ctx, agent) + lastIdle = idle + agent.send([{ type: 'text', text: step.text }]) + if (step.settle) await idle + } + await lastIdle + + // No message lost or reordered, regardless of batching. + expect(userMessageTexts(agent)).toEqual(steps.map(s => s.text)) + // Turn numbers are a strictly increasing 1..N prefix (N = turn count). + const turns = turnNumbers(agent) + expect(turns).toEqual(turns.map((_, i) => i + 1)) + // Every message landed in some turn; turns never exceed messages. + expect(turns.length).toBeLessThanOrEqual(steps.length) + expect(turns.length).toBeGreaterThanOrEqual(1) + } finally { + await ctx.fiber.dispose() + } + }, + ), { numRuns: 25, timeout: 3000 }) }) }) diff --git a/packages/llm/tests/properties.spec.ts b/packages/llm/tests/properties.spec.ts index 60c546aa39..f0118d9dfd 100644 --- a/packages/llm/tests/properties.spec.ts +++ b/packages/llm/tests/properties.spec.ts @@ -39,9 +39,12 @@ const chunkArb: fc.Arbitrary = indexArb.chain(index => fc.oneof( .map((r): StreamChunk => ({ type: 'tool-call-delta', index, id: CallId(r.id), argumentsDelta: r.argumentsDelta })), blockEndArb(index), fc.constant({ type: 'usage', usage: { inputTokens: 1, outputTokens: 1 } }), + fc.constant({ type: 'finish', reason: { kind: 'stop' } }), + fc.constant({ type: 'finish', reason: { kind: 'tool-calls' } }), + fc.string().map((message): StreamChunk => ({ type: 'finish', reason: { kind: 'error', message } })), )) -/** A stream is a list of chunks; we add the terminal `finish` ourselves. */ +/** A stream is an arbitrary list of chunks (we do NOT force a terminal finish). */ const streamArb = fc.array(chunkArb, { maxLength: 30 }) /** Feed a fresh assembler, return it. */ @@ -115,11 +118,33 @@ describe('BlockAssembler properties', () => { })) }) - it('result().finish defaults to stop when no finish chunk arrives', () => { + it('finish reflects the last finish chunk, or defaults to stop when none arrives', () => { fc.assert(fc.property(streamArb, (chunks) => { const a = feed(chunks) - const hasFinish = chunks.some(c => c.type === 'finish') - if (!hasFinish) expect(a.finish).toEqual({ kind: 'stop' }) + const finishes = chunks.filter(c => c.type === 'finish') + if (finishes.length === 0) { + expect(a.finish).toEqual({ kind: 'stop' }) + } else { + // last-write-wins: the assembler keeps the most recent finish reason. + const last = finishes[finishes.length - 1] + if (last?.type === 'finish') expect(a.finish).toEqual(last.reason) + } + })) + }) + + it('streaming and one-shot assembly agree on usage and finish', () => { + fc.assert(fc.property(streamArb, (chunks) => { + // Streaming consumer: push + flush as it goes. + const streaming = new BlockAssembler() + for (const chunk of chunks) { + streaming.push(chunk) + streaming.flushReady() + } + streaming.flushRemaining() + // One-shot consumer: push all, then read. + const oneShot = feed(chunks) + expect(streaming.usage).toEqual(oneShot.usage) + expect(streaming.finish).toEqual(oneShot.finish) })) }) }) diff --git a/packages/session/tests/properties.spec.ts b/packages/session/tests/properties.spec.ts index 2b31d6e83b..5c4e210c52 100644 --- a/packages/session/tests/properties.spec.ts +++ b/packages/session/tests/properties.spec.ts @@ -74,19 +74,24 @@ describe('Session properties', () => { })) }) - it('non-message events never affect derived history', () => { + it('non-message events never affect derived history (any interleaving)', () => { fc.assert(fc.property( fc.array(messageEventArb, { maxLength: 12 }), fc.array(nonMessageEventArb, { maxLength: 12 }), - (messages, noise) => { - // The same message events, with and without interleaved noise, derive - // the same history (noise is inserted at arbitrary positions). + // An arbitrary merge of the two streams that PRESERVES each stream's + // relative order (a random interleaving, not a fixed alternation). + fc.infiniteStream(fc.boolean()), + (messages, noise, pick) => { const clean = build(messages).deriveMessages() const interleaved: Appendable[] = [] - const maxLen = Math.max(messages.length, noise.length) - for (let i = 0; i < maxLen; i++) { - if (i < noise.length) interleaved.push(noise[i]!) - if (i < messages.length) interleaved.push(messages[i]!) + let mi = 0 + let ni = 0 + const picker = pick[Symbol.iterator]() + while (mi < messages.length || ni < noise.length) { + // take from noise when chosen and available, else from messages + const takeNoise = ni < noise.length && (mi >= messages.length || picker.next().value === true) + if (takeNoise) { interleaved.push(noise[ni]!); ni++ } + else { interleaved.push(messages[mi]!); mi++ } } const withNoise = build(interleaved).deriveMessages() expect(withNoise).toEqual(clean) diff --git a/packages/tools/tests/properties.spec.ts b/packages/tools/tests/properties.spec.ts index a5d44baf2d..49cb3cb640 100644 --- a/packages/tools/tests/properties.spec.ts +++ b/packages/tools/tests/properties.spec.ts @@ -47,7 +47,7 @@ function specArb(depth: number): fc.Arbitrary { function valueForProp(prop: SchemaProp): fc.Arbitrary { switch (prop.type) { case 'string': return prop.enum ? fc.constantFrom(...prop.enum) : fc.string() - case 'number': return fc.double({ noNaN: true }) + case 'number': return fc.double({ noNaN: true, noDefaultInfinity: true }) case 'boolean': return fc.boolean() case 'object': return prop.properties ? validArgsForSpec(prop.properties) : fc.constant({}) case 'array': return prop.items ? fc.array(valueForProp(prop.items), { maxLength: 3 }) : fc.constant([]) From a45bc8da67cb5d49d10de606e1fe1614174b2cc3 Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Sun, 14 Jun 2026 10:36:16 +0800 Subject: [PATCH 5/5] fix(invariants): deepFreeze walks already-frozen objects' descendants MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Session.append accepts event data from arbitrary plugins/tools, so a caller can pass a SHALLOW-frozen object with mutable descendants. The old Object.isFrozen early-return skipped such an object entirely, leaving its descendants mutable in the log — exactly the history mutation ADR 0012 means to catch. Now always descend, tracking visited objects in a WeakSet for cycle-termination and idempotence. Addresses PR review finding. --- packages/invariants/src/index.ts | 22 +++++++++------ packages/invariants/tests/invariants.spec.ts | 29 ++++++++++++++------ 2 files changed, 34 insertions(+), 17 deletions(-) diff --git a/packages/invariants/src/index.ts b/packages/invariants/src/index.ts index 6824ea76af..c5d9b30d00 100644 --- a/packages/invariants/src/index.ts +++ b/packages/invariants/src/index.ts @@ -66,19 +66,23 @@ interface SessionTrace { /** * Deep-freeze a value and everything reachable from it. * - * Sound because this is only ever called top-down on event objects we just - * appended: by the time a node is frozen, this same walk has already frozen - * its descendants, so a frozen node implies frozen descendants — skipping it - * is correct and avoids re-walking on HMR replay. (We never pass an - * externally shallow-frozen object, which is the only input that would make - * the early-return unsound.) + * Walks every object's own properties even when the object itself is already + * frozen: `Session.append()` accepts event data from arbitrary plugins/tools, + * so a caller can hand us a SHALLOW-frozen object whose descendants are still + * mutable. Skipping an already-frozen node (the obvious idempotence shortcut) + * would leave exactly the kind of mutable history ADR 0012 means to catch. A + * `WeakSet` of visited objects keeps it terminating on cycles and avoids + * re-walking shared subtrees / already-processed seed events. */ -function deepFreeze(value: unknown): void { +function deepFreeze(value: unknown, seen: WeakSet = new WeakSet()): void { if (value === null || typeof value !== 'object') return - if (Object.isFrozen(value)) return + if (seen.has(value)) return + seen.add(value) + // Freeze the node (no-op if a caller pre-froze it), then ALWAYS descend — + // a frozen container can still hold mutable children. Object.freeze(value) for (const key of Object.keys(value)) { - deepFreeze((value as Record)[key]) + deepFreeze((value as Record)[key], seen) } } diff --git a/packages/invariants/tests/invariants.spec.ts b/packages/invariants/tests/invariants.spec.ts index 7c8e383141..0a621434ab 100644 --- a/packages/invariants/tests/invariants.spec.ts +++ b/packages/invariants/tests/invariants.spec.ts @@ -222,16 +222,29 @@ describe('dev-freeze', () => { expect(Object.isFrozen(session.events[0])).toBe(true) }) - it('is idempotent over already-frozen sub-structures', async () => { + it('freezes mutable descendants of a shallow-frozen event datum', async () => { const { ctx } = await setup() const session = ctx.sessions.create() - // Pre-freeze a block before appending; deepFreeze must short-circuit on it - // (the already-frozen guard) while still freezing the enclosing event. - const block = Object.freeze({ type: 'text' as const, text: 'pre-frozen' }) - const event = session.append('user/message', { content: [block], source: { kind: 'user' } }) - expect(Object.isFrozen(event)).toBe(true) - expect(Object.isFrozen(event.data.content)).toBe(true) - expect(Object.isFrozen(event.data.content[0])).toBe(true) + // A caller hands in a SHALLOW-frozen block whose nested array is still + // mutable. deepFreeze must descend into the already-frozen object and + // freeze the descendant, not short-circuit on the frozen container — + // otherwise dev-mode misses exactly the history mutation ADR 0012 catches. + const innerContent: { type: 'text'; text: string }[] = [{ type: 'text', text: 'inner' }] + const block = Object.freeze({ type: 'tool-result' as const, toolCallId: CallId('c1'), content: innerContent, isError: false }) + session.append('user/message', { content: [block], source: { kind: 'user' } }) + expect(Object.isFrozen(block.content)).toBe(true) + expect(Object.isFrozen(block.content[0])).toBe(true) + expect(() => { block.content.push({ type: 'text', text: 'mutation' }) }).toThrow() + }) + + it('terminates on a cyclic event datum (WeakSet guard)', async () => { + const { ctx } = await setup() + const session = ctx.sessions.create() + // A self-referential structure must not loop forever. + const cyclic: Record = { type: 'text', text: 'x' } + cyclic['self'] = cyclic + expect(() => session.append('user/message', { content: [cyclic as never], source: { kind: 'user' } })).not.toThrow() + expect(Object.isFrozen(cyclic)).toBe(true) }) })