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', () => {