Coder/deepseek-harness
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(invariants): harden runtime contracts and gates

Browse Source
This commit is contained in:
Tianyi Cui
2026-07-21 00:25:38 +08:00
parent a4be4b5e52
commit 0c08e34a4a
27 changed files with 288 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
packages/core/agent-loop/README.md
View File

@@ -29,7 +29,7 @@ The config-driven `ctx.agentLoop.create()` path keeps its agent owned by the loo
### Invariant companion
The optional `@deepseek-ai/dsh-agent-loop/invariant` companion registers request reconstruction with `ctx.invariants`. For each frozen loop-built request carrying a live session id, it independently rebuilds the message boundary and folded request header from the session log; direct one-shot calls remain outside this marker contract.
The optional `@deepseek-ai/dsh-agent-loop/invariant` companion registers request reconstruction with `ctx.invariants`. The loop marks each request with an internal non-enumerable identity before freezing it; the companion then requires a live session and independently rebuilds the message boundary and folded request header from the log. Direct one-shot calls remain outside this contract even when callers freeze them or attach a session id.
### Configuration (schemastery)

7
packages/core/agent-loop/src/invariant.ts
View File

@@ -7,6 +7,7 @@ import type { Context } from 'cordis'
import type { GenerateOptions } from '@deepseek-ai/dsh-llm'
import type { InvariantFailure, InvariantInstaller } from '@deepseek-ai/dsh-invariants'
import { Session, SessionId, foldRequestHeader } from '@deepseek-ai/dsh-session'
import { isLoopRequest } from './request-marker.ts'
const PACKAGE_NAME = '@deepseek-ai/dsh-agent-loop'
@@ -20,9 +21,11 @@ const install: InvariantInstaller = Object.assign((ctx: Context, fail: Invariant
// Prepend prevents a short-circuiting replay listener from silencing the
// check; correctness itself comes from the sequence-bounded reconstruction.
ctx.on('llm/stream', (options: GenerateOptions, next) => {
if (options.sessionId === undefined || !Object.isFrozen(options)) return next()
if (!isLoopRequest(options)) return next()
if (!Object.isFrozen(options)) fail('a loop-built request must be frozen')
if (options.sessionId === undefined) fail('a loop-built request must carry a session id')
const session = ctx.sessions.get(options.sessionId)
if (!session) return next()
if (!session) fail(`a loop-built request must carry a live session id, got "${String(options.sessionId)}"`)
if (!Object.isFrozen(options.messages)) {
fail('a loop-built request must carry a frozen messages array')
}

5
packages/core/agent-loop/src/loop.ts
View File

@@ -15,6 +15,7 @@ import { canonicalHeader } from '@deepseek-ai/dsh-session'
import type { Session, TurnEndReason, TurnTrigger } from '@deepseek-ai/dsh-session'
import { createTransmissionLog, recordRequestHeader } from './request-log.ts'
import type { TransmissionLog } from './request-log.ts'
import { markLoopRequest } from './request-marker.ts'
import { renderPrompt } from '@deepseek-ai/dsh-system-prompt'
import type { PromptAssembly } from '@deepseek-ai/dsh-system-prompt'
import type {} from '@deepseek-ai/dsh-tools'
@@ -619,7 +620,7 @@ async function runStep(
recordRequestHeader(session, transmission, header)
// Freeze the logged header plus boundary snapshot; the prefix precedes derived history.
const request: GenerateOptions = deepFreeze({
const request: GenerateOptions = deepFreeze(markLoopRequest({
provider: header.config.provider,
model: header.config.model,
messages: [...header.messagePrefix ?? [], ...boundaryMessages],
@@ -630,7 +631,7 @@ async function runStep(
...header.config.stop !== undefined ? { stop: header.config.stop } : {},
sessionId: session.id,
signal,
})
}))
// --- Model call (streaming-first; raw chunks are the replay record) ---
const assembler = new BlockAssembler()

22
packages/core/agent-loop/src/request-marker.ts Normal file
View File

@@ -0,0 +1,22 @@
/** Internal identity shared by the independently bundled loop and invariant companion. */
const LOOP_REQUEST = Symbol.for('@deepseek-ai/dsh-agent-loop/request')
/**
* Mark a request as owned by the agent loop before it is frozen.
* @param request - mutable request object being assembled by the loop.
* @returns the same request with a non-enumerable loop identity.
*/
export function markLoopRequest<T extends object>(request: T): T {
Object.defineProperty(request, LOOP_REQUEST, { value: true })
return request
}
/**
* Test whether a request carries the agent loop's internal identity.
* @param request - request observed at the LLM stream boundary.
* @returns whether the loop marked this exact request object.
*/
export function isLoopRequest(request: object): boolean {
return Reflect.get(request, LOOP_REQUEST) === true
}

47
packages/core/agent-loop/tests/invariant.spec.ts
View File

@@ -3,6 +3,7 @@ import { Context } from 'cordis'
import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
import InvariantService from '@deepseek-ai/dsh-invariants'
import * as AgentLoopInvariant from '@deepseek-ai/dsh-agent-loop/invariant'
import { markLoopRequest } from '../src/request-marker.ts'
async function setup(): Promise<Context> {
const ctx = new Context()
@@ -16,6 +17,10 @@ function dispatch(ctx: Context, options: unknown): void {
void ctx.waterfall('llm/stream', options as never, () => (async function* () {})() as never)
}
function loopRequest<T extends object>(options: T): Readonly<T> {
return Object.freeze(markLoopRequest(options))
}
async function requestSetup() {
const ctx = await setup()
const session = ctx.sessions.create(SessionId('req-check'))
@@ -30,14 +35,14 @@ async function requestSetup() {
describe('request-reconstruction invariant', () => {
it('accepts a frozen request equal to the boundary derivation and folded header', async () => {
const { ctx, session, boundary } = await requestSetup()
const options = Object.freeze({ model: 'm', messages: Object.freeze(boundary), sessionId: session.id })
const options = loopRequest({ model: 'm', messages: Object.freeze(boundary), sessionId: session.id })
expect(() => { dispatch(ctx, options) }).not.toThrow()
})
it('uses the step boundary rather than content appended afterward', async () => {
const { ctx, session, boundary } = await requestSetup()
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 })
const options = loopRequest({ model: 'm', messages: Object.freeze(boundary), sessionId: session.id })
expect(() => { dispatch(ctx, options) }).not.toThrow()
})
@@ -45,20 +50,20 @@ describe('request-reconstruction invariant', () => {
const { ctx, session, boundary } = await requestSetup()
const prefix = { role: 'user' as const, content: [{ type: 'text' as const, text: '<system-reminder>catalog</system-reminder>' }] }
session.append('request/header', { header: { config: { provider: 'mock', model: 'm' }, messagePrefix: [prefix] }, reason: 'change' })
expect(() => { dispatch(ctx, Object.freeze({ model: 'm', messages: Object.freeze([prefix, ...boundary]), sessionId: session.id })) })
expect(() => { dispatch(ctx, loopRequest({ model: 'm', messages: Object.freeze([prefix, ...boundary]), sessionId: session.id })) })
.not.toThrow()
expect(() => { dispatch(ctx, Object.freeze({ model: 'm', messages: Object.freeze([...boundary]), sessionId: session.id })) })
expect(() => { dispatch(ctx, loopRequest({ model: 'm', messages: Object.freeze([...boundary]), sessionId: session.id })) })
.toThrow(/diverges from the boundary derivation/)
expect(() => { dispatch(ctx, Object.freeze({ model: 'm', messages: Object.freeze([...boundary, prefix]), sessionId: session.id })) })
expect(() => { dispatch(ctx, loopRequest({ model: 'm', messages: Object.freeze([...boundary, prefix]), sessionId: session.id })) })
.toThrow(/diverges from the boundary derivation/)
})
it('rejects message and header divergence', async () => {
const { ctx, session, boundary } = await requestSetup()
const divergent = [...boundary, { role: 'user', content: [{ type: 'text', text: 'phantom' }] }]
expect(() => { dispatch(ctx, Object.freeze({ model: 'm', messages: Object.freeze(divergent), sessionId: session.id })) })
expect(() => { dispatch(ctx, loopRequest({ model: 'm', messages: Object.freeze(divergent), sessionId: session.id })) })
.toThrow(/diverges from the boundary derivation/)
expect(() => { dispatch(ctx, Object.freeze({ model: 'other', messages: Object.freeze(boundary), sessionId: session.id })) })
expect(() => { dispatch(ctx, loopRequest({ model: 'other', messages: Object.freeze(boundary), sessionId: session.id })) })
.toThrow(/diverges from the folded request header/)
})
@@ -66,7 +71,7 @@ describe('request-reconstruction invariant', () => {
const ctx = await setup()
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 })
const bare = loopRequest({ 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/)
@@ -74,12 +79,34 @@ describe('request-reconstruction invariant', () => {
it('rejects an unfrozen messages array but skips requests outside the loop contract', async () => {
const { ctx, session, boundary } = await requestSetup()
expect(() => { dispatch(ctx, Object.freeze({ model: 'm', messages: [...boundary], sessionId: session.id })) })
expect(() => { dispatch(ctx, loopRequest({ model: 'm', messages: [...boundary], sessionId: session.id })) })
.toThrow(/frozen messages array/)
expect(() => { dispatch(ctx, { model: 'summarizer', messages: [], sessionId: session.id }) }).not.toThrow()
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()
const directSession = ctx.sessions.create(SessionId('direct-one-shot'))
expect(() => {
dispatch(ctx, Object.freeze({ model: 'one-shot', messages: Object.freeze([]), sessionId: directSession.id }))
}).not.toThrow()
})
it('rejects malformed requests carrying the loop marker', async () => {
const { ctx, session } = await requestSetup()
expect(() => {
dispatch(ctx, markLoopRequest({ model: 'm', messages: Object.freeze([]), sessionId: session.id }))
}).toThrow(/request must be frozen/)
expect(() => {
dispatch(ctx, loopRequest({ model: 'm', messages: Object.freeze([]) }))
}).toThrow(/carry a session id/)
expect(() => {
dispatch(ctx, loopRequest({
model: 'm',
messages: Object.freeze([]),
sessionId: SessionId('missing-loop-session'),
}))
}).toThrow(/live session id/)
})
it('prepends ahead of a short-circuiting stream listener', async () => {
@@ -93,7 +120,7 @@ describe('request-reconstruction invariant', () => {
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: { provider: 'mock', model: 'm' } }, reason: 'initial' })
const divergent = Object.freeze({
const divergent = loopRequest({
model: 'm',
messages: Object.freeze([{ role: 'user', content: [{ type: 'text', text: 'phantom' }] }]),
sessionId: session.id,