diff --git a/packages/support/invariants/README.md b/packages/support/invariants/README.md index b26170d48e..94c2682f9d 100644 --- a/packages/support/invariants/README.md +++ b/packages/support/invariants/README.md @@ -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 diff --git a/packages/support/invariants/src/index.ts b/packages/support/invariants/src/index.ts index 08d4365f4b..f35fbafa1d 100644 --- a/packages/support/invariants/src/index.ts +++ b/packages/support/invariants/src/index.ts @@ -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 }) } diff --git a/packages/support/invariants/tests/invariants.spec.ts b/packages/support/invariants/tests/invariants.spec.ts index d7e95c1103..489cfb9817 100644 --- a/packages/support/invariants/tests/invariants.spec.ts +++ b/packages/support/invariants/tests/invariants.spec.ts @@ -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/) + }) +})