fix(invariants): address Codex review of dev invariants (PR 2)

- 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
This commit is contained in:
Tianyi Cui
2026-06-13 23:50:43 +08:00
parent 11a29fdefe
commit 89e63f1436
5 changed files with 167 additions and 41 deletions
@@ -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
@@ -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<T>` 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<T>` utility type lands in dsh-llm next to the brand/never helpers.
+6 -4
View File
@@ -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
+76 -33
View File
@@ -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<string>
}
/** 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<Session, SessionTrace>()
// 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<Agent, AgentStatus>()
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
+82 -3
View File
@@ -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', () => {