invariants: a loop-built request must be exactly what the log reconstructs
The dev-mode cross-check on llm/stream: a frozen request with a live sessionId (the loop-built marker; hand-built one-shots stay unfrozen and skipped) must carry messages deep-equal to the derivation over the log prefix strictly before the in-flight step's step/start, and header fields equal to the fold of the log's request/header* events. The messages side rebuilds through a FRESH Session over the boundary prefix — same projection code, zero shared state — so the live cache cannot vouch for itself, and the seq-bounded rebuild is boundary- correct: content appended after step/start (an agent/request-window inject) legitimately belongs to the next request and cannot false-fire the check. prepend:true only defends against the replay adapter's short-circuit (append-registered); correctness never rests on listener ordering. There is no divergence-allowance caveat: nothing can shape request content outside the log.
This commit is contained in:
Tianyi Cui
@@ -40,6 +40,10 @@ 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.
|
||||
|
||||
Model requests (on `llm/stream`):
|
||||
|
||||
- **a loop-built request is exactly what the log reconstructs** — a frozen request with a live `sessionId` (the loop-built marker; hand-built one-shots like compaction's summarize are unfrozen and skipped) must carry frozen `messages` deep-equal to the derivation over the log prefix strictly before the in-flight step's `step/start` (rebuilt through a FRESH `Session`, so the live cache cannot vouch for itself — and boundary-correct: content logged after `step/start` legitimately belongs to the next request), and every non-content field must equal the fold of the log's `request/header*` events (see [the reconstructability RFC](../../../docs/rfc/implemented/architecture/2026-07-05-reconstructable-requests.md)). Registered with `prepend: true` so a short-circuiting `llm/stream` listener (the replay adapter) cannot silence it; prepend orders it against append-registered listeners only — correctness rests on the seq-bounded rebuild, never listener timing.
|
||||
|
||||
On any violation it throws `InvariantError` (`code: 'INVARIANT'`).
|
||||
|
||||
## Why runtime, not deep-readonly types
|
||||
|
||||
@@ -21,9 +21,10 @@
|
||||
|
||||
import type { Context } from 'cordis'
|
||||
import { HarnessError } from '@deepseek-ai/dsh-llm'
|
||||
import type { CallId } from '@deepseek-ai/dsh-llm'
|
||||
import type { CallId, GenerateOptions } from '@deepseek-ai/dsh-llm'
|
||||
import type { Agent, AgentStatus } from '@deepseek-ai/dsh-agent'
|
||||
import type { Session, SessionEvent, SurfaceEventType } from '@deepseek-ai/dsh-session'
|
||||
import { Session, SessionId, foldRequestHeader } from '@deepseek-ai/dsh-session'
|
||||
import type { SessionEvent, SurfaceEventType } from '@deepseek-ai/dsh-session'
|
||||
|
||||
export const name = 'invariants'
|
||||
export const inject = ['sessions']
|
||||
@@ -360,4 +361,74 @@ export function apply(ctx: Context, config: Config = {}): void {
|
||||
checkTransition(lastStatus.get(agent), status)
|
||||
lastStatus.set(agent, status)
|
||||
})
|
||||
|
||||
// Request-reconstruction cross-check (the reconstructability RFC): a
|
||||
// loop-built request — frozen envelope + live sessionId is the marker; a
|
||||
// hand-built one-shot (compaction summarize) is unfrozen and skipped — must
|
||||
// be EXACTLY what the session log reconstructs:
|
||||
//
|
||||
// - messages: the derivation over the log prefix strictly before the
|
||||
// in-flight step's `step/start` (the reconstruction boundary). Compared
|
||||
// against a FRESH Session built over that prefix — the same projection
|
||||
// code with zero shared state, so the live cache under test cannot vouch
|
||||
// for itself. Boundary-correct by construction: content appended after
|
||||
// the boundary (an `agent/request`-window inject) is legitimately absent
|
||||
// from this request, and a current-surface comparison would false-fire.
|
||||
// - header: every non-content field must equal the fold of the log's
|
||||
// `request/header*` events — the loop logs the header event BEFORE
|
||||
// dispatch, so the fold already covers this request.
|
||||
//
|
||||
// Registered with `prepend: true` so a short-circuiting llm/stream listener
|
||||
// (the replay adapter returns its chunks without calling next()) cannot
|
||||
// silence the check by registering first. Prepend beats APPEND-registered
|
||||
// listeners only — two prepended listeners have no defined mutual order
|
||||
// (cordis unshift) — which is fine: correctness rests on the seq-bounded
|
||||
// fold below, never on listener timing.
|
||||
ctx.on('llm/stream', (options: GenerateOptions, next) => {
|
||||
if (options.sessionId === undefined || !Object.isFrozen(options)) return next()
|
||||
// GenerateOptions types sessionId as Branded<'SessionId'>, which IS
|
||||
// SessionId (dsh-llm cannot import it without a cycle) — no cast needed.
|
||||
const session = ctx.sessions.get(options.sessionId)
|
||||
if (!session) return next()
|
||||
if (!Object.isFrozen(options.messages)) {
|
||||
throw new InvariantError('a loop-built request must carry a frozen messages array')
|
||||
}
|
||||
|
||||
const events = session.events
|
||||
// seq === index (checked above), so the last step/start's seq bounds the
|
||||
// prefix directly. The in-flight step's step/start is necessarily the
|
||||
// last one: the loop cannot open another step while this call streams.
|
||||
let boundary = -1
|
||||
for (let i = events.length - 1; i >= 0; i -= 1) {
|
||||
if (events[i]?.type === 'step/start') {
|
||||
boundary = i
|
||||
break
|
||||
}
|
||||
}
|
||||
if (boundary === -1) {
|
||||
throw new InvariantError('a loop-built request with no step/start in its session log')
|
||||
}
|
||||
const rebuilt = new Session(SessionId(`${String(session.id)}-invariant-rebuild`), structuredClone(events.slice(0, boundary)))
|
||||
// JSON equality is sound here: both sides are structuredClones produced by
|
||||
// the same projection code path, so key insertion order matches when the
|
||||
// values do.
|
||||
if (JSON.stringify(options.messages) !== JSON.stringify(rebuilt.deriveMessages())) {
|
||||
throw new InvariantError(`llm request for session "${String(session.id)}" diverges from the boundary derivation (log-reconstruction desync)`)
|
||||
}
|
||||
|
||||
const header = foldRequestHeader(events)
|
||||
if (header === undefined) {
|
||||
throw new InvariantError('a loop-built request with no request/header event in its session log')
|
||||
}
|
||||
const headerMatches = options.model === header.config.model
|
||||
&& options.system === header.system
|
||||
&& options.temperature === header.config.temperature
|
||||
&& options.maxTokens === header.config.maxTokens
|
||||
&& JSON.stringify(options.stop) === JSON.stringify(header.config.stop)
|
||||
&& JSON.stringify(options.tools ?? []) === JSON.stringify(header.tools ?? [])
|
||||
if (!headerMatches) {
|
||||
throw new InvariantError(`llm request for session "${String(session.id)}" diverges from the folded request header`)
|
||||
}
|
||||
return next()
|
||||
}, { prepend: true })
|
||||
}
|
||||
|
||||
@@ -670,3 +670,113 @@ describe('surface invariants', () => {
|
||||
.toThrow(/cannot carry surfaceOp/)
|
||||
})
|
||||
})
|
||||
|
||||
describe('request-reconstruction cross-check (llm/stream)', () => {
|
||||
/** Session with a boundary: one derivable user message, an open step, and the header event the loop would have logged. */
|
||||
async function requestSetup() {
|
||||
const { ctx } = await setup({ freeze: false })
|
||||
const session = ctx.sessions.create(SessionId('req-check'))
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
|
||||
const boundary = session.deriveMessages()
|
||||
session.append('step/start', { turn: 1, step: 1 })
|
||||
session.append('request/header', { header: { config: { model: 'm' } }, reason: 'initial' })
|
||||
return { ctx, session, boundary }
|
||||
}
|
||||
|
||||
/** Dispatch the llm/stream waterfall with a stub core, collecting the check's verdict. */
|
||||
function dispatch(ctx: Context, options: unknown): void {
|
||||
// The invariants listener runs synchronously at dispatch time (its checks
|
||||
// precede next()); the stub core just yields nothing.
|
||||
void ctx.waterfall('llm/stream', options as never, () => (async function* () {})() as never)
|
||||
}
|
||||
|
||||
it('passes a frozen request that equals the boundary derivation + the folded header', async () => {
|
||||
const { ctx, session, boundary } = await requestSetup()
|
||||
const options = Object.freeze({ model: 'm', messages: Object.freeze(boundary), sessionId: session.id })
|
||||
expect(() => { dispatch(ctx, options) }).not.toThrow()
|
||||
})
|
||||
|
||||
it('is boundary-correct: content logged after step/start is legitimately absent from this request', async () => {
|
||||
const { ctx, session, boundary } = await requestSetup()
|
||||
// An agent/request-window inject: lands in the log after the boundary,
|
||||
// belongs to the NEXT request. A current-surface comparison would
|
||||
// false-fire here; the seq-bounded rebuild must not.
|
||||
session.append('context/message', { content: [{ type: 'text', text: '[late]' }], source: { kind: 'plugin', plugin: 'x' } }, { surfaceOp: 'append' })
|
||||
const options = Object.freeze({ model: 'm', messages: Object.freeze(boundary), sessionId: session.id })
|
||||
expect(() => { dispatch(ctx, options) }).not.toThrow()
|
||||
})
|
||||
|
||||
it('rejects a frozen request whose messages diverge from the boundary derivation', async () => {
|
||||
const { ctx, session, boundary } = await requestSetup()
|
||||
const messages = [...boundary, { role: 'user', content: [{ type: 'text', text: 'phantom' }] }]
|
||||
const options = Object.freeze({ model: 'm', messages: Object.freeze(messages), sessionId: session.id })
|
||||
expect(() => { dispatch(ctx, options) }).toThrow(/diverges from the boundary derivation/)
|
||||
})
|
||||
|
||||
it('rejects a frozen request whose fields diverge from the folded header', async () => {
|
||||
const { ctx, session, boundary } = await requestSetup()
|
||||
const options = Object.freeze({ model: 'other', messages: Object.freeze(boundary), sessionId: session.id })
|
||||
expect(() => { dispatch(ctx, options) }).toThrow(/diverges from the folded request header/)
|
||||
})
|
||||
|
||||
it('rejects a loop-built request with no header event or no step/start in its log', async () => {
|
||||
const { ctx } = await setup({ freeze: false })
|
||||
const session = ctx.sessions.create(SessionId('req-bare'))
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
const bare = Object.freeze({ model: 'm', messages: Object.freeze([]), sessionId: session.id })
|
||||
expect(() => { dispatch(ctx, bare) }).toThrow(/no step\/start/)
|
||||
|
||||
session.append('step/start', { turn: 1, step: 1 })
|
||||
expect(() => { dispatch(ctx, bare) }).toThrow(/no request\/header event/)
|
||||
})
|
||||
|
||||
it('rejects a frozen request carrying an unfrozen messages array', async () => {
|
||||
const { ctx, session, boundary } = await requestSetup()
|
||||
const options = Object.freeze({ model: 'm', messages: [...boundary], sessionId: session.id })
|
||||
expect(() => { dispatch(ctx, options) }).toThrow(/frozen messages array/)
|
||||
})
|
||||
|
||||
it('skips hand-built (unfrozen) requests — compaction summarize is out of scope', async () => {
|
||||
const { ctx, session } = await requestSetup()
|
||||
// Unfrozen envelope + arbitrary messages: a direct one-shot call.
|
||||
const options = { model: 'summarizer', messages: [{ role: 'user', content: [{ type: 'text', text: 'summarize!' }] }], sessionId: session.id }
|
||||
expect(() => { dispatch(ctx, options) }).not.toThrow()
|
||||
})
|
||||
|
||||
it('skips requests without a sessionId or with an unknown session', async () => {
|
||||
const { ctx } = await requestSetup()
|
||||
expect(() => { dispatch(ctx, Object.freeze({ model: 'm', messages: Object.freeze([]) })) }).not.toThrow()
|
||||
expect(() => { dispatch(ctx, Object.freeze({ model: 'm', messages: Object.freeze([]), sessionId: SessionId('ghost') })) }).not.toThrow()
|
||||
})
|
||||
})
|
||||
|
||||
describe('request cross-check ordering (prepend)', () => {
|
||||
it('runs ahead of a short-circuiting llm/stream listener registered before it', async () => {
|
||||
// The replay adapter returns its chunks WITHOUT calling next(), which
|
||||
// would silence a later-registered check — snapshot compositions load
|
||||
// replay before the app bundle that loads invariants. The check prepends,
|
||||
// so it fires ahead of append-registered listeners regardless of load
|
||||
// order. (Prepend orders it against APPENDED listeners only; correctness
|
||||
// rests on the seq-bounded rebuild, not on listener timing.)
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
ctx.on('llm/stream', () => (async function* () {})() as never) // short-circuits, no next()
|
||||
await ctx.plugin(Invariants, { freeze: false })
|
||||
|
||||
const session = ctx.sessions.create(SessionId('prepend-check'))
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
|
||||
session.append('step/start', { turn: 1, step: 1 })
|
||||
session.append('request/header', { header: { config: { model: 'm' } }, reason: 'initial' })
|
||||
|
||||
const divergent = Object.freeze({
|
||||
model: 'm',
|
||||
messages: Object.freeze([{ role: 'user', content: [{ type: 'text', text: 'phantom' }] }]),
|
||||
sessionId: session.id,
|
||||
})
|
||||
expect(() => {
|
||||
void ctx.waterfall('llm/stream', divergent as never, () => (async function* () {})() as never)
|
||||
}).toThrow(/diverges from the boundary derivation/)
|
||||
})
|
||||
})
|
||||
|
||||
Reference in New Issue
Block a user